feat(catalog): support column indnkeyatts in pg_index (#16339)

e2e_test/batch/catalog/pg_attribute.slt.part

@@ -38,7 +38,9 @@ select i.relname, a.attname, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_attribute a on t.oid = a.attrelid and a.attnum = ANY(ix.indkey)
     where t.relname = 'tmp' order by a.attnum;
 ----
-tmp_idx id2 {2}
+tmp_idx	id1	{2,1,3,5}
+tmp_idx	id2	{2,1,3,5}
+tmp_idx	id3	{2,1,3,5}
 
 statement ok
 drop table tmp;

e2e_test/batch/catalog/pg_index.slt.part

@@ -5,34 +5,34 @@ statement ok
 create index tmp_id2_idx on tmp(id2) include(id2);
 
 query IT
-select ix.indnatts, ix.indkey from pg_catalog.pg_class t
+select ix.indnatts, ix.indnkeyatts, 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}
+2	1	{2,3}
 
 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
+select ix.indnatts, ix.indnkeyatts, 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';
 ----
-1 {2}
+3	1	{2,1,3}
 
 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
+select ix.indnatts, ix.indnkeyatts, 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}
+3	2	{1,2,3}
 
 statement ok
 drop table tmp;

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

@@ -243,7 +243,7 @@
         │       │ │ │ │ │ └─LogicalProject { exprs: [rw_sources.id, rw_sources.name, 'source':Varchar, rw_sources.schema_id, rw_sources.owner, rw_sources.definition, rw_sources.acl] }
         │       │ │ │ │ │   └─LogicalSysScan { table: rw_sources, columns: [rw_sources.id, rw_sources.name, rw_sources.schema_id, rw_sources.owner, rw_sources.connector, rw_sources.columns, rw_sources.format, rw_sources.row_encode, rw_sources.append_only, rw_sources.connection_id, rw_sources.definition, rw_sources.acl, rw_sources.initialized_at, rw_sources.created_at, rw_sources.initialized_at_cluster_version, rw_sources.created_at_cluster_version] }
         │       │ │ │ │ └─LogicalProject { exprs: [rw_indexes.id, rw_indexes.name, 'index':Varchar, rw_indexes.schema_id, rw_indexes.owner, rw_indexes.definition, rw_indexes.acl] }
-        │       │ │ │ │   └─LogicalSysScan { table: rw_indexes, columns: [rw_indexes.id, rw_indexes.name, rw_indexes.primary_table_id, rw_indexes.indkey, rw_indexes.schema_id, rw_indexes.owner, rw_indexes.definition, rw_indexes.acl, rw_indexes.initialized_at, rw_indexes.created_at, rw_indexes.initialized_at_cluster_version, rw_indexes.created_at_cluster_version] }
+        │       │ │ │ │   └─LogicalSysScan { table: rw_indexes, columns: [rw_indexes.id, rw_indexes.name, rw_indexes.primary_table_id, rw_indexes.key_columns, rw_indexes.include_columns, rw_indexes.schema_id, rw_indexes.owner, rw_indexes.definition, rw_indexes.acl, rw_indexes.initialized_at, rw_indexes.created_at, rw_indexes.initialized_at_cluster_version, rw_indexes.created_at_cluster_version] }
         │       │ │ │ └─LogicalProject { exprs: [rw_sinks.id, rw_sinks.name, 'sink':Varchar, rw_sinks.schema_id, rw_sinks.owner, rw_sinks.definition, rw_sinks.acl] }
         │       │ │ │   └─LogicalSysScan { table: rw_sinks, columns: [rw_sinks.id, rw_sinks.name, rw_sinks.schema_id, rw_sinks.owner, rw_sinks.connector, rw_sinks.sink_type, rw_sinks.connection_id, rw_sinks.definition, rw_sinks.acl, rw_sinks.initialized_at, rw_sinks.created_at, rw_sinks.initialized_at_cluster_version, rw_sinks.created_at_cluster_version] }
         │       │ │ └─LogicalProject { exprs: [rw_subscriptions.id, rw_subscriptions.name, 'subscription':Varchar, rw_subscriptions.schema_id, rw_subscriptions.owner, rw_subscriptions.definition, rw_subscriptions.acl] }

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

@@ -22,21 +22,31 @@ use risingwave_frontend_macro::system_catalog;
     "pg_catalog.pg_index",
     "SELECT id AS indexrelid,
         primary_table_id AS indrelid,
-        ARRAY_LENGTH(indkey)::smallint AS indnatts,
+        ARRAY_LENGTH(key_columns || include_columns)::smallint AS indnatts,
+        ARRAY_LENGTH(key_columns)::smallint AS indnkeyatts,
         false AS indisunique,
-        indkey,
+        key_columns || include_columns AS indkey,
         ARRAY[]::smallint[] as indoption,
         NULL AS indexprs,
         NULL AS indpred,
         FALSE AS indisprimary,
-        ARRAY[]::int[] AS indclass
+        ARRAY[]::int[] AS indclass,
+        false AS indisexclusion,
+        true AS indimmediate,
+        false AS indisclustered,
+        true AS indisvalid,
+        false AS indcheckxmin,
+        true AS indisready,
+        true AS indislive,
+        false AS indisreplident
     FROM rw_catalog.rw_indexes"
 )]
 #[derive(Fields)]
 struct PgIndex {
     indexrelid: i32,
     indrelid: i32,
     indnatts: i16,
+    indnkeyatts: i16,
     // We return false as default to indicate that this is NOT a unique index
     indisunique: bool,
     indkey: Vec<i16>,
@@ -49,4 +59,14 @@ struct PgIndex {
     indisprimary: bool,
     // Empty array. We only have a dummy implementation of `pg_opclass` yet.
     indclass: Vec<i32>,
+
+    // Unused columns. Kept for compatibility with PG.
+    indisexclusion: bool,
+    indimmediate: bool,
+    indisclustered: bool,
+    indisvalid: bool,
+    indcheckxmin: bool,
+    indisready: bool,
+    indislive: bool,
+    indisreplident: bool,
 }

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

@@ -24,7 +24,8 @@ struct RwIndex {
     id: i32,
     name: String,
     primary_table_id: i32,
-    indkey: Vec<i16>,
+    key_columns: Vec<i16>,
+    include_columns: Vec<i16>,
     schema_id: i32,
     owner: i32,
     definition: String,
@@ -46,7 +47,7 @@ fn read_rw_indexes(reader: &SysCatalogReaderImpl) -> Result<Vec<RwIndex>> {
                 id: index.id.index_id as i32,
                 name: index.name.clone(),
                 primary_table_id: index.primary_table.id().table_id as i32,
-                indkey: index
+                key_columns: index
                     .index_item
                     .iter()
                     .take(index.index_columns_len as usize)
@@ -59,6 +60,19 @@ fn read_rw_indexes(reader: &SysCatalogReaderImpl) -> Result<Vec<RwIndex>> {
                         ind as i16
                     })
                     .collect(),
+                include_columns: index
+                    .index_item
+                    .iter()
+                    .skip(index.index_columns_len as usize)
+                    .map(|index| {
+                        let ind = if let Some(input_ref) = index.as_input_ref() {
+                            input_ref.index() + 1
+                        } else {
+                            0
+                        };
+                        ind as i16
+                    })
+                    .collect(),
                 schema_id: schema.id() as i32,
                 owner: index.index_table.owner as i32,
                 definition: index.index_table.create_sql(),