refactor(meta): clarify the completeness of internal table catalogs

[BugenZhao]

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

What's changed and what's your intention?

When creating a streaming job, the internal tables directly scraped from the frontend's StreamFragmentGraph are incomplete, because several fields like fragment_id or vnode_count are not filled yet and will be determined later.

Previously, we notified the frontend nodes with these catalogs but did not update them afterward. This will cause problems since variable vnode count support is introduced, as the vnode count of a table is a significant property when scheduling a batch scan over it.

This PR changes to notify the internal table catalogs only after a TableFragments is build, where all information is final and complete.

This PR revises the documentation, comments, and naming of related snippets to clarify the completeness of internal table catalogs during various phases of creating a streaming job.

Checklist

• I have written necessary rustdoc comments
• I have added necessary unit tests and integration tests
• All checks passed in ./risedev check (or alias, ./risedev c)

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.

[BugenZhao]

• fix(meta): align job's max parallelism while replacing table #19052
• refactor: use 1 for vnode count of singletons #18753
• refactor: distinguish between placeholder and compat for vnode count #18976
• refactor(meta): clarify the completeness of internal table catalogs #18944 👈
• main

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

Join @BugenZhao and the rest of your teammates on Graphite

[BugenZhao]

As the notification now goes after successfully scheduling and building of the job, a failure here does not always mean that the notification was sent. That's why we changed all DROP handlers in the frontend to be DROP .. IF EXISTS.

An alternative could be introducing a new operation like Cleanup for this purposes, ensuring Delete remains strictly applied. But it seems too ad-hoc to me as there's no much other usage of it.

[kwannoel]

Can there be a race condition? Whereby we have Delete of some catalog followed by Add of that catalog?

[BugenZhao]

I think not because try_abort_creating_streaming_job is called only after create_streaming_job_inner has returned with error, where Add is either pushed into the notification queue or not issued at all. After entering this try_abort function, no further Add operations will occur.

[yezizp2012]

🤔 We ensured that CREATE and DROP will be properly paired in #18476 .Even if the CREATE notification fails to send due to either a FE or meta node reboot or a broken connection between them, the creation of the mview and its internal table catalogs will still be synchronized through an initial snapshot.

[BugenZhao]

Oops, I realized that we'll eventually notify the frontend nodes to complete the catalogs for both the job and the internal tables in finish_streaming_job.

risingwave/src/meta/src/controller/streaming_job.rs

Lines 771 to 797 in 265a7ac

	// notify frontend: job, internal tables.
	let internal_table_objs = Table::find()
	.find_also_related(Object)
	.filter(table::Column::BelongsToJobId.eq(job_id))
	.all(&txn)
	.await?;
	let mut relations = internal_table_objs
	.iter()
	.map(|(table, obj)| PbRelation {
	relation_info: Some(PbRelationInfo::Table(
	ObjectModel(table.clone(), obj.clone().unwrap()).into(),
	)),
	})
	.collect_vec();
	let mut notification_op = NotificationOperation::Add;
	match job_type {
	ObjectType::Table => {
	let (table, obj) = Table::find_by_id(job_id)
	.find_also_related(Object)
	.one(&txn)
	.await?
	.ok_or_else(|| MetaError::catalog_id_not_found("table", job_id))?;
	if table.table_type == TableType::MaterializedView {
	notification_op = NotificationOperation::Update;
	}

So this PR is more like a refactor rather than a fix. Actually the main motivation of this PR is to enable #18976, where we don't expect the frontend nodes to receive a VnodeCount::Placeholder to avoid potential confusion.

However, I think it's also okay to bypass the check and allow VnodeCount::Placeholder if the table is in StreamJobStatus::Creating. This is because informing the frontends about these tables is primarily for reference in DROP statements, where querying them is not permitted. As a result, refactoring in this PR may not be necessary anymore.

[BugenZhao]

As a result, refactoring in this PR may not be necessary anymore.

Have updated the PR to only

revises the documentation, comments, and naming of related snippets to clarify the completeness of internal table catalogs during various phases of creating a streaming job.

[BugenZhao]

The internal_tables field in CreateStreamingJobContext will now be complete.