feat(meta, frontend): support drop cascade for tables, mviews, views, indexes, sources and sinks

e2e_test/ddl/drop/drop_index_cascade.slt

@@ -0,0 +1,29 @@
+statement ok
+set rw_streaming_enable_delta_join = true;
+
+statement ok
+create table a (a1 int, a2 int);
+
+statement ok
+create index i_a1 on a(a1);
+
+statement ok
+create table b (b1 int, b2 int);
+
+statement ok
+create index i_b1 on b(b1);
+
+statement ok
+create materialized view v as select * from a join b on a.a1 = b.b1;
+
+statement ok
+drop index i_a1 cascade;
+
+statement ok
+drop table a;
+
+statement ok
+drop table b;
+
+statement ok
+set rw_streaming_enable_delta_join = false;

e2e_test/ddl/drop/drop_mv_cascade.slt

@@ -0,0 +1,16 @@
+statement ok
+create materialized view mv as select 1 a
+
+statement ok
+CREATE view v AS select * from mv;
+
+statement ok
+CREATE view mv2 AS select * from v;
+
+statement ok
+CREATE SINK my_sink AS select * from mv2 WITH (
+    connector = 'blackhole'
+);
+
+statement ok
+drop materialized view mv cascade;

e2e_test/ddl/drop/drop_source_cascade.slt

@@ -0,0 +1,21 @@
+statement ok
+create source my_source (a int, b int) with (
+    connector = 'datagen',
+    datagen.rows.per.second = '15',
+    datagen.split.num = '1'
+);
+
+statement ok
+create view v as select * from my_source;
+
+statement ok
+create materialized view mv as select * from v;
+
+statement ok
+CREATE SINK my_sink AS select * from mv WITH (
+    connector = 'blackhole'
+);
+
+statement ok
+drop source my_source cascade;
+

e2e_test/ddl/drop/drop_table_cascade.slt

@@ -0,0 +1,17 @@
+statement ok
+create table t (a int, b int);
+
+statement ok
+create view v as select * from t;
+
+statement ok
+create materialized view mv as select * from v;
+
+statement ok
+CREATE SINK my_sink AS select * from mv WITH (
+    connector = 'blackhole'
+);
+
+statement ok
+drop table t cascade;
+

e2e_test/ddl/drop/drop_view_cascade.slt

@@ -0,0 +1,14 @@
+statement ok
+create view v as select 1;
+
+statement ok
+create materialized view mv as select * from v;
+
+statement ok
+CREATE SINK my_sink AS select * from mv WITH (
+    connector = 'blackhole'
+);
+
+statement ok
+drop view v cascade;
+

proto/ddl_service.proto

