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

src/frontend/src/observer/observer_manager.rs

@@ -245,6 +245,7 @@ impl FrontendObserverNode {
         let Some(info) = resp.info.as_ref() else {
             return;
         };
+        tracing::trace!(op = ?resp.operation(), ?info, "handle catalog notification");
 
         let mut catalog_guard = self.catalog.write();
         match info {

src/meta/src/controller/catalog.rs

@@ -66,10 +66,11 @@ use tracing::info;
 use super::utils::{check_subscription_name_duplicate, get_fragment_ids_by_jobs};
 use crate::controller::rename::{alter_relation_rename, alter_relation_rename_refs};
 use crate::controller::utils::{
-    build_relation_group, check_connection_name_duplicate, check_database_name_duplicate,
-    check_function_signature_duplicate, check_relation_name_duplicate, check_schema_name_duplicate,
-    check_secret_name_duplicate, ensure_object_id, ensure_object_not_refer, ensure_schema_empty,
-    ensure_user_id, extract_external_table_name_from_definition, get_referring_objects,
+    build_relation_group_for_delete, check_connection_name_duplicate,
+    check_database_name_duplicate, check_function_signature_duplicate,
+    check_relation_name_duplicate, check_schema_name_duplicate, check_secret_name_duplicate,
+    ensure_object_id, ensure_object_not_refer, ensure_schema_empty, ensure_user_id,
+    extract_external_table_name_from_definition, get_referring_objects,
     get_referring_objects_cascade, get_user_privilege, list_user_info_by_ids,
     resolve_source_register_info_for_jobs, PartialObject,
 };
@@ -891,7 +892,7 @@ impl CatalogController {
 
         txn.commit().await?;
 
-        let relation_group = build_relation_group(
+        let relation_group = build_relation_group_for_delete(
             dirty_mview_objs
                 .into_iter()
                 .chain(dirty_mview_internal_table_objs.into_iter())
@@ -2305,7 +2306,7 @@ impl CatalogController {
 
         // notify about them.
         self.notify_users_update(user_infos).await;
-        let relation_group = build_relation_group(to_drop_objects);
+        let relation_group = build_relation_group_for_delete(to_drop_objects);
 
         let version = self
             .notify_frontend(NotificationOperation::Delete, relation_group)

src/meta/src/controller/streaming_job.rs

@@ -63,7 +63,7 @@ use crate::barrier::{ReplaceTablePlan, Reschedule};
 use crate::controller::catalog::CatalogController;
 use crate::controller::rename::ReplaceTableExprRewriter;
 use crate::controller::utils::{
-    build_relation_group, check_relation_name_duplicate, check_sink_into_table_cycle,
+    build_relation_group_for_delete, check_relation_name_duplicate, check_sink_into_table_cycle,
     ensure_object_id, ensure_user_id, get_fragment_actor_ids, get_fragment_mappings,
     rebuild_fragment_mapping_from_actors, PartialObject,
 };
@@ -99,6 +99,11 @@ impl CatalogController {
         Ok(obj.oid)
     }
 
+    /// Create catalogs for the streaming job, then notify frontend about them if the job is a
+    /// materialized view.
+    ///
+    /// Some of the fields in the given streaming job are placeholders, which will
+    /// be updated later in `prepare_streaming_job` and notify again in `finish_streaming_job`.
     pub async fn create_job_catalog(
         &self,
         streaming_job: &mut StreamingJob,
@@ -333,16 +338,23 @@ impl CatalogController {
         Ok(())
     }
 
+    /// Create catalogs for internal tables, then notify frontend about them if the job is a
+    /// materialized view.
+    ///
+    /// Some of the fields in the given "incomplete" internal tables are placeholders, which will
+    /// be updated later in `prepare_streaming_job` and notify again in `finish_streaming_job`.
+    ///
+    /// Returns a mapping from the temporary table id to the actual global table id.
     pub async fn create_internal_table_catalog(
         &self,
         job: &StreamingJob,
-        mut internal_tables: Vec<PbTable>,
+        mut incomplete_internal_tables: Vec<PbTable>,
     ) -> MetaResult<HashMap<u32, u32>> {
         let job_id = job.id() as ObjectId;
         let inner = self.inner.write().await;
         let txn = inner.db.begin().await?;
         let mut table_id_map = HashMap::new();
-        for table in &mut internal_tables {
+        for table in &mut incomplete_internal_tables {
             let table_id = Self::create_object(
                 &txn,
                 ObjectType::Table,
@@ -354,10 +366,13 @@ impl CatalogController {
             .oid;
             table_id_map.insert(table.id, table_id as u32);
             table.id = table_id as _;
-            let mut table_model: table::ActiveModel = table.clone().into();
-            table_model.table_id = Set(table_id as _);
-            table_model.belongs_to_job_id = Set(Some(job_id));
-            table_model.fragment_id = NotSet;
+
+            let table_model = table::ActiveModel {
+                table_id: Set(table_id as _),
+                belongs_to_job_id: Set(Some(job_id)),
+                fragment_id: NotSet,
+                ..table.clone().into()
+            };
             Table::insert(table_model).exec(&txn).await?;
         }
         txn.commit().await?;
@@ -366,7 +381,7 @@ impl CatalogController {
             self.notify_frontend(
                 Operation::Add,
                 Info::RelationGroup(RelationGroup {
-                    relations: internal_tables
+                    relations: incomplete_internal_tables
                         .iter()
                         .map(|table| Relation {
                             relation_info: Some(RelationInfo::Table(table.clone())),
@@ -565,7 +580,9 @@ impl CatalogController {
         txn.commit().await?;
 
         if !objs.is_empty() {
-            self.notify_frontend(Operation::Delete, build_relation_group(objs))
+            // We also have notified the frontend about these objects,
+            // so we need to notify the frontend to delete them here.
+            self.notify_frontend(Operation::Delete, build_relation_group_for_delete(objs))
                 .await;
         }
         Ok(true)

src/meta/src/controller/utils.rs

@@ -1082,7 +1082,13 @@ where
     ))
 }
 
-pub(crate) fn build_relation_group(relation_objects: Vec<PartialObject>) -> NotificationInfo {
+/// Build a relation group for notifying the deletion of the given objects.
+///
+/// Note that only id fields are filled in the relation info, as the arguments are partial objects.
+/// As a result, the returned notification info should only be used for deletion.
+pub(crate) fn build_relation_group_for_delete(
+    relation_objects: Vec<PartialObject>,
+) -> NotificationInfo {
     let mut relations = vec![];
     for obj in relation_objects {
         match obj.obj_type {

src/meta/src/model/stream.rs

@@ -18,8 +18,10 @@ use std::ops::AddAssign;
 use itertools::Itertools;
 use risingwave_common::catalog::TableId;
 use risingwave_common::hash::{VirtualNode, WorkerSlotId};
+use risingwave_common::util::stream_graph_visitor;
 use risingwave_connector::source::SplitImpl;
 use risingwave_meta_model::{SourceId, WorkerId};
+use risingwave_pb::catalog::Table;
 use risingwave_pb::common::PbActorLocation;
 use risingwave_pb::meta::table_fragments::actor_status::ActorState;
 use risingwave_pb::meta::table_fragments::{ActorStatus, Fragment, State};
@@ -564,6 +566,26 @@ impl TableFragments {
         }
     }
 
+    /// Retrieve the **complete** internal tables map of the whole graph.
+    ///
+    /// Compared to [`crate::stream::StreamFragmentGraph::incomplete_internal_tables`],
+    /// the table catalogs returned here are complete, with all fields filled.
+    pub fn internal_tables(&self) -> BTreeMap<u32, Table> {
+        let mut tables = BTreeMap::new();
+        for fragment in self.fragments.values() {
+            stream_graph_visitor::visit_stream_node_internal_tables(
+                &mut fragment.actors[0].nodes.clone().unwrap(),
+                |table, _| {
+                    let table_id = table.id;
+                    tables
+                        .try_insert(table_id, table.clone())
+                        .unwrap_or_else(|_| panic!("duplicated table id `{}`", table_id));
+                },
+            );
+        }
+        tables
+    }
+
     /// Returns the internal table ids without the mview table.
     pub fn internal_table_ids(&self) -> Vec<u32> {
         self.fragments

src/meta/src/rpc/ddl_controller.rs

@@ -1040,11 +1040,14 @@ impl DdlController {
         streaming_job.set_dml_fragment_id(fragment_graph.dml_fragment_id());
 
         // create internal table catalogs and refill table id.
-        let internal_tables = fragment_graph.internal_tables().into_values().collect_vec();
+        let incomplete_internal_tables = fragment_graph
+            .incomplete_internal_tables()
+            .into_values()
+            .collect_vec();
         let table_id_map = self
             .metadata_manager
             .catalog_controller
-            .create_internal_table_catalog(&streaming_job, internal_tables)
+            .create_internal_table_catalog(&streaming_job, incomplete_internal_tables)
             .await?;
         fragment_graph.refill_internal_table_ids(table_id_map);
 
@@ -1560,7 +1563,6 @@ impl DdlController {
     ) -> MetaResult<(CreateStreamingJobContext, TableFragments)> {
         let id = stream_job.id();
         let specified_parallelism = fragment_graph.specified_parallelism();
-        let internal_tables = fragment_graph.internal_tables();
         let expr_context = stream_ctx.to_expr_context();
         let max_parallelism = NonZeroUsize::new(fragment_graph.max_parallelism()).unwrap();
 
@@ -1643,6 +1645,7 @@ impl DdlController {
             table_parallelism,
             max_parallelism.get(),
         );
+        let internal_tables = table_fragments.internal_tables();
 
         if let Some(mview_fragment) = table_fragments.mview_fragment() {
             stream_job.set_table_vnode_count(mview_fragment.vnode_count());

src/meta/src/stream/stream_graph/fragment.rs

@@ -349,6 +349,9 @@ impl StreamFragmentGraph {
     ) -> MetaResult<Self> {
         let fragment_id_gen =
             GlobalFragmentIdGen::new(env.id_gen_manager(), proto.fragments.len() as u64);
+        // Note: in SQL backend, the ids generated here are fake and will be overwritten again
+        // with `refill_internal_table_ids` later.
+        // TODO: refactor the code to remove this step.
         let table_id_gen = GlobalTableIdGen::new(env.id_gen_manager(), proto.table_ids_cnt as u64);
 
         // Create nodes.
@@ -417,8 +420,14 @@ impl StreamFragmentGraph {
         })
     }
 
-    /// Retrieve the internal tables map of the whole graph.
-    pub fn internal_tables(&self) -> BTreeMap<u32, Table> {
+    /// Retrieve the **incomplete** internal tables map of the whole graph.
+    ///
+    /// Note that some fields in the table catalogs are not filled during the current phase, e.g.,
+    /// `fragment_id`, `vnode_count`. They will be all filled after a `TableFragments` is built.
+    /// Be careful when using the returned values.
+    ///
+    /// See also [`crate::model::TableFragments::internal_tables`].
+    pub fn incomplete_internal_tables(&self) -> BTreeMap<u32, Table> {
         let mut tables = BTreeMap::new();
         for fragment in self.fragments.values() {
             for table in fragment.extract_internal_tables() {
@@ -431,6 +440,8 @@ impl StreamFragmentGraph {
         tables
     }
 
+    /// Refill the internal tables' `table_id`s according to the given map, typically obtained from
+    /// `create_internal_table_catalog`.
     pub fn refill_internal_table_ids(&mut self, table_id_map: HashMap<u32, u32>) {
         for fragment in self.fragments.values_mut() {
             stream_graph_visitor::visit_internal_tables(

src/meta/src/stream/test_fragmenter.rs

@@ -448,7 +448,7 @@ async fn test_graph_builder() -> MetaResult<()> {
         time_zone: graph.ctx.as_ref().unwrap().timezone.clone(),
     };
     let fragment_graph = StreamFragmentGraph::new(&env, graph, &job)?;
-    let internal_tables = fragment_graph.internal_tables();
+    let internal_tables = fragment_graph.incomplete_internal_tables();
 
     let actor_graph_builder = ActorGraphBuilder::new(
         job.id(),