fix(catalog): add views' dependencies in rw_depend (#18596) (#18623)

Co-authored-by: Kexiang Wang <kx.wang@hotmail.com>

e2e_test/batch/catalog/rw_depend.slt.part

@@ -16,6 +16,9 @@ create materialized view mv1 as select t1.a from t1 join s1 on t1.a = s1.a;
 statement ok
 create materialized view mv2 as select * from mv1;
 
+statement ok
+create view v as select * from mv1;
+
 statement ok
 create sink sink1 from mv2 with (connector='blackhole');
 
@@ -37,13 +40,17 @@ mv2   mv1
 sink1 mv2
 sink2 t1
 t2    sink2
+v     mv1
 
 statement ok
 drop sink sink1;
 
 statement ok
 drop table t2 cascade;
 
+statement ok
+drop view v;
+
 statement ok
 drop materialized view mv2;
 

src/meta/src/controller/catalog.rs

@@ -674,6 +674,28 @@ impl CatalogController {
             })
             .collect_vec();
 
+        let view_dependencies: Vec<(ObjectId, ObjectId)> = ObjectDependency::find()
+            .select_only()
+            .columns([
+                object_dependency::Column::Oid,
+                object_dependency::Column::UsedBy,
+            ])
+            .join(
+                JoinType::InnerJoin,
+                object_dependency::Relation::Object1.def(),
+            )
+            .join(JoinType::InnerJoin, object::Relation::View.def())
+            .into_tuple()
+            .all(&inner.db)
+            .await?;
+
+        obj_dependencies.extend(view_dependencies.into_iter().map(|(view_id, table_id)| {
+            PbObjectDependencies {
+                object_id: table_id as _,
+                referenced_object_id: view_id as _,
+            }
+        }));
+
         let sink_dependencies: Vec<(SinkId, TableId)> = Sink::find()
             .select_only()
             .columns([sink::Column::SinkId, sink::Column::TargetTable])

src/meta/src/manager/catalog/mod.rs

@@ -4224,6 +4224,14 @@ impl CatalogManager {
                 }
             }
         }
+        for view in core.views.values() {
+            for referenced in &view.dependent_relations {
+                dependencies.push(PbObjectDependencies {
+                    object_id: view.id,
+                    referenced_object_id: *referenced,
+                });
+            }
+        }
         for sink in core.sinks.values() {
             for referenced in &sink.dependent_relations {
                 dependencies.push(PbObjectDependencies {