feat(catalog): add system catalog rw_depend (#15385)

Signed-off-by: Runji Wang <[email protected]>

e2e_test/batch/catalog/rw_depend.slt.part

@@ -0,0 +1,43 @@
+statement ok
+create table t1 (a int);
+
+statement ok
+create source s1 (a int) with (connector='datagen');
+
+statement ok
+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 sink sink1 from mv2 with (connector='blackhole');
+
+
+# equivalent to:
+# select objid::regclass, refobjid::regclass from rw_depend;
+query TT rowsort
+select r1.name, r2.name relname
+from rw_depend d
+join rw_relations r1 on d.objid = r1.id
+join rw_relations r2 on d.refobjid = r2.id;
+----
+mv1   s1
+mv1   t1
+mv2   mv1
+sink1 mv2
+
+statement ok
+drop sink sink1;
+
+statement ok
+drop materialized view mv2;
+
+statement ok
+drop materialized view mv1;
+
+statement ok
+drop source s1;
+
+statement ok
+drop table t1;

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

@@ -18,6 +18,7 @@ mod rw_columns;
 mod rw_connections;
 mod rw_databases;
 mod rw_ddl_progress;
+mod rw_depend;
 mod rw_description;
 mod rw_event_logs;
 mod rw_fragment_parallelism;

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

@@ -0,0 +1,57 @@
+// Copyright 2024 RisingWave Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use risingwave_common::types::Fields;
+use risingwave_frontend_macro::system_catalog;
+
+use crate::catalog::system_catalog::SysCatalogReaderImpl;
+use crate::error::Result;
+
+/// The catalog `rw_depend` records the dependency relationships between tables, mviews etc.
+#[derive(Fields)]
+#[primary_key(objid, refobjid)]
+struct RwDepend {
+    /// The OID of the specific dependent object
+    objid: i32,
+    /// The OID of the specific referenced object
+    refobjid: i32,
+}
+
+#[system_catalog(table, "rw_catalog.rw_depend")]
+fn read_rw_depend(reader: &SysCatalogReaderImpl) -> Result<Vec<RwDepend>> {
+    let catalog_reader = reader.catalog_reader.read_guard();
+
+    let mut depends = vec![];
+    for schema in catalog_reader.iter_schemas(&reader.auth_context.database)? {
+        for table in schema.iter_table().chain(schema.iter_mv()) {
+            for referenced in &table.dependent_relations {
+                let depend = RwDepend {
+                    objid: table.id.table_id as i32,
+                    refobjid: referenced.table_id as i32,
+                };
+                depends.push(depend);
+            }
+        }
+        for sink in schema.iter_sink() {
+            for referenced in &sink.dependent_relations {
+                let depend = RwDepend {
+                    objid: sink.id.sink_id as i32,
+                    refobjid: referenced.table_id as i32,
+                };
+                depends.push(depend);
+            }
+        }
+    }
+    Ok(depends)
+}