refactor(meta): commit finish catalog in barrier manager

ci/scripts/deterministic-recovery-test.sh

@@ -13,7 +13,9 @@ export RUST_LOG="risingwave_meta::barrier::recovery=debug,\
 risingwave_meta::manager::catalog=debug,\
 risingwave_meta::rpc::ddl_controller=debug,\
 risingwave_meta::barrier::mod=debug,\
-risingwave_simulation=debug"
+risingwave_simulation=debug,\
+risingwave_meta::stream::stream_manager=debug,\
+risingwave_meta::barrier::progress=debug"
 
 # Extra logs you can enable if the existing trace does not give enough info.
 #risingwave_stream::executor::backfill=trace,

src/meta/src/barrier/command.rs

@@ -24,7 +24,7 @@ use risingwave_common::types::Timestamptz;
 use risingwave_common::util::epoch::Epoch;
 use risingwave_connector::source::SplitImpl;
 use risingwave_hummock_sdk::HummockEpoch;
-use risingwave_pb::catalog::CreateType;
+use risingwave_pb::catalog::{CreateType, Table};
 use risingwave_pb::meta::table_fragments::PbActorStatus;
 use risingwave_pb::meta::PausedReason;
 use risingwave_pb::source::{ConnectorSplit, ConnectorSplits};
@@ -44,7 +44,7 @@ use tracing::warn;
 use super::info::{CommandActorChanges, CommandFragmentChanges, InflightActorInfo};
 use super::trace::TracedEpoch;
 use crate::barrier::GlobalBarrierManagerContext;
-use crate::manager::{DdlType, MetadataManager, WorkerId};
+use crate::manager::{DdlType, MetadataManager, StreamingJob, WorkerId};
 use crate::model::{ActorId, DispatcherId, FragmentId, TableFragments, TableParallelism};
 use crate::stream::{build_actor_connector_splits, SplitAssignment, ThrottleConfig};
 use crate::MetaResult;