@@ -59,6 +59,7 @@ message CreateSourceResponse {
 
 message DropSourceRequest {
   uint32 source_id = 1;
+  bool cascade = 2;
 }
 
 message DropSourceResponse {
@@ -79,6 +80,7 @@ message CreateSinkResponse {
 
 message DropSinkRequest {
   uint32 sink_id = 1;
+  bool cascade = 2;
 }
 
 message DropSinkResponse {
@@ -99,6 +101,7 @@ message CreateMaterializedViewResponse {
 
 message DropMaterializedViewRequest {
   uint32 table_id = 1;
+  bool cascade = 2;
 }
 
 message DropMaterializedViewResponse {
@@ -118,6 +121,7 @@ message CreateViewResponse {
 
 message DropViewRequest {
   uint32 view_id = 1;
+  bool cascade = 2;
 }
 
 message DropViewResponse {
@@ -179,6 +183,7 @@ message DropTableRequest {
     uint32 id = 1;
   }
   uint32 table_id = 2;
+  bool cascade = 3;
 }
 
 message DropTableResponse {
@@ -208,6 +213,7 @@ message CreateIndexResponse {
 
 message DropIndexRequest {
   uint32 index_id = 1;
+  bool cascade = 2;
 }
 
 message DropIndexResponse {

src/frontend/planner_test/src/lib.rs

@@ -35,7 +35,9 @@ use risingwave_frontend::{
     build_graph, explain_stream_graph, Binder, Explain, FrontendOpts, OptimizerContext,
     OptimizerContextRef, PlanRef, Planner, WithOptions,
 };
-use risingwave_sqlparser::ast::{EmitMode, ExplainOptions, ObjectName, Statement};
+use risingwave_sqlparser::ast::{
+    AstOption, DropMode, EmitMode, ExplainOptions, ObjectName, Statement,
+};
 use risingwave_sqlparser::parser::Parser;
 use serde::{Deserialize, Serialize};
 
@@ -508,6 +510,7 @@ impl TestCase {
                         handler_args,
                         drop_statement.object_name,
                         drop_statement.if_exists,
+                        matches!(drop_statement.drop_mode, AstOption::Some(DropMode::Cascade)),
                     )
                     .await?;
                 }

src/frontend/src/catalog/catalog_service.rs

@@ -108,21 +108,26 @@ pub trait CatalogWriter: Send + Sync {
         connection: create_connection_request::Payload,
     ) -> Result<()>;
 
-    async fn drop_table(&self, source_id: Option<u32>, table_id: TableId) -> Result<()>;
+    async fn drop_table(
+        &self,
+        source_id: Option<u32>,
+        table_id: TableId,
+        cascade: bool,
+    ) -> Result<()>;
 
-    async fn drop_materialized_view(&self, table_id: TableId) -> Result<()>;
+    async fn drop_materialized_view(&self, table_id: TableId, cascade: bool) -> Result<()>;
 
-    async fn drop_view(&self, view_id: u32) -> Result<()>;
+    async fn drop_view(&self, view_id: u32, cascade: bool) -> Result<()>;
 
-    async fn drop_source(&self, source_id: u32) -> Result<()>;
+    async fn drop_source(&self, source_id: u32, cascade: bool) -> Result<()>;
 
-    async fn drop_sink(&self, sink_id: u32) -> Result<()>;
+    async fn drop_sink(&self, sink_id: u32, cascade: bool) -> Result<()>;
 
     async fn drop_database(&self, database_id: u32) -> Result<()>;
 
     async fn drop_schema(&self, schema_id: u32) -> Result<()>;
 
-    async fn drop_index(&self, index_id: IndexId) -> Result<()>;
+    async fn drop_index(&self, index_id: IndexId, cascade: bool) -> Result<()>;
 
     async fn drop_function(&self, function_id: FunctionId) -> Result<()>;
 
@@ -264,33 +269,44 @@ impl CatalogWriter for CatalogWriterImpl {
         self.wait_version(version).await
     }
 
-    async fn drop_table(&self, source_id: Option<u32>, table_id: TableId) -> Result<()> {
-        let version = self.meta_client.drop_table(source_id, table_id).await?;
+    async fn drop_table(
+        &self,
+        source_id: Option<u32>,
+        table_id: TableId,
+        cascade: bool,
+    ) -> Result<()> {
+        let version = self
+            .meta_client
+            .drop_table(source_id, table_id, cascade)
+            .await?;
         self.wait_version(version).await
     }
 
-    async fn drop_materialized_view(&self, table_id: TableId) -> Result<()> {
-        let version = self.meta_client.drop_materialized_view(table_id).await?;
+    async fn drop_materialized_view(&self, table_id: TableId, cascade: bool) -> Result<()> {
+        let version = self
+            .meta_client
+            .drop_materialized_view(table_id, cascade)
+            .await?;
         self.wait_version(version).await
     }
 
-    async fn drop_view(&self, view_id: u32) -> Result<()> {
-        let version = self.meta_client.drop_view(view_id).await?;
+    async fn drop_view(&self, view_id: u32, cascade: bool) -> Result<()> {
+        let version = self.meta_client.drop_view(view_id, cascade).await?;
         self.wait_version(version).await
     }
 
-    async fn drop_source(&self, source_id: u32) -> Result<()> {
-        let version = self.meta_client.drop_source(source_id).await?;
+    async fn drop_source(&self, source_id: u32, cascade: bool) -> Result<()> {
+        let version = self.meta_client.drop_source(source_id, cascade).await?;
         self.wait_version(version).await
     }
 
-    async fn drop_sink(&self, sink_id: u32) -> Result<()> {
-        let version = self.meta_client.drop_sink(sink_id).await?;
+    async fn drop_sink(&self, sink_id: u32, cascade: bool) -> Result<()> {
+        let version = self.meta_client.drop_sink(sink_id, cascade).await?;
         self.wait_version(version).await
     }
 
-    async fn drop_index(&self, index_id: IndexId) -> Result<()> {
-        let version = self.meta_client.drop_index(index_id).await?;
+    async fn drop_index(&self, index_id: IndexId, cascade: bool) -> Result<()> {
+        let version = self.meta_client.drop_index(index_id, cascade).await?;
         self.wait_version(version).await
     }
 

src/frontend/src/handler/drop_index.rs

@@ -28,6 +28,7 @@ pub async fn handle_drop_index(
     handler_args: HandlerArgs,
     index_name: ObjectName,
     if_exists: bool,
+    cascade: bool,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
     let db_name = session.database();
@@ -81,7 +82,7 @@ pub async fn handle_drop_index(
     };
 
     let catalog_writer = session.catalog_writer()?;
-    catalog_writer.drop_index(index_id).await?;
+    catalog_writer.drop_index(index_id, cascade).await?;
 
     Ok(PgResponse::empty_result(StatementType::DROP_INDEX))
 }

src/frontend/src/handler/drop_mv.rs

@@ -27,6 +27,7 @@ pub async fn handle_drop_mv(
     handler_args: HandlerArgs,
     table_name: ObjectName,
     if_exists: bool,
+    cascade: bool,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
     let db_name = session.database();
@@ -71,7 +72,9 @@ pub async fn handle_drop_mv(
     };
 
     let catalog_writer = session.catalog_writer()?;
-    catalog_writer.drop_materialized_view(table_id).await?;
+    catalog_writer
+        .drop_materialized_view(table_id, cascade)
+        .await?;
 
     Ok(PgResponse::empty_result(
         StatementType::DROP_MATERIALIZED_VIEW,

src/frontend/src/handler/drop_sink.rs

@@ -25,6 +25,7 @@ pub async fn handle_drop_sink(
     handler_args: HandlerArgs,
     sink_name: ObjectName,
     if_exists: bool,
+    cascade: bool,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
     let db_name = session.database();
@@ -55,7 +56,7 @@ pub async fn handle_drop_sink(
     };
 
     let catalog_writer = session.catalog_writer()?;
-    catalog_writer.drop_sink(sink_id.sink_id).await?;
+    catalog_writer.drop_sink(sink_id.sink_id, cascade).await?;
 
     Ok(PgResponse::empty_result(StatementType::DROP_SINK))
 }

src/frontend/src/handler/drop_source.rs

@@ -25,6 +25,7 @@ pub async fn handle_drop_source(
     handler_args: HandlerArgs,
     name: ObjectName,
     if_exists: bool,
+    cascade: bool,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
     let db_name = session.database();
@@ -62,7 +63,7 @@ pub async fn handle_drop_source(
     session.check_privilege_for_drop_alter(schema_name, &*source)?;
 
     let catalog_writer = session.catalog_writer()?;
-    catalog_writer.drop_source(source.id).await?;
+    catalog_writer.drop_source(source.id, cascade).await?;
 
     Ok(PgResponse::empty_result(StatementType::DROP_SOURCE))
 }

src/frontend/src/handler/drop_table.rs

@@ -26,6 +26,7 @@ pub async fn handle_drop_table(
     handler_args: HandlerArgs,
     table_name: ObjectName,
     if_exists: bool,
+    cascade: bool,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
     let db_name = session.database();
@@ -62,7 +63,7 @@ pub async fn handle_drop_table(
 
     let catalog_writer = session.catalog_writer()?;
     catalog_writer
-        .drop_table(source_id.map(|id| id.table_id), table_id)
+        .drop_table(source_id.map(|id| id.table_id), table_id, cascade)
         .await?;
 
     Ok(PgResponse::empty_result(StatementType::DROP_TABLE))

src/frontend/src/handler/drop_view.rs

@@ -25,6 +25,7 @@ pub async fn handle_drop_view(
     handler_args: HandlerArgs,
     table_name: ObjectName,
     if_exists: bool,
+    cascade: bool,
 ) -> Result<RwPgResponse> {
     let session = handler_args.session;
     let db_name = session.database();
@@ -56,7 +57,7 @@ pub async fn handle_drop_view(
     };
 
     let catalog_writer = session.catalog_writer()?;
-    catalog_writer.drop_view(view_id).await?;
+    catalog_writer.drop_view(view_id, cascade).await?;
 
     Ok(PgResponse::empty_result(StatementType::DROP_VIEW))
 }