feat(catalog): support DBeaver constraints view (risingwavelabs#15227)

e2e_test/batch/catalog/pg_class.slt.part

@@ -11,7 +11,7 @@ SELECT oid,relname,relowner,relkind FROM pg_catalog.pg_class ORDER BY oid limit
 8 pg_cast 1 r
 9 pg_class 1 v
 10 pg_collation 1 v
-11 pg_constraint 1 v
+11 pg_constraint 1 r
 12 pg_conversion 1 v
 13 pg_database 1 v
 14 pg_depend 1 v
@@ -20,4 +20,4 @@ SELECT oid,relname,relowner,relkind FROM pg_catalog.pg_class ORDER BY oid limit
 query ITIT
 SELECT oid,relname,relowner,relkind FROM pg_catalog.pg_class WHERE oid = 'pg_namespace'::regclass;
 ----
-24 pg_namespace 1 v
+25 pg_namespace 1 v

e2e_test/batch/catalog/pg_constraint.slt.part

@@ -0,0 +1,10 @@
+statement ok
+create table t(a int, b int, c varchar, primary key(a,b));
+
+query TTTT
+select conname, contype, conkey from pg_constraint where conname='t_pkey';
+----
+t_pkey p {1,2}
+
+statement ok
+drop table t;

src/frontend/planner_test/tests/planner_test_runner.rs

@@ -43,7 +43,7 @@ fn main() {
             let file_name = path.file_name().unwrap().to_string_lossy().to_string();
             let test_case_name = file_name.split('.').next().unwrap().to_string();
 
-            tests.push(Trial::test(format!("{test_case_name}_test"), move || {
+            tests.push(Trial::test(test_case_name, move || {
                 let path = test_data_dir().join("input").join(file_name);
 
                 let file_content = std::fs::read_to_string(&path).unwrap();

src/frontend/planner_test/tests/testdata/input/dbeaver.yaml

@@ -0,0 +1,43 @@
+- sql: |
+    SELECT DISTINCT dep.deptype, dep.classid, dep.objid, cl.relkind, attr.attname,pg_get_expr(ad.adbin, ad.adrelid) adefval,
+        CASE WHEN cl.relkind IS NOT NULL THEN cl.relkind::text || COALESCE(dep.objsubid::text, '')::text
+            WHEN tg.oid IS NOT NULL THEN 'T'::text
+            WHEN ty.oid IS NOT NULL THEN 'y'::text
+            WHEN ns.oid IS NOT NULL THEN 'n'::text
+            WHEN pr.oid IS NOT NULL THEN 'p'::text
+            WHEN la.oid IS NOT NULL THEN 'l'::text
+            WHEN rw.oid IS NOT NULL THEN 'R'::text
+            WHEN co.oid IS NOT NULL THEN 'C'::text || contype::text
+            WHEN ad.oid IS NOT NULL THEN 'A'::text
+            ELSE ''
+        END AS type,
+        COALESCE(coc.relname, clrw.relname, tgr.relname) AS ownertable,
+        CASE WHEN cl.relname IS NOT NULL AND att.attname IS NOT NULL THEN cl.relname || '.' || att.attname
+        ELSE COALESCE(cl.relname, co.conname, pr.proname, tg.tgname, ty.typname, la.lanname, rw.rulename, ns.nspname)
+        END AS refname,
+        COALESCE(nsc.nspname, nso.nspname, nsp.nspname, nst.nspname, nsrw.nspname, tgrn.nspname) AS nspname
+    FROM pg_depend dep
+    LEFT JOIN pg_class cl ON dep.objid=cl.oid
+    LEFT JOIN pg_attribute att ON dep.objid=att.attrelid AND dep.objsubid=att.attnum
+    LEFT JOIN pg_namespace nsc ON cl.relnamespace=nsc.oid
+    LEFT JOIN pg_proc pr ON dep.objid=pr.oid
+    LEFT JOIN pg_namespace nsp ON pr.pronamespace=nsp.oid
+    LEFT JOIN pg_trigger tg ON dep.objid=tg.oid
+    LEFT JOIN pg_class tgr ON tg.tgrelid=tgr.oid
+    LEFT JOIN pg_namespace tgrn ON tgr.relnamespace=tgrn.oid
+    LEFT JOIN pg_type ty ON dep.objid=ty.oid
+    LEFT JOIN pg_namespace nst ON ty.typnamespace=nst.oid
+    LEFT JOIN pg_constraint co ON dep.objid=co.oid
+    LEFT JOIN pg_class coc ON co.conrelid=coc.oid
+    LEFT JOIN pg_namespace nso ON co.connamespace=nso.oid
+    LEFT JOIN pg_rewrite rw ON dep.objid=rw.oid
+    LEFT JOIN pg_class clrw ON clrw.oid=rw.ev_class
+    LEFT JOIN pg_namespace nsrw ON clrw.relnamespace=nsrw.oid
+    LEFT JOIN pg_language la ON dep.objid=la.oid
+    LEFT JOIN pg_namespace ns ON dep.objid=ns.oid
+    LEFT JOIN pg_attrdef ad ON ad.oid=dep.objid
+    LEFT JOIN pg_attribute attr ON attr.attrelid=ad.adrelid and attr.attnum=ad.adnum
+    WHERE dep.refobjid=$1
+    ORDER BY type
+  expected_outputs:
+    - batch_plan

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

@@ -227,7 +227,7 @@
     └─LogicalFilter { predicate: In($expr1, 'r':Varchar, 'p':Varchar, 'v':Varchar, 'm':Varchar, 'S':Varchar, 'f':Varchar, '':Varchar) AND (rw_schemas.name <> 'pg_catalog':Varchar) AND Not(RegexpEq(rw_schemas.name, '^pg_toast':Varchar)) AND (rw_schemas.name <> 'information_schema':Varchar) }
       └─LogicalJoin { type: LeftOuter, on: (rw_schemas.id = rw_tables.schema_id), output: all }
         ├─LogicalShare { id: 16 }
-        │ └─LogicalProject { exprs: [rw_tables.id, rw_tables.name, rw_tables.schema_id, rw_tables.owner, 'p':Varchar, Case(('table':Varchar = 'table':Varchar), 'r':Varchar, ('table':Varchar = 'system table':Varchar), 'r':Varchar, ('table':Varchar = 'index':Varchar), 'i':Varchar, ('table':Varchar = 'view':Varchar), 'v':Varchar, ('table':Varchar = 'materialized view':Varchar), 'm':Varchar) as $expr1, 0:Int32, 0:Int32, Array as $expr2] }
+        │ └─LogicalProject { exprs: [rw_tables.id, rw_tables.name, rw_tables.schema_id, rw_tables.owner, 'p':Varchar, Case(('table':Varchar = 'table':Varchar), 'r':Varchar, ('table':Varchar = 'system table':Varchar), 'r':Varchar, ('table':Varchar = 'index':Varchar), 'i':Varchar, ('table':Varchar = 'view':Varchar), 'v':Varchar, ('table':Varchar = 'materialized view':Varchar), 'm':Varchar) as $expr1, 0:Int32, 0:Int32, Array as $expr2, null:Varchar] }
         │   └─LogicalShare { id: 14 }
         │     └─LogicalUnion { all: true }
         │       ├─LogicalUnion { all: true }

src/frontend/src/binder/expr/function.rs

@@ -1323,6 +1323,7 @@ impl Binder {
                 ("pg_table_is_visible", raw_literal(ExprImpl::literal_bool(true))),
                 ("pg_type_is_visible", raw_literal(ExprImpl::literal_bool(true))),
                 ("pg_get_constraintdef", raw_literal(ExprImpl::literal_null(DataType::Varchar))),
+                ("pg_get_partkeydef", raw_literal(ExprImpl::literal_null(DataType::Varchar))),
                 ("pg_encoding_to_char", raw_literal(ExprImpl::literal_varchar("UTF8".into()))),
                 ("has_database_privilege", raw_literal(ExprImpl::literal_bool(true))),
                 ("pg_backend_pid", raw(|binder, _inputs| {

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

@@ -30,20 +30,23 @@ mod pg_index;
 mod pg_indexes;
 mod pg_inherits;
 mod pg_keywords;
+mod pg_language;
 mod pg_locks;
 mod pg_matviews;
 mod pg_namespace;
 mod pg_opclass;
 mod pg_operator;
 mod pg_partitioned_table;
 mod pg_proc;
+mod pg_rewrite;
 mod pg_roles;
 mod pg_settings;
 mod pg_shadow;
 mod pg_shdescription;
 mod pg_stat_activity;
 mod pg_tables;
 mod pg_tablespace;
+mod pg_trigger;
 mod pg_type;
 mod pg_user;
 mod pg_views;

src/frontend/src/catalog/system_catalog/pg_catalog/pg_class.rs

@@ -29,7 +29,8 @@ use risingwave_frontend_macro::system_catalog;
     END relkind,
     0 AS relam,
     0 AS reltablespace,
-    ARRAY[]::varchar[] AS reloptions
+    ARRAY[]::varchar[] AS reloptions,
+    null AS relpartbound
     FROM rw_catalog.rw_relations
 ")]
 #[derive(Fields)]
@@ -46,4 +47,6 @@ struct PgClass {
     relam: i32,
     reltablespace: i32,
     reloptions: Vec<String>,
+    // PG uses pg_node_tree type but RW doesn't support it
+    relpartbound: Option<String>,
 }

src/frontend/src/catalog/system_catalog/pg_catalog/pg_constraint.rs

@@ -15,12 +15,17 @@
 use risingwave_common::types::Fields;
 use risingwave_frontend_macro::system_catalog;
 
+use crate::catalog::schema_catalog::SchemaCatalog;
+use crate::catalog::system_catalog::{SysCatalogReaderImpl, SystemTableCatalog};
+use crate::error::Result;
+use crate::TableCatalog;
+
 /// The catalog `pg_constraint` records information about table and index inheritance hierarchies.
 /// Ref: [`https://www.postgresql.org/docs/current/catalog-pg-constraint.html`]
 /// This is introduced only for pg compatibility and is not used in our system.
-#[system_catalog(view, "pg_catalog.pg_constraint")]
 #[derive(Fields)]
 struct PgConstraint {
+    #[primary_key]
     oid: i32,
     conname: String,
     connamespace: i32,
@@ -38,12 +43,105 @@ struct PgConstraint {
     conislocal: bool,
     coninhcount: i32,
     connoinherit: bool,
-    conkey: Vec<i16>,
-    confkey: Vec<i16>,
-    conpfeqop: Vec<i32>,
-    conppeqop: Vec<i32>,
-    conffeqop: Vec<i32>,
-    confdelsetcols: Vec<i16>,
-    conexclop: Vec<i32>,
-    conbin: String,
+    conkey: Option<Vec<i16>>,
+    confkey: Option<Vec<i16>>,
+    conpfeqop: Option<Vec<i32>>,
+    conppeqop: Option<Vec<i32>>,
+    conffeqop: Option<Vec<i32>>,
+    confdelsetcols: Option<Vec<i16>>,
+    conexclop: Option<Vec<i32>>,
+    conbin: Option<String>,
+}
+
+impl PgConstraint {
+    fn from_system_table(schema: &SchemaCatalog, table: &SystemTableCatalog) -> PgConstraint {
+        // List of the constrained columns. First column starts from 1.
+        let conkey: Vec<_> = table.pk.iter().map(|i| (*i + 1) as i16).collect();
+        PgConstraint {
+            oid: table.id.table_id() as i32, // Use table_id as a mock oid of constraint here.
+            conname: format!("{}_pkey", &table.name),
+            connamespace: schema.id() as i32,
+            contype: "p".to_owned(), // p = primary key constraint
+            condeferrable: false,
+            convalidated: true,
+            conrelid: table.id.table_id() as i32,
+            contypid: 0,
+            // Use table_id as a mock index oid of constraint here.
+            conindid: table.id.table_id() as i32,
+            conparentid: 0,
+            confrelid: 0,
+            confupdtype: " ".to_owned(),
+            confdeltype: " ".to_owned(),
+            confmatchtype: " ".to_owned(),
+            conislocal: true,
+            coninhcount: 0,
+            connoinherit: true,
+            conkey: Some(conkey),
+            confkey: None,
+            conpfeqop: None,
+            conppeqop: None,
+            conffeqop: None,
+            confdelsetcols: None,
+            conexclop: None,
+            conbin: None,
+        }
+    }
+
+    fn from_table(schema: &SchemaCatalog, table: &TableCatalog) -> PgConstraint {
+        // List of the constrained columns. First column starts from 1.
+        let conkey: Vec<_> = table
+            .pk
+            .iter()
+            .map(|i| (i.column_index + 1) as i16)
+            .collect();
+        PgConstraint {
+            oid: table.id.table_id() as i32, // Use table_id as a mock oid of constraint here.
+            conname: format!("{}_pkey", &table.name),
+            connamespace: schema.id() as i32,
+            contype: "p".to_owned(), // p = primary key constraint
+            condeferrable: false,
+            convalidated: true,
+            conrelid: table.id.table_id() as i32,
+            contypid: 0,
+            // Use table_id as a mock index oid of constraint here.
+            conindid: table.id.table_id() as i32,
+            conparentid: 0,
+            confrelid: 0,
+            confupdtype: " ".to_owned(),
+            confdeltype: " ".to_owned(),
+            confmatchtype: " ".to_owned(),
+            conislocal: true,
+            coninhcount: 0,
+            connoinherit: true,
+            conkey: Some(conkey),
+            confkey: None,
+            conpfeqop: None,
+            conppeqop: None,
+            conffeqop: None,
+            confdelsetcols: None,
+            conexclop: None,
+            conbin: None,
+        }
+    }
+}
+
+#[system_catalog(table, "pg_catalog.pg_constraint")]
+fn read_pg_constraint(reader: &SysCatalogReaderImpl) -> Result<Vec<PgConstraint>> {
+    let catalog_reader = reader.catalog_reader.read_guard();
+    let schemas = catalog_reader.iter_schemas(&reader.auth_context.database)?;
+
+    Ok(schemas.flat_map(read_pg_constraint_in_schema).collect())
+}
+
+fn read_pg_constraint_in_schema(schema: &SchemaCatalog) -> Vec<PgConstraint> {
+    // Note: We only support primary key constraints now.
+    let system_table_rows = schema
+        .iter_system_tables()
+        .map(|table| PgConstraint::from_system_table(schema, table.as_ref()));
+
+    let table_rows = schema
+        .iter_valid_table()
+        .map(|table| PgConstraint::from_table(schema, table.as_ref()));
+
+    system_table_rows.chain(table_rows).collect()
 }

src/frontend/src/catalog/system_catalog/pg_catalog/pg_language.rs

@@ -0,0 +1,34 @@
+// 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;
+
+/// The catalog `pg_language` registers languages in which you can write functions or stored procedures.
+/// Ref: [`https://www.postgresql.org/docs/current/catalog-pg-language.html`]
+/// This is introduced only for pg compatibility and is not used in our system.
+#[system_catalog(view, "pg_catalog.pg_language")]
+#[derive(Fields)]
+struct PgLanguage {
+    #[primary_key]
+    oid: i32,
+    lanname: String,
+    lanowner: i32,
+    lanispl: bool,
+    lanpltrusted: bool,
+    lanplcallfoid: i32,
+    laninline: i32,
+    lanvalidator: i32,
+    lanacl: Vec<String>,
+}

src/frontend/src/catalog/system_catalog/pg_catalog/pg_rewrite.rs

@@ -0,0 +1,33 @@
+// 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;
+
+/// The catalog `pg_rewrite` stores rewrite rules for tables and views.
+/// Ref: [`https://www.postgresql.org/docs/current/catalog-pg-rewrite.html`]
+/// This is introduced only for pg compatibility and is not used in our system.
+#[system_catalog(view, "pg_catalog.pg_rewrite")]
+#[derive(Fields)]
+struct PgRewrite {
+    #[primary_key]
+    oid: i32,
+    rulename: String,
+    ev_class: i32,
+    ev_type: String,
+    ev_enabled: String,
+    is_instead: bool,
+    ev_qual: String,
+    ev_action: String,
+}

src/frontend/src/catalog/system_catalog/pg_catalog/pg_trigger.rs

@@ -0,0 +1,44 @@
+// 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;
+
+/// The catalog `pg_trigger` stores triggers on tables and views.
+/// Ref: [`https://www.postgresql.org/docs/current/catalog-pg-trigger.html`]
+/// This is introduced only for pg compatibility and is not used in our system.
+#[system_catalog(view, "pg_catalog.pg_trigger")]
+#[derive(Fields)]
+struct PgTrigger {
+    #[primary_key]
+    oid: i32,
+    tgrelid: i32,
+    tgparentid: i32,
+    tgname: String,
+    tgfoid: i32,
+    tgtype: i16,
+    tgenabled: String,
+    tgisinternal: bool,
+    tgconstrrelid: i32,
+    tgconstrindid: i32,
+    tgconstraint: i32,
+    tgdeferrable: bool,
+    tginitdeferred: bool,
+    tgnargs: i16,
+    tgattr: Vec<i16>,
+    tgargs: Vec<u8>,
+    tgqual: Option<String>,
+    tgoldtable: Option<String>,
+    tgnewtable: Option<String>,
+}