Feat: Arrangement Backfill

[kwannoel]

https://www.notion.so/risingwave-labs/Arrangement-Backfill-31d4d0044726451c9cc77dc84a0f8bb2

Main tasks:

• Implement temporary storage.
• Implement Arrangement backfill executor.
• Implement frontend StreamTableScan -> Arrangement Backfill executor. feat(stream): support arrangement backfill frontend #13622
• Add dispatcher logic for scale in / scale out of backfill.
• Test arrangement backfill with scale in / scale out.
• Replication feat(storage): fill missing columns with null for replicated state_table #13621
• Make sure at least 1 row read per barrier, so we don't lose progress for backfill with large tombstone sizes. Be sure to avoid deadlock, if stream is paused, don't try to read. feat(stream): read exactly 1 row if no snapshot read per barrier for arrangement backfill #14842
• Support rate limit in arrangement backfill.
• Support recoverable row count estimate in arrangement backfill. feat(stream): support row count for arrangement backfill #14836
• Test arrangement backfill performance. perf(stream): optimize arrangement backfill snapshot read #14617
• Test arrangement backfill correctness with Better testing for backfill executor #11461.
• Support pause on next bootstrap feat(stream): Support pause_on_next_bootstrap for snapshot read #14745
• Compare the performance of arrangement backfill for q0, q4, q7, q8, q16. These tests have no backfill.
• Debug why e2e test timeout.
• Debug why barrier cancel test timeout.

Lower priority:

• Can optimize arrangement backfill. Only need to replicate output columns + PK, instead of entire table.
  • Requires more work though. When reading checkpointed data, we need to project output rows.
  • Otherwise the schema will be inconsistent when merging replicated and checkpointed data.
• Can optimize arrangement backfill: No need replicate if on the same CN as upstream. This logic needs to take place in meta though. It will need to rewrite the state table after rescheduling.
• Support spill anytime in arrangement backfill.

Misc

• Make sure replicated state table cannot use the other constructors used by normal state table.
• Make sure replicated state table can only use getters / setters corresponding to it.
• refactor stream scan proto, decouple variant specific fields #13871
• Parallelize backfill: Make backfill vnode level within each partition #14905

[github-actions]

This issue has been open for 60 days with no activity.

If you think it is still relevant today, and needs to be done in the near future, you can comment to update the status, or just manually remove the no-issue-activity label.

You can also confidently close this issue as not planned to keep our backlog clean.
Don't worry if you think the issue is still valuable to continue in the future.
It's searchable and can be reopened when it's time. 😄