feat(meta): support drop creating materialized views for v2 backend

ci/scripts/run-e2e-test.sh

@@ -88,7 +88,7 @@ cluster_stop
 echo "--- e2e, $mode, batch"
 RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
 cluster_start
-sqllogictest -p 4566 -d dev './e2e_test/ddl/**/*.slt' --junit "batch-ddl-${profile}"
+sqllogictest -p 4566 -d dev './e2e_test/ddl/**/*.slt' --junit "batch-ddl-${profile}" --label "can-use-recover"
 if [[ "$mode" != "single-node" ]]; then
   sqllogictest -p 4566 -d dev './e2e_test/background_ddl/basic.slt' --junit "batch-ddl-${profile}"
 fi

e2e_test/ddl/drop/drop_creating_mv.slt

@@ -0,0 +1,76 @@
+statement ok
+create table t(v1 int);
+
+statement ok
+insert into t select * from generate_series(1, 10000);
+
+statement ok
+flush;
+
+statement ok
+set streaming_rate_limit=1;
+
+############## Test drop foreground mv
+onlyif can-use-recover
+system ok
+psql -h localhost -p 4566 -d dev -U root -c 'create materialized view m1 as select * from t;' &
+
+onlyif can-use-recover
+sleep 5s
+
+onlyif can-use-recover
+statement ok
+drop materialized view m1;
+
+############## Test drop background mv BEFORE recovery
+statement ok
+set background_ddl=true;
+
+onlyif can-use-recover
+statement ok
+create materialized view m1 as select * from t;
+
+onlyif can-use-recover
+sleep 5s
+
+onlyif can-use-recover
+statement ok
+drop materialized view m1;
+
+############## Test drop background mv AFTER recovery
+statement ok
+set background_ddl=true;
+
+onlyif can-use-recover
+statement ok
+create materialized view m1 as select * from t;
+
+onlyif can-use-recover
+sleep 5s
+
+onlyif can-use-recover
+statement ok
+recover;
+
+onlyif can-use-recover
+sleep 10s
+
+onlyif can-use-recover
+statement ok
+drop materialized view m1;
+
+############## Make sure the mv can still be successfully created later.
+statement ok
+set streaming_rate_limit=default;
+
+statement ok
+set background_ddl=false;
+
+statement ok
+create materialized view m1 as select * from t;
+
+statement ok
+drop materialized view m1;
+
+statement ok
+drop table t;

src/batch/src/worker_manager/worker_node_manager.rs

@@ -18,6 +18,7 @@ use std::time::Duration;
 
 use rand::seq::SliceRandom;
 use risingwave_common::bail;
+use risingwave_common::catalog::OBJECT_ID_PLACEHOLDER;
 use risingwave_common::hash::{WorkerSlotId, WorkerSlotMapping};
 use risingwave_common::vnode_mapping::vnode_placement::place_vnode;
 use risingwave_pb::common::{WorkerNode, WorkerType};
@@ -219,10 +220,20 @@ impl WorkerNodeManager {
 
     pub fn remove_streaming_fragment_mapping(&self, fragment_id: &FragmentId) {
         let mut guard = self.inner.write().unwrap();
-        guard
-            .streaming_fragment_vnode_mapping
-            .remove(fragment_id)
-            .unwrap();
+
+        let res = guard.streaming_fragment_vnode_mapping.remove(fragment_id);
+        match &res {
+            Some(_) => {}
+            None if OBJECT_ID_PLACEHOLDER == *fragment_id => {
+                // Do nothing for placeholder fragment.
+            }
+            None => {
+                panic!(
+                    "Streaming vnode mapping not found for fragment_id: {}",
+                    fragment_id
+                )
+            }
+        };
     }
 
     /// Returns fragment's vnode mapping for serving.

src/frontend/src/catalog/schema_catalog.rs

@@ -168,10 +168,7 @@ impl SchemaCatalog {
     pub fn create_index(&mut self, prost: &PbIndex) {
         let name = prost.name.clone();
         let id = prost.id.into();
-
-        let index_table = self
-            .get_created_table_by_id(&prost.index_table_id.into())
-            .unwrap();
+        let index_table = self.get_table_by_id(&prost.index_table_id.into()).unwrap();
         let primary_table = self
             .get_created_table_by_id(&prost.primary_table_id.into())
             .unwrap();

src/meta/src/barrier/mod.rs

@@ -819,24 +819,20 @@ impl GlobalBarrierManager {
         self.context.tracker.lock().await.abort_all(&err);
         self.checkpoint_control.clear_on_err(&err).await;
 
-        if self.enable_recovery {
-            self.context
-                .set_status(BarrierManagerStatus::Recovering(RecoveryReason::Adhoc));
-            let latest_snapshot = self.context.hummock_manager.latest_snapshot();
-            let prev_epoch = TracedEpoch::new(latest_snapshot.committed_epoch.into()); // we can only recover from the committed epoch
-            let span = tracing::info_span!(
-                "adhoc_recovery",
-                error = %err.as_report(),
-                prev_epoch = prev_epoch.value().0
-            );
+        self.context
+            .set_status(BarrierManagerStatus::Recovering(RecoveryReason::Adhoc));
+        let latest_snapshot = self.context.hummock_manager.latest_snapshot();
+        let prev_epoch = TracedEpoch::new(latest_snapshot.committed_epoch.into()); // we can only recover from the committed epoch
+        let span = tracing::info_span!(
+            "adhoc_recovery",
+            error = %err.as_report(),
+            prev_epoch = prev_epoch.value().0
+        );
 
-            // No need to clean dirty tables for barrier recovery,
-            // The foreground stream job should cleanup their own tables.
-            self.recovery(None).instrument(span).await;
-            self.context.set_status(BarrierManagerStatus::Running);
-        } else {
-            panic!("failed to execute barrier: {}", err.as_report());
-        }
+        // No need to clean dirty tables for barrier recovery,
+        // The foreground stream job should cleanup their own tables.
+        self.recovery(None).instrument(span).await;
+        self.context.set_status(BarrierManagerStatus::Running);
     }
 }