feat(frontend): support get_indexes in sqlalchemy (risingwavelabs#6636

)

* add indkey in pg_index

* add pg_attribute

* fix name of pg_matviews

* fix dashboard

* fix e2e

* remove not related dashboard file

* fix a slip and add e2e

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

e2e_test/batch/catalog/pg_attribute.slt.part

@@ -0,0 +1,27 @@
+statement ok
+create table tmp(id1 int, id2 int);
+
+query TIII
+select a.attname, a.atttypid, a.attlen, a.attnum from pg_catalog.pg_class t
+    join pg_catalog.pg_attribute a on t.oid = a.attrelid 
+    where t.relname = 'tmp' order by a.attnum;
+----
+id1 23 4 1
+id2 23 4 2
+
+statement ok
+create view view1 as select id2 from tmp;
+
+query TIII
+select a.attname, a.atttypid, a.attlen, a.attnum from pg_catalog.pg_class t
+    join pg_catalog.pg_attribute a on t.oid = a.attrelid 
+    where t.relname = 'view1';
+----
+id2 23 4 1
+
+
+statement ok
+drop view view1;
+
+statement ok
+drop table tmp;

e2e_test/batch/catalog/pg_class.slt.part

@@ -4,7 +4,7 @@ SELECT oid,relname,relowner,relkind FROM pg_catalog.pg_class ORDER BY oid limit
 1 pg_type 1 r
 2 pg_namespace 1 r
 3 pg_cast 1 r
-4 pg_matviews_info 1 r
+4 pg_matviews 1 r
 5 pg_user 1 r
 6 pg_class 1 r
 7 pg_index 1 r

e2e_test/batch/catalog/pg_index.slt.part

@@ -0,0 +1,38 @@
+statement ok
+create table tmp(id1 int, id2 int);
+
+statement ok
+create index tmp_id2_idx on tmp(id2);
+
+query IT
+select ix.indnatts, ix.indkey from pg_catalog.pg_class t 
+    join pg_catalog.pg_index ix on t.oid = ix.indrelid
+    join pg_catalog.pg_class i on i.oid = ix.indexrelid
+    where t.relname = 'tmp' and i.relname = 'tmp_id2_idx';
+----
+1 {2}
+
+statement ok
+create index tmp_id2_idx_include_id1 on tmp(id2) include(id1);
+
+query IT
+select ix.indnatts, ix.indkey from pg_catalog.pg_class t 
+    join pg_catalog.pg_index ix on t.oid = ix.indrelid
+    join pg_catalog.pg_class i on i.oid = ix.indexrelid
+    where t.relname = 'tmp' and i.relname = 'tmp_id2_idx_include_id1';
+----
+2 {2,1}
+
+statement ok
+create index tmp_id1_id2_idx on tmp(id1, id2);
+
+query IT
+select ix.indnatts, ix.indkey from pg_catalog.pg_class t 
+    join pg_catalog.pg_index ix on t.oid = ix.indrelid
+    join pg_catalog.pg_class i on i.oid = ix.indexrelid
+    where t.relname = 'tmp' and i.relname = 'tmp_id1_id2_idx';
+----
+2 {1,2}
+
+statement ok
+drop table tmp;

e2e_test/ddl/show.slt

@@ -79,21 +79,21 @@ statement ok
 drop table t3;
 
 query TT
-describe pg_matviews_info;
+describe pg_matviews;
 ----
-matviewid Int32
+schemaname Varchar
 matviewname Varchar
-matviewschema Varchar
 matviewowner Int32
-matviewgraph Varchar
 definition Varchar
+matviewid Int32
+matviewgraph Varchar
 
 query TT
-show columns from pg_catalog.pg_matviews_info;
+show columns from pg_catalog.pg_matviews;
 ----
-matviewid Int32
+schemaname Varchar
 matviewname Varchar
-matviewschema Varchar
 matviewowner Int32
-matviewgraph Varchar
 definition Varchar
+matviewid Int32
+matviewgraph Varchar

proto/catalog.proto

@@ -75,6 +75,7 @@ message Index {
   // Only `InputRef` type index is supported Now.
   // The index of `InputRef` is the column index of the primary table.
   repeated expr.ExprNode index_item = 8;
+  repeated int32 original_columns = 9;
 }
 
 // See `TableCatalog` struct in frontend crate for more information.

src/frontend/src/catalog/index_catalog.rs

@@ -21,6 +21,7 @@ use risingwave_common::types::DataType;
 use risingwave_pb::catalog::Index as ProstIndex;
 use risingwave_pb::expr::expr_node::RexNode;
 
