feat: track progress for SourceBackfill (blocking DDL)

[xxchan]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Support blocking DDL for source #18338

Core changes:

1. Add SourceScan to tracking_progress_actor_ids of a MV. So a MV will not enter Created state until the SourceBackfillExecutor's actor reported progress.
2. In SourceBackfillExecutor, the progress reporting is a dummy one: it simply reused MV's mechanism, and only uses finish.

---

A subtle case to consider: whether to report finish after entering SourceCatchingUp state, or after Finished state?

• Finished might be more intuitive and safe. However, it may make a foreground DDL blocked forever until upstream messages come. (See state diagram in Let source backfill finish when there's no data from upstream #18299. We will be in SourceCatchingUp.) It may seem strange: why with refactor: use high watermark to finish backfill faster #18342 (end at high watermark), we still cannot enter Finished state? Or why no messages come from upstream when we already backfilled to high watermark?

  Here's a illustration: Assume there's no messages from upstream, and we reached high_watermark. Now we cannot distinguish whether upstream has reached high_watermark, so we enter SourceCatchingUp state. If it hasn't caught up, we can enter Finished when it catches up. If it's already at latest, we will not be able to transition state. (Let source backfill finish when there's no data from upstream #18299's point 4 mentioned that we might distinguish this situation if we can do has_message_available)
  

• So it seems we'd better end at SourceCatchingUp to avoid blocking forever. It's also reasonable, since now we already exited backfill stream. But the problem is that after reporting finish, and the DDL returns, online scaling may happen now. But we need to handle it carefully (Problem of online scaling for source backfill #18300).

  Note: for scaling, "backfill finished" == "progress reported finish". Not exactly the same as the loop code in the executor. So we can't use the old hack like scanning whole state table to delay finished. Now we do want to finish earlier! But need to make it scaling-safe.

  So the solution is to decouple 2 timings:

  • T1: "when scale is disallowed" (also the time when DDL returns)
  • T2: "when it's safe to scale" (when we scan the state table, and exit loop. This PR used try_wait_epoch to fix the problem described in Problem of online scaling for source backfill #18300)

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• I have added test labels as necessary. See details.
• I have added fuzzing tests or opened an issue to track them. (Optional, recommended for new SQL features Sqlsmith: Sql feature generation #7934).
• My PR contains breaking changes. (If it deprecates some features, please create a tracking issue to remove them in the future).
• All checks passed in ./risedev check (or alias, ./risedev c)
• My PR changes performance-critical code. (Please run macro/micro-benchmarks and show the results.)

• My PR contains critical fixes that are necessary to be merged into the latest release. (Please check out the details)

Documentation

• My PR needs documentation updates. (Please use the Release note section below to summarize the impact on users)

Release note

If this PR includes changes that directly affect users or other significant modifications relevant to the community, kindly draft a release note to provide a concise summary of these changes. Please prioritize highlighting the impact these changes will have on users.

[xxchan]

• feat: track progress for SourceBackfill (blocking DDL) #18112 : 2 dependent PRs (#16825 , #18355 ) 👈
• refactor: use high watermark to finish backfill faster #18342 : 1 other dependent PR (#18368 )
• refactor: add target_offsets to determinine if source backfill finished #18297
• refactor: simplify SourceBackfill's memory state #18296
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @xxchan and the rest of your teammates on Graphite

[xxchan]

This is core change 1

[xxchan]

This is a minor bug. (not related

[xxchan]

Remove the early exit condition, let it only exit in one place to make it safer & simpler

[fuyufjh]

Generally LGTM

[fuyufjh]

The 2nd argument of progress.finish() is current_consumed_rows, which I think it's easy to count and report an accurate number. Why not doing it?

[xxchan]

You are right. Not did it just because no total_consumed_rows and thought it meaningless

[fuyufjh]

Following the #18112 (comment) will make it more meaningful. I think we can do the counting part in this PR.

[fuyufjh]

An idea: We may report the process and total rows in the progress as well.

current: (show jobs)

  Id  |                     Statement                      | Progress
------+----------------------------------------------------+----------
 1010 | CREATE MATERIALIZED VIEW mv1 AS SELECT * FROM mv1  | 2.21%

expected:

  Id  |                     Statement                         | Processed | Total | Progress
------+-------------------------------------------------------+-----------+-------+---------
 1010 | CREATE MATERIALIZED VIEW mv1 AS SELECT * FROM mv1     | 221       | 1000  | 2.21% 
 1012 | CREATE MATERIALIZED VIEW mv2 AS SELECT * FROM source1 | 114514    |       | 

[xxchan]

I found #17735 already did similar thing when set streaming_use_snapshot_backfill = true. It just changed the Progress string.

dev=> show jobs;
┌────┬────────────────────────────────────────────────┬───────────────────┐
│ Id │                   Statement                    │     Progress      │
├────┼────────────────────────────────────────────────┼───────────────────┤
│ 29 │ CREATE MATERIALIZED VIEW mv AS SELECT * FROM t │ Snapshot [32.08%] │
└────┴────────────────────────────────────────────────┴───────────────────┘
(1 row)

risingwave/src/meta/src/barrier/creating_job/mod.rs

Lines 173 to 227 in 3133a72

	pub(super) fn gen_ddl_progress(&self) -> DdlProgress {
	let progress = match &self.status {
	CreatingStreamingJobStatus::ConsumingSnapshot {
	create_mview_tracker,
	..
	} => {
	if create_mview_tracker.has_pending_finished_jobs() {
	"Snapshot finished".to_string()
	} else {
	let progress = create_mview_tracker
	.gen_ddl_progress()
	.remove(&self.info.table_fragments.table_id().table_id)
	.expect("should exist");
	format!("Snapshot [{}]", progress.progress)
	}
	}
	CreatingStreamingJobStatus::ConsumingLogStore {
	start_consume_log_store_epoch,
	..
	} => {
	let max_collected_epoch = max(
	self.barrier_control.max_collected_epoch().unwrap_or(0),
	self.backfill_epoch,
	);
	let lag = Duration::from_millis(
	Epoch(*start_consume_log_store_epoch)
	.physical_time()
	.saturating_sub(Epoch(max_collected_epoch).physical_time()),
	);
	format!(
	"LogStore [remain lag: {:?}, epoch cnt: {}]",
	lag,
	self.barrier_control.inflight_barrier_count()
	)
	}
	CreatingStreamingJobStatus::ConsumingUpstream { .. } => {
	format!(
	"Upstream [unattached: {}, epoch cnt: {}]",
	self.barrier_control.unattached_epochs().count(),
	self.barrier_control.inflight_barrier_count(),
	)
	}
	CreatingStreamingJobStatus::Finishing { .. } => {
	format!(
	"Finishing [epoch count: {}]",
	self.barrier_control.inflight_barrier_count()
	)
	}
	};
	DdlProgress {
	id: self.info.table_fragments.table_id().table_id as u64,
	statement: self.info.definition.clone(),
	progress,
	}
	}

[fuyufjh]

Looks good!

[xxchan]

Let me do this in a separate PR, because it's more complex than I expected. Since one MV may have multiple backfill upstreams, and I want to distinguish MV backfill rows and source backfill rows.