fix: synchronize the index on the mview when changing the owner or schema (#20093)

Author: yezizp2012

e2e_test/ddl/alter_owner.slt

@@ -67,6 +67,9 @@ v user1
 statement ok
 CREATE MATERIALIZED VIEW mv AS SELECT v1, (t.v2).v1 AS v21 FROM t;
 
+statement ok
+CREATE INDEX mv_idx ON mv(v1);
+
 statement ok
 ALTER MATERIALIZED VIEW mv OWNER TO user1;
 
@@ -85,6 +88,21 @@ WHERE
 ----
 mv user1
 
+query TT
+SELECT
+    pg_class.relname AS rel_name,
+    pg_roles.rolname AS owner
+FROM
+    pg_class
+    JOIN pg_namespace ON pg_namespace.oid = pg_class.relnamespace
+    JOIN pg_roles ON pg_roles.oid = pg_class.relowner
+WHERE
+    pg_namespace.nspname NOT LIKE 'pg_%'
+    AND pg_namespace.nspname != 'information_schema'
+    AND pg_class.relname = 'mv_idx';
+----
+mv_idx user1
+
 statement ok
 CREATE SOURCE src (v INT) WITH (
     connector = 'datagen',

e2e_test/ddl/alter_set_schema.slt

@@ -23,12 +23,21 @@ CREATE INDEX test_index1 ON test_table(u);
 statement ok
 CREATE INDEX test_index2 ON test_table(v);
 
+statement ok
+CREATE MATERIALIZED VIEW test_mv AS SELECT u FROM test_table;
+
+statement ok
+CREATE INDEX test_mv_index ON test_mv(u);
+
 statement ok
 ALTER TABLE test_table SET SCHEMA public;
 
 statement ok
 ALTER TABLE test_table SET SCHEMA test_schema;
 
+statement ok
+ALTER MATERIALIZED VIEW test_mv SET SCHEMA test_schema;
+
 query TT
 SELECT tablename, schemaname FROM pg_tables WHERE schemaname = 'test_schema';
 ----
@@ -39,6 +48,7 @@ SELECT indexname, schemaname FROM pg_indexes WHERE schemaname = 'test_schema';
 ----
 test_index1 test_schema
 test_index2 test_schema
+test_mv_index test_schema
 
 statement ok
 CREATE SOURCE test_source (v INT) WITH (
@@ -104,7 +114,7 @@ statement ok
 DROP SOURCE test_schema.test_source;
 
 statement ok
-DROP TABLE test_schema.test_table;
+DROP TABLE test_schema.test_table cascade;
 
 statement ok
 DROP SCHEMA test_schema;

src/meta/src/controller/catalog/alter_op.rs

@@ -331,20 +331,15 @@ impl CatalogController {
                 }
 
                 // indexes.
-                let (index_ids, mut table_ids): (Vec<IndexId>, Vec<TableId>) =
-                    if table.table_type == TableType::Table {
-                        Index::find()
-                            .select_only()
-                            .columns([index::Column::IndexId, index::Column::IndexTableId])
-                            .filter(index::Column::PrimaryTableId.eq(object_id))
-                            .into_tuple::<(IndexId, TableId)>()
-                            .all(&txn)
-                            .await?
-                            .into_iter()
-                            .unzip()
-                    } else {
-                        (vec![], vec![])
-                    };
+                let (index_ids, mut table_ids): (Vec<IndexId>, Vec<TableId>) = Index::find()
+                    .select_only()
+                    .columns([index::Column::IndexId, index::Column::IndexTableId])
+                    .filter(index::Column::PrimaryTableId.eq(object_id))
+                    .into_tuple::<(IndexId, TableId)>()
+                    .all(&txn)
+                    .await?
+                    .into_iter()
+                    .unzip();
                 relations.push(PbRelationInfo::Table(ObjectModel(table, obj).into()));
 
                 // internal tables.
@@ -501,8 +496,7 @@ impl CatalogController {
                     .await?
                     .ok_or_else(|| MetaError::catalog_id_not_found("table", object_id))?;
                 check_relation_name_duplicate(&table.name, database_id, new_schema, &txn).await?;
-                let (associated_src_id, table_type) =
-                    (table.optional_associated_source_id, table.table_type);
+                let associated_src_id = table.optional_associated_source_id;
 
                 let mut obj = obj.into_active_model();
                 obj.schema_id = Set(Some(new_schema));
@@ -531,24 +525,20 @@ impl CatalogController {
                 let (index_ids, (index_names, mut table_ids)): (
                     Vec<IndexId>,
                     (Vec<String>, Vec<TableId>),
-                ) = if table_type == TableType::Table {
-                    Index::find()
-                        .select_only()
-                        .columns([
-                            index::Column::IndexId,
-                            index::Column::Name,
-                            index::Column::IndexTableId,
-                        ])
-                        .filter(index::Column::PrimaryTableId.eq(object_id))
-                        .into_tuple::<(IndexId, String, TableId)>()
-                        .all(&txn)
-                        .await?
-                        .into_iter()
-                        .map(|(id, name, t_id)| (id, (name, t_id)))
-                        .unzip()
-                } else {
-                    (vec![], (vec![], vec![]))
-                };
+                ) = Index::find()
+                    .select_only()
+                    .columns([
+                        index::Column::IndexId,
+                        index::Column::Name,
+                        index::Column::IndexTableId,
+                    ])
+                    .filter(index::Column::PrimaryTableId.eq(object_id))
+                    .into_tuple::<(IndexId, String, TableId)>()
+                    .all(&txn)
+                    .await?
+                    .into_iter()
+                    .map(|(id, name, t_id)| (id, (name, t_id)))
+                    .unzip();
 
                 // internal tables.
                 let internal_tables: Vec<TableId> = Table::find()