+use super::ColumnId;
 use crate::catalog::{DatabaseId, SchemaId, TableCatalog};
 use crate::expr::{Expr, InputRef};
 use crate::optimizer::property::FieldOrder;
@@ -43,6 +44,8 @@ pub struct IndexCatalog {
     pub primary_to_secondary_mapping: HashMap<usize, usize>,
 
     pub secondary_to_primary_mapping: HashMap<usize, usize>,
+
+    pub original_columns: Vec<ColumnId>,
 }
 
 impl IndexCatalog {
@@ -76,6 +79,13 @@ impl IndexCatalog {
             .map(|(i, input_ref)| (i, input_ref.index))
             .collect();
 
+        let original_columns = index_prost
+            .original_columns
+            .clone()
+            .into_iter()
+            .map(Into::into)
+            .collect();
+
         IndexCatalog {
             id: index_prost.id.into(),
             name: index_prost.name.clone(),
@@ -84,6 +94,7 @@ impl IndexCatalog {
             primary_table: Arc::new(primary_table.clone()),
             primary_to_secondary_mapping,
             secondary_to_primary_mapping,
+            original_columns,
         }
     }
 
@@ -138,6 +149,7 @@ impl IndexCatalog {
                 .iter()
                 .map(InputRef::to_expr_proto)
                 .collect_vec(),
+            original_columns: self.original_columns.iter().map(Into::into).collect_vec(),
         }
     }
 }

src/frontend/src/catalog/schema_catalog.rs

@@ -194,6 +194,12 @@ impl SchemaCatalog {
             .map(|(_, v)| v)
     }
 
+    pub fn iter_valid_table(&self) -> impl Iterator<Item = &Arc<TableCatalog>> {
+        self.table_by_name
+            .iter()
+            .filter_map(|(key, v)| valid_table_name(key).then_some(v))
+    }
+
     /// Iterate all materialized views, excluding the indices.
     pub fn iter_mv(&self) -> impl Iterator<Item = &Arc<TableCatalog>> {
         self.table_by_name

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

@@ -181,7 +181,7 @@ prepare_sys_catalog! {
     { PG_CATALOG, PG_TYPE, vec![0], read_types },
     { PG_CATALOG, PG_NAMESPACE, vec![0], read_namespace },
     { PG_CATALOG, PG_CAST, vec![0], read_cast },
-    { PG_CATALOG, PG_MATVIEWS_INFO, vec![0], read_mviews_info await },
+    { PG_CATALOG, PG_MATVIEWS, vec![0], read_mviews_info await },
     { PG_CATALOG, PG_USER, vec![0], read_user_info },
     { PG_CATALOG, PG_CLASS, vec![0], read_class_info },
     { PG_CATALOG, PG_INDEX, vec![0], read_index_info },
@@ -190,6 +190,7 @@ prepare_sys_catalog! {
     { PG_CATALOG, PG_AM, vec![0], read_am_info },
     { PG_CATALOG, PG_OPERATOR, vec![0], read_operator_info },
     { PG_CATALOG, PG_VIEWS, vec![], read_views_info },
+    { PG_CATALOG, PG_ATTRIBUTE, vec![0, 4], read_pg_attribute },
     { INFORMATION_SCHEMA, COLUMNS, vec![], read_columns_info },
     { INFORMATION_SCHEMA, TABLES, vec![], read_tables_info },
 }

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

@@ -13,11 +13,12 @@
 // limitations under the License.
 
 pub mod pg_am;
+pub mod pg_attribute;
 pub mod pg_cast;
 pub mod pg_class;
 pub mod pg_collation;
 pub mod pg_index;
-pub mod pg_matviews_info;
+pub mod pg_matviews;
 pub mod pg_namespace;
 pub mod pg_opclass;
 pub mod pg_operator;
@@ -29,17 +30,19 @@ use std::collections::HashMap;
 
 use itertools::Itertools;
 pub use pg_am::*;
+pub use pg_attribute::*;
 pub use pg_cast::*;
 pub use pg_class::*;
 pub use pg_collation::*;
 pub use pg_index::*;
-pub use pg_matviews_info::*;
+pub use pg_matviews::*;
 pub use pg_namespace::*;
 pub use pg_opclass::*;
 pub use pg_operator::*;
 pub use pg_type::*;
 pub use pg_user::*;
 pub use pg_views::*;
+use risingwave_common::array::ListValue;
 use risingwave_common::error::Result;
 use risingwave_common::row::Row;
 use risingwave_common::types::ScalarImpl;
@@ -297,7 +300,14 @@ impl SysCatalogReaderImpl {
                     Row::new(vec![
                         Some(ScalarImpl::Int32(index.id.index_id() as i32)),
                         Some(ScalarImpl::Int32(index.primary_table.id.table_id() as i32)),
-                        Some(ScalarImpl::Int16(index.index_item.len() as i16)),
+                        Some(ScalarImpl::Int16(index.original_columns.len() as i16)),
+                        Some(ScalarImpl::List(ListValue::new(
+                            index
+                                .original_columns
+                                .iter()
+                                .map(|index| Some(ScalarImpl::Int16(index.get_id() as i16 + 1)))
+                                .collect_vec(),
+                        ))),
                     ])
                 })
             })
