feat(frontend): allow users to confirm if background_ddl is enabled f…

…or a materialized view (#19085)

e2e_test/ddl/drop/drop_creating_mv.slt

@@ -18,11 +18,18 @@ risedev psql -c 'create materialized view m1 as select * from t;' &
 onlyif can-use-recover
 sleep 5s
 
+onlyif can-use-recover
+query I
+select background_ddl from rw_catalog.rw_materialized_views where name='m1';
+----
+f
+
 onlyif can-use-recover
 statement ok
 drop materialized view m1;
 
 ############## Test drop background mv BEFORE recovery
+
 statement ok
 set background_ddl=true;
 
@@ -33,6 +40,12 @@ create materialized view m1 as select * from t;
 onlyif can-use-recover
 sleep 5s
 
+onlyif can-use-recover
+query I
+select background_ddl from rw_catalog.rw_materialized_views where name='m1';
+----
+t
+
 onlyif can-use-recover
 statement ok
 drop materialized view m1;
@@ -48,12 +61,24 @@ create materialized view m1 as select * from t;
 onlyif can-use-recover
 sleep 5s
 
+onlyif can-use-recover
+query I
+select background_ddl from rw_catalog.rw_materialized_views where name='m1';
+----
+t
+
 onlyif can-use-recover
 statement ok
 recover;
 
 onlyif can-use-recover
-sleep 10s
+sleep 5s
+
+onlyif can-use-recover
+query I
+select background_ddl from rw_catalog.rw_materialized_views where name='m1';
+----
+t
 
 onlyif can-use-recover
 statement ok
@@ -69,6 +94,11 @@ set background_ddl=false;
 statement ok
 create materialized view m1 as select * from t;
 
+query I
+select background_ddl from rw_catalog.rw_materialized_views where name='m1';
+----
+f
+
 statement ok
 drop materialized view m1;
 

src/frontend/planner_test/tests/testdata/output/subquery.yaml

@@ -249,7 +249,7 @@
         │       │ │ └─LogicalProject { exprs: [rw_subscriptions.id, rw_subscriptions.name, 'subscription':Varchar, rw_subscriptions.schema_id, rw_subscriptions.owner, rw_subscriptions.definition, rw_subscriptions.acl] }
         │       │ │   └─LogicalSysScan { table: rw_subscriptions, columns: [rw_subscriptions.id, rw_subscriptions.name, rw_subscriptions.schema_id, rw_subscriptions.owner, rw_subscriptions.definition, rw_subscriptions.acl, rw_subscriptions.initialized_at, rw_subscriptions.created_at, rw_subscriptions.initialized_at_cluster_version, rw_subscriptions.created_at_cluster_version] }
         │       │ └─LogicalProject { exprs: [rw_materialized_views.id, rw_materialized_views.name, 'materialized view':Varchar, rw_materialized_views.schema_id, rw_materialized_views.owner, rw_materialized_views.definition, rw_materialized_views.acl] }
-        │       │   └─LogicalSysScan { table: rw_materialized_views, columns: [rw_materialized_views.id, rw_materialized_views.name, rw_materialized_views.schema_id, rw_materialized_views.owner, rw_materialized_views.definition, rw_materialized_views.append_only, rw_materialized_views.acl, rw_materialized_views.initialized_at, rw_materialized_views.created_at, rw_materialized_views.initialized_at_cluster_version, rw_materialized_views.created_at_cluster_version] }
+        │       │   └─LogicalSysScan { table: rw_materialized_views, columns: [rw_materialized_views.id, rw_materialized_views.name, rw_materialized_views.schema_id, rw_materialized_views.owner, rw_materialized_views.definition, rw_materialized_views.append_only, rw_materialized_views.acl, rw_materialized_views.initialized_at, rw_materialized_views.created_at, rw_materialized_views.initialized_at_cluster_version, rw_materialized_views.created_at_cluster_version, rw_materialized_views.background_ddl] }
         │       └─LogicalProject { exprs: [rw_views.id, rw_views.name, 'view':Varchar, rw_views.schema_id, rw_views.owner, rw_views.definition, rw_views.acl] }
         │         └─LogicalSysScan { table: rw_views, columns: [rw_views.id, rw_views.name, rw_views.schema_id, rw_views.owner, rw_views.definition, rw_views.acl] }
         └─LogicalShare { id: 20 }

src/frontend/src/catalog/schema_catalog.rs

@@ -557,6 +557,14 @@ impl SchemaCatalog {
     }
 
     /// Iterate all materialized views, excluding the indices.
+    pub fn iter_all_mvs(&self) -> impl Iterator<Item = &Arc<TableCatalog>> {
+        self.table_by_name
+            .iter()
+            .filter(|(_, v)| v.is_mview() && valid_table_name(&v.name))
+            .map(|(_, v)| v)
+    }
+
+    /// Iterate created materialized views, excluding the indices.
     pub fn iter_created_mvs(&self) -> impl Iterator<Item = &Arc<TableCatalog>> {
         self.table_by_name
             .iter()

src/frontend/src/catalog/system_catalog/rw_catalog/rw_materialized_views.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use risingwave_common::catalog::CreateType;
 use risingwave_common::types::{Fields, Timestamptz};
 use risingwave_frontend_macro::system_catalog;
 use risingwave_pb::user::grant_privilege::Object;
@@ -33,6 +34,7 @@ struct RwMaterializedView {
     created_at: Option<Timestamptz>,
     initialized_at_cluster_version: Option<String>,
     created_at_cluster_version: Option<String>,
+    background_ddl: bool,
 }
 
 #[system_catalog(table, "rw_catalog.rw_materialized_views")]
@@ -45,7 +47,7 @@ fn read_rw_materialized_views(reader: &SysCatalogReaderImpl) -> Result<Vec<RwMat
 
     Ok(schemas
         .flat_map(|schema| {
-            schema.iter_created_mvs().map(|table| RwMaterializedView {
+            schema.iter_all_mvs().map(|table| RwMaterializedView {
                 id: table.id.table_id as i32,
                 name: table.name().into(),
                 schema_id: schema.id() as i32,
@@ -62,6 +64,7 @@ fn read_rw_materialized_views(reader: &SysCatalogReaderImpl) -> Result<Vec<RwMat
                 created_at: table.created_at_epoch.map(|e| e.as_timestamptz()),
                 initialized_at_cluster_version: table.initialized_at_cluster_version.clone(),
                 created_at_cluster_version: table.created_at_cluster_version.clone(),
+                background_ddl: table.create_type == CreateType::Background,
             })
         })
         .collect())