feat: improve progress msg in SHOW JOBS for source backfill

[xxchan]

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

What's changed and what's your intention?

Previous discussion here: #18112 (comment)

Previously, progress in SHOW JOBS is a percentile string. After we introduced different type of backfill, things become tricker:

• For some types, like SourceBackfill, we cannot even tell what's the end position, thus no percentile at all.
• We may have different type of backfills in one MV. How to display them together is a problem.
• Snapshot backfill introduced complex string as progress:

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

  Lines 130 to 175 in 05b68bc

  	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()
  	)
  	}

The idea of this PR is simple: just report separately, and abuse the string.
(Ideas to improve how the progress string look like is welcomed)

match (mv_progress, source_progress) {
    (Some(mv_progress), Some(source_progress)) => {
        format!(
            "MView Backfill: {}, Source Backfill: {}",
            mv_progress, source_progress
        )
    }
    (Some(mv_progress), None) => mv_progress,
    (None, Some(source_progress)) => source_progress,
    (None, None) => "Unknown".to_string(),
}

Example:

┌────┬──────────────────────────────────────────────────────────────────────────────────────┬────────────────────────────────────────────────────────────────┐
│ Id │                                      Statement                                       │                            Progress                            │
├────┼──────────────────────────────────────────────────────────────────────────────────────┼────────────────────────────────────────────────────────────────┤
│ 22 │ CREATE MATERIALIZED VIEW mv AS SELECT * FROM t UNION SELECT v1 FROM s_before_produce │ MView Backfill: 99.99% (3/3), Source Backfill: 0 rows consumed │
└────┴──────────────────────────────────────────────────────────────────────────────────────┴────────────────────────────────────────────────────────────────┘

Part of #18338

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: improve progress msg in SHOW JOBS for source backfill #18925 👈
• feat: persist processed rows for SourceBackfill for SHOW JOBS #18983
• 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]

Follow-up here #18983, where I'm not quite satisfied with the solution. 😕

[xxchan]

core logic here

[kwannoel]

Can we show progress via backfilled offset instead?

I assume offsets are monotonically increasing.

So something like:

average backfilled offset / upstream offset

[xxchan]

Can we show progress via backfilled offset instead?

Good point. 🤔

• If so, we don't need to persist num of rows any more feat: persist processed rows for SourceBackfill for SHOW JOBS #18983
• We will need a new code path in progress reporting. Cannot rely on the same code as MV backfill.
• This information is in state table (Can we just use state table for SHOW JOBS? And do not need to report progress any more 🤣 (still need report finish though.))
  • Maybe we can also add a column state tables in SHOW JOBS for users to check detailed progress.

[xxchan]

But there's still a problem: we have per-partition offset. (i.e., in state table we have detailed per-split progress.) How to aggregate it for users?

---

So something like:

average backfilled offset / upstream offset

Oh, I see your point here.

[xxchan]

Some points:

• We don't always have the ability to calculate "Lag size" (or average backfilled offset / upstream offset)

  Note that, the "offset" may mean very different things in different connectors, unlike Kafka offset is as simple as array index.

  e.g., Kinesis offset is not consecutive. looks like "SequenceNumber": "49639543389862895699365693304023657829928072185817071618". It can't be averaged, and we have no idea about the progress just from the sequence number.

  See also https://arc.net/l/quote/rdvyjwyz (Let source backfill finish when there's no data from upstream #18299)

  Therefore, num_backfilled_rows might still be a valuable thing.

• We actually have a metric on Grafana: "Kafka Consumer Lag Size".

  Not sure whether we can get similar things for other connectors like Kinesis.

• Most of the time, backfill should be finished very quickly and not many users will care about observability of the backfill progress. (And we do have some advanced tricks for advanced users as mentioned above.) So I'm thinking it may not worth the effort to implement it perfectly. 🤪 Of course we can gradually improve it later.

[kwannoel]

For kinesis offset, I think as long as long as it is monotonically increasing, we can still use its average.

[xxchan]

@kwannoel Do you think we can merge it in current form and refine later?