@@ -330,12 +340,12 @@ impl SysCatalogReaderImpl {
                 .for_each(|t| {
                     if let Some(fragments) = table_fragments.get(&t.id.table_id) {
                         rows.push(Row::new(vec![
-                            Some(ScalarImpl::Int32(t.id.table_id as i32)),
-                            Some(ScalarImpl::Utf8(t.name.clone())),
                             Some(ScalarImpl::Utf8(schema.clone())),
+                            Some(ScalarImpl::Utf8(t.name.clone())),
                             Some(ScalarImpl::Int32(t.owner as i32)),
-                            Some(ScalarImpl::Utf8(json!(fragments).to_string())),
                             Some(ScalarImpl::Utf8(t.definition.clone())),
+                            Some(ScalarImpl::Int32(t.id.table_id as i32)),
+                            Some(ScalarImpl::Utf8(json!(fragments).to_string())),
                         ]));
                     }
                 });
@@ -368,4 +378,47 @@ impl SysCatalogReaderImpl {
             })
             .collect_vec())
     }
+
+    pub(super) fn read_pg_attribute(&self) -> Result<Vec<Row>> {
+        let reader = self.catalog_reader.read_guard();
+        let schemas = reader.iter_schemas(&self.auth_context.database)?;
+
+        Ok(schemas
+            .flat_map(|schema| {
+                let view_rows = schema.iter_view().flat_map(|view| {
+                    view.columns.iter().enumerate().map(|(index, column)| {
+                        Row::new(vec![
+                            Some(ScalarImpl::Int32(view.id as i32)),
+                            Some(ScalarImpl::Utf8(column.name.clone())),
+                            Some(ScalarImpl::Int32(column.data_type().to_oid())),
+                            Some(ScalarImpl::Int16(column.data_type().type_len())),
+                            Some(ScalarImpl::Int16(index as i16 + 1)),
+                            Some(ScalarImpl::Bool(false)),
+                        ])
+                    })
+                });
+
+                schema
+                    .iter_valid_table()
+                    .flat_map(|table| {
+                        table
+                            .columns()
+                            .iter()
+                            .enumerate()
+                            .filter(|(_, column)| !column.is_hidden())
+                            .map(|(index, column)| {
+                                Row::new(vec![
+                                    Some(ScalarImpl::Int32(table.id.table_id() as i32)),
+                                    Some(ScalarImpl::Utf8(column.name().to_string())),
+                                    Some(ScalarImpl::Int32(column.data_type().to_oid())),
+                                    Some(ScalarImpl::Int16(column.data_type().type_len())),
+                                    Some(ScalarImpl::Int16(index as i16 + 1)),
+                                    Some(ScalarImpl::Bool(false)),
+                                ])
+                            })
+                    })
+                    .chain(view_rows)
+            })
+            .collect_vec())
+    }
 }

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

@@ -0,0 +1,34 @@
+// Copyright 2022 Singularity Data
+//
+// 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::DataType;
+
+use crate::catalog::system_catalog::SystemCatalogColumnsDef;
+
+/// The catalog `pg_attribute` stores information about table columns. There will be exactly one
+/// `pg_attribute` row for every column in every table in the database. (There will also be
+/// attribute entries for indexes, and indeed all objects that have `pg_class` entries.)
+/// Ref: [`https://www.postgresql.org/docs/current/catalog-pg-attribute.html`]
+///
+/// In RisingWave, we simply make it contain the columns of the view and all the columns of the
+/// tables that are not internal tables.
+pub const PG_ATTRIBUTE_TABLE_NAME: &str = "pg_attribute";
+pub const PG_ATTRIBUTE_COLUMNS: &[SystemCatalogColumnsDef<'_>] = &[
+    (DataType::Int32, "attrelid"),
+    (DataType::Varchar, "attname"),
+    (DataType::Int32, "atttypid"),
+    (DataType::Int16, "attlen"),
+    (DataType::Int16, "attnum"),
+    (DataType::Boolean, "attnotnull"),
+];