Tracking: support partial checkpoint

[wenym1]

Original rfc: risingwavelabs/rfcs#84

This rfc involves changes in 3 layers: batch query, streaming job and hummock storage. We are going to implement the whole rfc from bottom layer to upper layer step by step. Since the current global checkpoint is a special case of partial checkpoint, though the some features of partial checkpoint will have been implemented in bottom layer, we can keep the same global checkpoint logic in upper layer before we have implemented logic in upper layer.

After the partial checkpoint is supported, we can then work on partial recovery to isolate the failure from different MVs.

Following are some changes in each layer

hummock Storage

First we will refactor the current code to implement part of the features required by partial checkpoint while remaining the same as current logic. This includes the following:

• change the hummock version metadata to maintain per table max committed epoch and safe epoch
• change the process of collecting barrier from two separate one-shot rpc calls to a long existing streaming rpc between CN and meta to report barrier progress.

Meanwhile, we can develop L0 as a log so that we can reuse the data of MV.

streaming job

First we will implement new streaming executor similar to a source executor that consume the logs of upstream MV.

Second we will implement a partial checkpoint manager that comprehend the streaming graph and collect the barriers reported from each MV parallelism and trigger partial checkpoint.

batch query

By now the batch query will use the global max committed epoch as the query epoch. We can implement different query consistency for batch query. Different query consistency means the policy to choose the query epoch of different state table. The default one is to use the global max committed epoch.

partial recovery

First we can support the failure isolation among MVs that are not connected.

Second we can support failure isolation between upstream and downstream MVs .

Development progress:

Ongoing:

• batch query
  • support different query consistency
    • Tracking: deprecate safe epoch and generalize time travel query #18214
    • feat(storage): notify frontend with more hummock version info #18570
    • feat(frontend): frontend wait hummock version id for ddl #18613
    • feat(frontend): generate query epoch by committed epoch of involved tables #18592
    • feat(storage): per table try wait epoch #18622
  • isolate checkpoint between different databases
• support scale in a single barrier
  • Support configuration change in a single barrier #18312
• spawn actor in inject barrier
  • fix: collect actor sender in actor creation #15776
  • feat(storage): register and unregister member table id at commit epoch #15852
  • feat: remote input subscribe on barrier mutation #17612
  • refactor: actor wait barrier manager inject barrier #17613
  • refactor(meta): simplify InflightActorInfo and extract InflightGraphInfo #17956
  • refactor: register actor barrier sender asynchronously #18104
  • feat(snapshot-backfill): only receive mutation from barrier worker for snapshot backfill #18210
  • refactor(meta): build actors in inject barrier #18270
  • refactor: remove drop_actors rpc and simplify barrier worker state #18354
  • refactor: unify to subscribe mutation via barrier sender #18255
  • refactor: maintain subscriptions in local barrier worker #18516

All tasks:

• hummock Storage
  • Support L0 as a log
    • Upload old value in separate SST
      • feat(storage): support enabling consistent op on the flight #14681
      • feat(storage): store old value in memtable #15301
      • feat(meta): use version object ids view to clean stale objects and deprecate gc_object_ids #16309
      • feat(meta): collect non-checkpoint epoch for checkpoint barrier commit #16400
      • feat(meta): support inject barrier to enable and disable writing old value for subscription #16460
      • refactor(state-table): make consistent op an enum in state table #16471
      • feat(storage): add metrics to monitor old value size #16559
      • fix(storage): include change log sst in object size map #16561
      • fix(storage): fix unexpected dropping change log on compaction version delta #16613
      • refactor(meta): deprecate branched sst #16698
    • Support scan the change log between two epochs
      • feat(storage): define iter_log method and in-mem trivial implementation #15709
      • refactor(storage): extract logic of key bounded iterator #16027
      • feat(storage): implement iter_log for hummock #15666
      • feat(storage): add metrics for iter_log #16658
      • refactor(storage-table): ensure UpdataInsert emit next to UpdateDelete when reading change log #17770
      • fix(storage): generate correct user key range in iter_log #17805
      • feat(storage): collect and report iter log local metrics #17862
  • sync epoch in per-table manner
    • Hummock version maintains per table max committed epoch and safe epoch
      • refactor(storage): replace PbHummockVersion with new HummockVersion struct #14101
      • feat(storage): support different snapshot for streaming jobs #15896
      • feat(storage): store compaction group id in per table info #17125
      • refactor(storage): limit use of hummock version safe epoch #17161
      • fix(storage): use per-table committed epoch in read version update #17744
      • feat: call clear shared buffer with latest version id #17972
      • refactor(storage): hide usage on global max committed epoch #17973
      • feat(meta): support commit epoch on subset of tables #17997
    • per-table sync
      • feat(storage): avoid uploader depending on seal_epoch #16985
      • refactor(storage): trigger uploader sync on SyncEpoch event #17113
      • refactor(meta): maintain fragment actor and table info in barrier manager #17192
      • refactor(storage): pass table ids to state store sync #17332
      • refactor(storage): emit sst on single upload task succeeds and remove sync finish event #17333
      • feat(storage): decouple upload task finish order from epoch #17460
      • refactor(storage): remove global epoch in uploader #17500
      • feat(storage): decouple spill task from epoch #17539
      • feat(storage): pass epoch and table id before barrier #17635