@@ -97,6 +97,8 @@ pub struct ReplaceTablePlan {
     /// Note that there's no `SourceBackfillExecutor` involved for table with connector, so we don't need to worry about
     /// `backfill_splits`.
     pub init_split_assignment: SplitAssignment,
+    pub streaming_job: StreamingJob,
+    pub dummy_id: u32,
 }
 
 impl ReplaceTablePlan {
@@ -183,6 +185,8 @@ pub enum Command {
     /// for a while** until the `finish` channel is signaled, then the state of `TableFragments`
     /// will be set to `Created`.
     CreateStreamingJob {
+        streaming_job: StreamingJob,
+        internal_tables: Vec<Table>,
         table_fragments: TableFragments,
         /// Refer to the doc on [`MetadataManager::get_upstream_root_fragments`] for the meaning of "root".
         upstream_root_actors: HashMap<TableId, Vec<ActorId>>,
@@ -551,6 +555,7 @@ impl CommandContext {
                         merge_updates,
                         dispatchers,
                         init_split_assignment,
+                        ..
                     }) = replace_table
                     {
                         // TODO: support in v2.
@@ -1019,6 +1024,7 @@ impl CommandContext {
                             merge_updates,
                             dispatchers,
                             init_split_assignment,
+                            ..
                         }) = replace_table
                         {
                             self.clean_up(old_table_fragments.actor_ids()).await?;
@@ -1104,6 +1110,7 @@ impl CommandContext {
                 merge_updates,
                 dispatchers,
                 init_split_assignment,
+                ..
             }) => {
                 self.clean_up(old_table_fragments.actor_ids()).await?;
 

src/meta/src/barrier/mod.rs

@@ -62,7 +62,6 @@ use crate::manager::{
     ActiveStreamingWorkerChange, ActiveStreamingWorkerNodes, LocalNotification, MetaSrvEnv,
     MetadataManager, SystemParamsManagerImpl, WorkerId,
 };
-use crate::model::{ActorId, TableFragments};
 use crate::rpc::metrics::MetaMetrics;
 use crate::stream::{ScaleControllerRef, SourceManagerRef};
 use crate::{MetaError, MetaResult};
@@ -87,12 +86,6 @@ pub(crate) struct TableMap<T> {
     inner: HashMap<TableId, T>,
 }
 
-impl<T> TableMap<T> {
-    pub fn remove(&mut self, table_id: &TableId) -> Option<T> {
-        self.inner.remove(table_id)
-    }
-}
-
 impl<T> From<HashMap<TableId, T>> for TableMap<T> {
     fn from(inner: HashMap<TableId, T>) -> Self {
         Self { inner }
@@ -105,12 +98,6 @@ impl<T> From<TableMap<T>> for HashMap<TableId, T> {
     }
 }
 
-pub(crate) type TableActorMap = TableMap<HashSet<ActorId>>;
-pub(crate) type TableUpstreamMvCountMap = TableMap<HashMap<TableId, usize>>;
-pub(crate) type TableDefinitionMap = TableMap<String>;
-pub(crate) type TableNotifierMap = TableMap<Notifier>;
-pub(crate) type TableFragmentMap = TableMap<TableFragments>;
-
 /// The reason why the cluster is recovering.
 enum RecoveryReason {
     /// After bootstrap.
@@ -801,7 +788,12 @@ impl GlobalBarrierManager {
     }
 
     async fn failure_recovery(&mut self, err: MetaError) {
-        self.context.tracker.lock().await.abort_all(&err);
+        self.context
+            .tracker
+            .lock()
+            .await
+            .abort_all(&err, &self.context)
+            .await;
         self.checkpoint_control.clear_on_err(&err).await;
         self.pending_non_checkpoint_barriers.clear();
 
@@ -829,7 +821,12 @@ impl GlobalBarrierManager {
 
     async fn adhoc_recovery(&mut self) {
         let err = MetaErrorInner::AdhocRecovery.into();
-        self.context.tracker.lock().await.abort_all(&err);
+        self.context
+            .tracker
+            .lock()
+            .await
+            .abort_all(&err, &self.context)
+            .await;
         self.checkpoint_control.clear_on_err(&err).await;
 
         if self.enable_recovery {
@@ -880,7 +877,7 @@ impl GlobalBarrierManagerContext {
             notifier.notify_collected();
         });
         let has_remaining = self
-            .update_tracking_jobs(notifiers, command_ctx.clone(), create_mview_progress)
+            .update_tracking_jobs(command_ctx.clone(), create_mview_progress)
             .await?;
         let duration_sec = enqueue_time.stop_and_record();
         self.report_complete_event(duration_sec, &command_ctx);
@@ -942,7 +939,6 @@ impl GlobalBarrierManagerContext {
 
     async fn update_tracking_jobs(
         &self,
-        notifiers: Vec<Notifier>,
         command_ctx: Arc<CommandContext>,
         create_mview_progress: Vec<CreateMviewProgress>,
     ) -> MetaResult<bool> {
@@ -959,7 +955,6 @@ impl GlobalBarrierManagerContext {
                     if let Some(command) = tracker.add(
                         TrackingCommand {
                             context: command_ctx.clone(),
-                            notifiers,
                         },
                         &version_stats,
                     ) {

src/meta/src/barrier/notifier.rs

@@ -36,9 +36,6 @@ pub(crate) struct Notifier {
 
     /// Get notified when scheduled barrier is collected or failed.
     pub collected: Option<oneshot::Sender<MetaResult<()>>>,
-
-    /// Get notified when scheduled barrier is finished.
-    pub finished: Option<oneshot::Sender<MetaResult<()>>>,
 }
 
 impl Notifier {
@@ -63,31 +60,10 @@ impl Notifier {
         }
     }
 
-    /// Notify when we have finished a barrier from all actors. This function consumes `self`.
-    ///
-    /// Generally when a barrier is collected, it's also finished since it does not require further
-    /// report of finishing from actors.
-    /// However for creating MV, this is only called when all `BackfillExecutor` report it finished.
-    pub fn notify_finished(self) {
-        if let Some(tx) = self.finished {
-            tx.send(Ok(())).ok();
-        }
-    }
-
-    /// Notify when we failed to finish a barrier. This function consumes `self`.
-    pub fn notify_finish_failed(self, err: MetaError) {
-        if let Some(tx) = self.finished {
-            tx.send(Err(err)).ok();
-        }
-    }
-
     /// Notify when we failed to collect or finish a barrier. This function consumes `self`.
     pub fn notify_failed(self, err: MetaError) {
         if let Some(tx) = self.collected {
             tx.send(Err(err.clone())).ok();
         }
-        if let Some(tx) = self.finished {
-            tx.send(Err(err)).ok();
-        }
     }
 }