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

src/frontend/src/binder/relation/table_or_source.rs

@@ -120,9 +120,9 @@ impl Binder {
                                 table_name
                             );
                         }
-                    } else if let Ok((table_catalog, schema_name)) =
-                        self.catalog
-                            .get_table_by_name(&self.db_name, schema_path, table_name)
+                    } else if let Ok((table_catalog, schema_name)) = self
+                        .catalog
+                        .get_created_table_by_name(&self.db_name, schema_path, table_name)
                     {
                         self.resolve_table_relation(table_catalog.clone(), schema_name, as_of)?
                     } else if let Ok((source_catalog, _)) =
@@ -163,7 +163,9 @@ impl Binder {
                             if let Ok(schema) =
                                 self.catalog.get_schema_by_name(&self.db_name, schema_name)
                             {
-                                if let Some(table_catalog) = schema.get_table_by_name(table_name) {
+                                if let Some(table_catalog) =
+                                    schema.get_created_table_by_name(table_name)
+                                {
                                     return self.resolve_table_relation(
                                         table_catalog.clone(),
                                         &schema_name.clone(),
@@ -314,7 +316,7 @@ impl Binder {
         };
         let (table_catalog, schema_name) =
             self.catalog
-                .get_table_by_name(db_name, schema_path, table_name)?;
+                .get_created_table_by_name(db_name, schema_path, table_name)?;
         let table_catalog = table_catalog.clone();
 
         let table_id = table_catalog.id();
@@ -352,7 +354,7 @@ impl Binder {
 
         let (table, _schema_name) =
             self.catalog
-                .get_table_by_name(db_name, schema_path, table_name)?;
+                .get_created_table_by_name(db_name, schema_path, table_name)?;
 
         match table.table_type() {
             TableType::Table => {}

src/frontend/src/catalog/database_catalog.rs

@@ -97,7 +97,7 @@ impl DatabaseCatalog {
     pub fn find_schema_containing_table_id(&self, table_id: &TableId) -> Option<&SchemaCatalog> {
         self.schema_by_name
             .values()
-            .find(|schema| schema.get_table_by_id(table_id).is_some())
+            .find(|schema| schema.get_created_table_by_id(table_id).is_some())
     }
 
     pub fn get_grant_object_by_oid(&self, oid: u32) -> Option<Object> {

src/frontend/src/catalog/root_catalog.rs

@@ -320,7 +320,7 @@ impl Catalog {
                 .iter_schemas_mut()
                 .find(|schema| {
                     schema.id() != proto.schema_id
-                        && schema.get_table_by_id(&proto.id.into()).is_some()
+                        && schema.get_created_table_by_id(&proto.id.into()).is_some()
                 })
                 .unwrap()
                 .drop_table(proto.id.into());
@@ -587,7 +587,7 @@ impl Catalog {
     }
 
     pub fn get_table_name_by_id(&self, table_id: TableId) -> CatalogResult<String> {
-        self.get_table_by_id(&table_id)
+        self.get_any_table_by_id(&table_id)
             .map(|table| table.name.clone())
     }
 
@@ -636,7 +636,8 @@ impl Catalog {
     }
 
     /// Used to get `TableCatalog` for Materialized Views, Tables and Indexes.
-    pub fn get_table_by_name<'a>(
+    /// Retrieves all tables, created or creating.
+    pub fn get_any_table_by_name<'a>(
         &self,
         db_name: &str,
         schema_path: SchemaPath<'a>,
@@ -651,7 +652,24 @@ impl Catalog {
             .ok_or_else(|| CatalogError::NotFound("table", table_name.to_string()))
     }
 
-    pub fn get_table_by_id(&self, table_id: &TableId) -> CatalogResult<&Arc<TableCatalog>> {
+    /// Used to get `TableCatalog` for Materialized Views, Tables and Indexes.
+    /// Retrieves only created tables.
+    pub fn get_created_table_by_name<'a>(
+        &self,
+        db_name: &str,
+        schema_path: SchemaPath<'a>,
+        table_name: &str,
+    ) -> CatalogResult<(&Arc<TableCatalog>, &'a str)> {
+        schema_path
+            .try_find(|schema_name| {
+                Ok(self
+                    .get_schema_by_name(db_name, schema_name)?
+                    .get_created_table_by_name(table_name))
+            })?
+            .ok_or_else(|| CatalogError::NotFound("table", table_name.to_string()))
+    }
+
+    pub fn get_any_table_by_id(&self, table_id: &TableId) -> CatalogResult<&Arc<TableCatalog>> {
         self.table_by_id
             .get(table_id)
             .ok_or_else(|| CatalogError::NotFound("table id", table_id.to_string()))
@@ -665,7 +683,7 @@ impl Catalog {
     ) -> CatalogResult<&Arc<TableCatalog>> {
         let table_id = TableId::from(table_id);
         for schema in self.get_database_by_name(db_name)?.iter_schemas() {
-            if let Some(table) = schema.get_table_by_id(&table_id) {
+            if let Some(table) = schema.get_created_table_by_id(&table_id) {
                 return Ok(table);
             }
         }
@@ -702,7 +720,7 @@ impl Catalog {
 
         if found {
             let mut table = self
-                .get_table_by_id(table_id)
+                .get_any_table_by_id(table_id)
                 .unwrap()
                 .to_prost(schema_id, database_id);
             table.name = table_name.to_string();
@@ -939,7 +957,7 @@ impl Catalog {
     ) -> CatalogResult<()> {
         let schema = self.get_schema_by_name(db_name, schema_name)?;
 
-        if let Some(table) = schema.get_table_by_name(relation_name) {
+        if let Some(table) = schema.get_created_table_by_name(relation_name) {
             if table.is_index() {
                 Err(CatalogError::Duplicated("index", relation_name.to_string()))
             } else if table.is_mview() {
@@ -1069,7 +1087,7 @@ impl Catalog {
                 #[allow(clippy::manual_map)]
                 if let Some(item) = schema.get_system_table_by_name(class_name) {
                     Ok(Some(item.id().into()))
-                } else if let Some(item) = schema.get_table_by_name(class_name) {
+                } else if let Some(item) = schema.get_created_table_by_name(class_name) {
                     Ok(Some(item.id().into()))
                 } else if let Some(item) = schema.get_index_by_name(class_name) {
                     Ok(Some(item.id.into()))

src/frontend/src/catalog/schema_catalog.rs

@@ -17,7 +17,7 @@ use std::collections::HashMap;
 use std::sync::Arc;
 
 use itertools::Itertools;
-use risingwave_common::catalog::{valid_table_name, FunctionId, IndexId, TableId};
+use risingwave_common::catalog::{valid_table_name, FunctionId, IndexId, StreamJobStatus, TableId};
 use risingwave_common::types::DataType;
 use risingwave_connector::sink::catalog::SinkCatalog;
 pub use risingwave_expr::sig::*;
@@ -128,9 +128,11 @@ impl SchemaCatalog {
         let name = prost.name.clone();
         let id = prost.id.into();
         let old_index = self.index_by_id.get(&id).unwrap();
-        let index_table = self.get_table_by_id(&prost.index_table_id.into()).unwrap();
+        let index_table = self
+            .get_created_table_by_id(&prost.index_table_id.into())
+            .unwrap();
         let primary_table = self
-            .get_table_by_id(&prost.primary_table_id.into())
+            .get_created_table_by_id(&prost.primary_table_id.into())
             .unwrap();
         let index: IndexCatalog = IndexCatalog::build_from(prost, index_table, primary_table);
         let index_ref = Arc::new(index);
@@ -167,9 +169,11 @@ impl SchemaCatalog {
         let name = prost.name.clone();
         let id = prost.id.into();
 
-        let index_table = self.get_table_by_id(&prost.index_table_id.into()).unwrap();
+        let index_table = self
+            .get_created_table_by_id(&prost.index_table_id.into())
+            .unwrap();
         let primary_table = self
-            .get_table_by_id(&prost.primary_table_id.into())
+            .get_created_table_by_id(&prost.primary_table_id.into())
             .unwrap();
         let index: IndexCatalog = IndexCatalog::build_from(prost, index_table, primary_table);
         let index_ref = Arc::new(index);
@@ -559,7 +563,7 @@ impl SchemaCatalog {
     pub fn iter_mv(&self) -> impl Iterator<Item = &Arc<TableCatalog>> {
         self.table_by_name
             .iter()
-            .filter(|(_, v)| v.is_mview() && valid_table_name(&v.name))
+            .filter(|(_, v)| v.is_mview() && valid_table_name(&v.name) && v.is_created())
             .map(|(_, v)| v)
     }
 
@@ -605,10 +609,22 @@ impl SchemaCatalog {
         self.table_by_name.get(table_name)
     }
 
+    pub fn get_created_table_by_name(&self, table_name: &str) -> Option<&Arc<TableCatalog>> {
+        self.table_by_name
+            .get(table_name)
+            .filter(|&table| table.stream_job_status == StreamJobStatus::Created)
+    }
+
     pub fn get_table_by_id(&self, table_id: &TableId) -> Option<&Arc<TableCatalog>> {
         self.table_by_id.get(table_id)
     }
 
+    pub fn get_created_table_by_id(&self, table_id: &TableId) -> Option<&Arc<TableCatalog>> {
+        self.table_by_id
+            .get(table_id)
+            .filter(|&table| table.stream_job_status == StreamJobStatus::Created)
+    }
+
     pub fn get_view_by_name(&self, view_name: &str) -> Option<&Arc<ViewCatalog>> {
         self.view_by_name.get(view_name)
     }
@@ -765,7 +781,7 @@ impl SchemaCatalog {
 
     pub fn get_grant_object_by_oid(&self, oid: u32) -> Option<Object> {
         #[allow(clippy::manual_map)]
-        if self.get_table_by_id(&TableId::new(oid)).is_some()
+        if self.get_created_table_by_id(&TableId::new(oid)).is_some()
             || self.get_index_by_id(&IndexId::new(oid)).is_some()
         {
             Some(Object::TableId(oid))

src/frontend/src/catalog/table_catalog.rs

@@ -493,6 +493,10 @@ impl TableCatalog {
                 .collect(),
         )
     }
+
+    pub fn is_created(&self) -> bool {
+        self.stream_job_status == StreamJobStatus::Created
+    }
 }
 
 impl From<PbTable> for TableCatalog {

src/frontend/src/handler/alter_owner.rs

@@ -76,8 +76,11 @@ pub async fn handle_alter_owner(
         (
             match stmt_type {
                 StatementType::ALTER_TABLE | StatementType::ALTER_MATERIALIZED_VIEW => {
-                    let (table, schema_name) =
-                        catalog_reader.get_table_by_name(db_name, schema_path, &real_obj_name)?;
+                    let (table, schema_name) = catalog_reader.get_created_table_by_name(
+                        db_name,
+                        schema_path,
+                        &real_obj_name,
+                    )?;
                     session.check_privilege_for_drop_alter(schema_name, &**table)?;
                     let schema_id = catalog_reader
                         .get_schema_by_name(db_name, schema_name)?

src/frontend/src/handler/alter_parallelism.rs

@@ -52,7 +52,7 @@ pub async fn handle_alter_parallelism(
             | StatementType::ALTER_MATERIALIZED_VIEW
             | StatementType::ALTER_INDEX => {
                 let (table, schema_name) =
-                    reader.get_table_by_name(db_name, schema_path, &real_table_name)?;
+                    reader.get_created_table_by_name(db_name, schema_path, &real_table_name)?;
 
                 match (table.table_type(), stmt_type) {
                     (TableType::Internal, _) => {

src/frontend/src/handler/alter_rename.rs

@@ -43,7 +43,7 @@ pub async fn handle_rename_table(
     let table_id = {
         let reader = session.env().catalog_reader().read_guard();
         let (table, schema_name) =
-            reader.get_table_by_name(db_name, schema_path, &real_table_name)?;
+            reader.get_created_table_by_name(db_name, schema_path, &real_table_name)?;
         if table_type != table.table_type {
             return Err(ErrorCode::InvalidInputSyntax(format!(
                 "\"{table_name}\" is not a {}",
@@ -332,7 +332,6 @@ pub async fn handle_rename_database(
 
 #[cfg(test)]
 mod tests {
-
     use risingwave_common::catalog::{DEFAULT_DATABASE_NAME, DEFAULT_SCHEMA_NAME};
 
     use crate::catalog::root_catalog::SchemaPath;
@@ -350,7 +349,7 @@ mod tests {
         let table_id = {
             let catalog_reader = session.env().catalog_reader().read_guard();
             catalog_reader
-                .get_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "t")
+                .get_created_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "t")
                 .unwrap()
                 .0
                 .id
@@ -362,7 +361,7 @@ mod tests {
 
         let catalog_reader = session.env().catalog_reader().read_guard();
         let altered_table_name = catalog_reader
-            .get_table_by_id(&table_id)
+            .get_any_table_by_id(&table_id)
             .unwrap()
             .name()
             .to_string();

src/frontend/src/handler/alter_set_schema.rs

@@ -48,8 +48,11 @@ pub async fn handle_alter_set_schema(
 
         match stmt_type {
             StatementType::ALTER_TABLE | StatementType::ALTER_MATERIALIZED_VIEW => {
-                let (table, old_schema_name) =
-                    catalog_reader.get_table_by_name(db_name, schema_path, &real_obj_name)?;
+                let (table, old_schema_name) = catalog_reader.get_created_table_by_name(
+                    db_name,
+                    schema_path,
+                    &real_obj_name,
+                )?;
                 if old_schema_name == new_schema_name {
                     return Ok(RwPgResponse::empty_result(stmt_type));
                 }
@@ -209,7 +212,7 @@ pub mod tests {
             let catalog_reader = session.env().catalog_reader().read_guard();
             let schema_path = SchemaPath::Name(schema_name);
             catalog_reader
-                .get_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "test_table")
+                .get_created_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "test_table")
                 .unwrap()
                 .0
                 .clone()

src/frontend/src/handler/alter_streaming_rate_limit.rs

@@ -42,7 +42,7 @@ pub async fn handle_alter_streaming_rate_limit(
         PbThrottleTarget::Mv => {
             let reader = session.env().catalog_reader().read_guard();
             let (table, schema_name) =
-                reader.get_table_by_name(db_name, schema_path, &real_table_name)?;
+                reader.get_created_table_by_name(db_name, schema_path, &real_table_name)?;
             if table.table_type != TableType::MaterializedView {
                 return Err(ErrorCode::InvalidInputSyntax(format!(
                     "\"{table_name}\" is not a materialized view",
@@ -62,7 +62,7 @@ pub async fn handle_alter_streaming_rate_limit(
         PbThrottleTarget::TableWithSource => {
             let reader = session.env().catalog_reader().read_guard();
             let (table, schema_name) =
-                reader.get_table_by_name(db_name, schema_path, &real_table_name)?;
+                reader.get_created_table_by_name(db_name, schema_path, &real_table_name)?;
             session.check_privilege_for_drop_alter(schema_name, &**table)?;
             // Get the corresponding source catalog.
             let source_id = if let Some(id) = table.associated_source_id {

src/frontend/src/handler/alter_table_column.rs

@@ -250,7 +250,7 @@ pub fn fetch_table_catalog_for_alter(
     let original_catalog = {
         let reader = session.env().catalog_reader().read_guard();
         let (table, schema_name) =
-            reader.get_table_by_name(db_name, schema_path, &real_table_name)?;
+            reader.get_created_table_by_name(db_name, schema_path, &real_table_name)?;
 
         match table.table_type() {
             TableType::Table => {}
@@ -290,7 +290,7 @@ mod tests {
         let get_table = || {
             let catalog_reader = session.env().catalog_reader().read_guard();
             catalog_reader
-                .get_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "t")
+                .get_created_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "t")
                 .unwrap()
                 .0
                 .clone()

src/frontend/src/handler/create_index.rs

@@ -61,7 +61,8 @@ pub(crate) fn resolve_index_schema(
 
     let catalog_reader = session.env().catalog_reader();
     let read_guard = catalog_reader.read_guard();
-    let (table, schema_name) = read_guard.get_table_by_name(db_name, schema_path, &table_name)?;
+    let (table, schema_name) =
+        read_guard.get_created_table_by_name(db_name, schema_path, &table_name)?;
     Ok((schema_name.to_string(), table.clone(), index_table_name))
 }
 

src/frontend/src/handler/create_mv.rs

@@ -277,7 +277,7 @@ pub mod tests {
 
         // Check table exists.
         let (table, _) = catalog_reader
-            .get_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "mv1")
+            .get_created_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "mv1")
             .unwrap();
         assert_eq!(table.name(), "mv1");
 

src/frontend/src/handler/create_sink.rs

@@ -561,7 +561,7 @@ fn check_cycle_for_sink(
             path: &mut Vec<String>,
         ) -> Result<()> {
             for table_id in dependent_jobs {
-                if let Ok(table) = self.reader.get_table_by_id(table_id) {
+                if let Ok(table) = self.reader.get_any_table_by_id(table_id) {
                     path.push(table.name.clone());
                     self.visit_table(table.as_ref(), target_table_id, path)?;
                     path.pop();
@@ -922,7 +922,7 @@ pub mod tests {
 
         // Check table exists.
         let (table, schema_name) = catalog_reader
-            .get_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "mv1")
+            .get_created_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "mv1")
             .unwrap();
         assert_eq!(table.name(), "mv1");
 

src/frontend/src/handler/create_table.rs

@@ -1348,7 +1348,7 @@ mod tests {
 
         // Check table exists.
         let (table, _) = catalog_reader
-            .get_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "t")
+            .get_created_table_by_name(DEFAULT_DATABASE_NAME, schema_path, "t")
             .unwrap();
         assert_eq!(table.name(), "t");