• meta node refactor
  • Send and collect barrier in streaming rpc
    • feat(meta): inject and collect barrier in bidi stream #14887
      • local barrier manager
        • refactor: implement local barrier manager as event handler #14342
        • feat: handle state store sync in local barrier manager #14377
        • refactor: own global barrier manager in worker loop #14410
        • feat: poll barrier complete rpc in barrier manager worker loop #14419
        • refactor: local barrier manager use prev epoch to track barrier state #14436
        • refactor: decouple build actor and spawn actors #14641
        • feat: handle streaming control rpc in worker loop #14737
        • feat(storage): reset to the latest committed epoch in recovery #14923
        • refactor(meta): handle all stream client request in stream rpc manager #14844
        • refactor(stream): own and recreate shared context and local barrier manager in recovery #15157
      • global barrier manager
        • refactor(meta): merge periodic and scheduled barrier and non-async handle barrier #14977
        • feat(meta): maintain snapshot of active streaming compute node in worker loop #15033
        • feat(meta): make complete barrier non-async in worker loop #15109
        • refactor(meta): track finished create mv job in tracker #15112
    • spawn actor in inject barrier
      • fix: collect actor sender in actor creation #15776
      • feat(storage): register and unregister member table id at commit epoch #15852
      • feat: remote input subscribe on barrier mutation #17612
      • refactor: actor wait barrier manager inject barrier #17613
      • refactor(meta): simplify InflightActorInfo and extract InflightGraphInfo #17956
      • refactor: register actor barrier sender asynchronously #18104
      • feat(snapshot-backfill): only receive mutation from barrier worker for snapshot backfill #18210
      • refactor(meta): build actors in inject barrier #18270
      • refactor: remove drop_actors rpc and simplify barrier worker state #18354
      • refactor: unify to subscribe mutation via barrier sender #18255
      • refactor: maintain subscriptions in local barrier worker #18516
  • support scale in a single barrier
    • Support configuration change in a single barrier #18312
  • isolate checkpoint between different databases
• streaming job
  • new streaming executor that consume logs of upstream MV
    • snapshot backfill
      • feat: support partial checkpoint based backfill (a.k.a snapshot backfill) #17735
        • refactor(meta): commit finish streaming job in barrier manager #17674
        • refactor(meta): only store CreateStreamingJob command in tracker #17742
        • feat: support inject and collect barrier from partial graph #17758
        • refactor(meta): let global barrier manager own create mview tracker #17839
        • refactor: add subscription via add mutation #17897
        • feat: enable shuffle for snapshot backfill #18063
  • partial checkpoint manager
• batch query
  • support different query consistency
    • Tracking: deprecate safe epoch and generalize time travel query #18214
    • feat(storage): notify frontend with more hummock version info #18570
• partial recovery

[kwannoel]

batch query

By now the batch query will use the global max committed epoch as the query epoch. We can implement different query consistency for batch query. Different query consistency means the policy to choose the query epoch of different state table. The default one is to use the global max committed epoch.

For batch query can we use global checkpoint of created stream jobs i.e. excluding those stream jobs which are in creating process. Such that stream jobs being created will not affect freshness of batch query.