feat(expr): support pg_get_indexdef (#14073)

e2e_test/batch/catalog/sysinfo.slt.part

@@ -12,3 +12,58 @@ query T
 select (SELECT pg_catalog.pg_get_userbyid(1));
 ----
 root
+
+statement ok
+create table tab(num int, name varchar);
+
+statement ok
+create index tab_idx on tab(num desc);
+
+query T
+select pg_get_indexdef('tab_idx'::regclass);
+----
+CREATE INDEX tab_idx ON tab(num DESC)
+
+query error Invalid parameter oid: index not found:
+select pg_get_indexdef('tab'::regclass);
+
+query error Invalid parameter name: class not found: tab_null
+select pg_get_indexdef('tab_null'::regclass);
+
+statement ok
+drop index tab_idx;
+
+statement ok
+drop table tab;
+
+statement ok
+create table tab(a int, b int, c int, d int);
+
+statement ok
+CREATE INDEX tab_idx ON tab (a, (b + c + (1 + 1))) include (d);
+
+query T
+select pg_get_indexdef('tab_idx'::regclass), pg_get_indexdef('tab_idx'::regclass, 0, true);
+----
+CREATE INDEX tab_idx ON tab(a, (b + c + (1 + 1))) INCLUDE(d) CREATE INDEX tab_idx ON tab(a, (b + c + (1 + 1))) INCLUDE(d)
+
+query T
+select pg_get_indexdef('tab_idx'::regclass, 1, true), pg_get_indexdef('tab_idx'::regclass, 2, true);
+----
+a ((b + c) + (1:Int32 + 1:Int32))
+
+query T
+select pg_get_indexdef('tab_idx'::regclass, 3, true);
+----
+d
+
+query T
+select pg_get_indexdef('tab_idx'::regclass, -1, true), pg_get_indexdef('tab_idx'::regclass, 4, true);
+----
+(empty) (empty)
+
+statement ok
+drop index tab_idx;
+
+statement ok
+drop table tab;

proto/expr.proto

@@ -269,9 +269,11 @@ message ExprNode {
     PG_SLEEP_FOR = 2025;
     PG_SLEEP_UNTIL = 2026;
 
-    // Adminitration functions
-    COL_DESCRIPTION = 2100;
-    CAST_REGCLASS = 2101;
+    // System administration functions
+    CAST_REGCLASS = 2100;
+    // System information functions
+    PG_GET_INDEXDEF = 2400;
+    COL_DESCRIPTION = 2401;
   }
   Type function_type = 1;
   data.DataType return_type = 3;

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

@@ -994,6 +994,7 @@ impl Binder {
                         ))))
                     }
                 ))),
+                ("pg_get_indexdef", raw_call(ExprType::PgGetIndexdef)),
                 ("pg_relation_size", dispatch_by_len(vec![
                     (1, raw(|binder, inputs|{
                         let table_name = &inputs[0];

src/frontend/src/catalog/index_catalog.rs

@@ -18,14 +18,14 @@ use std::sync::Arc;
 
 use educe::Educe;
 use itertools::Itertools;
-use risingwave_common::catalog::IndexId;
+use risingwave_common::catalog::{Field, IndexId, Schema};
 use risingwave_common::util::epoch::Epoch;
 use risingwave_common::util::sort_util::ColumnOrder;
 use risingwave_pb::catalog::{PbIndex, PbStreamJobStatus};
 
 use super::ColumnId;
 use crate::catalog::{DatabaseId, OwnedByUserCatalog, SchemaId, TableCatalog};
-use crate::expr::{Expr, ExprImpl, FunctionCall};
+use crate::expr::{Expr, ExprDisplay, ExprImpl, FunctionCall};
 use crate::user::UserId;
 
 #[derive(Clone, Debug, Educe)]
@@ -188,6 +188,30 @@ impl IndexCatalog {
         }
     }
 
+    pub fn get_column_def(&self, column_idx: usize) -> Option<String> {
+        if let Some(col) = self.index_table.columns.get(column_idx) {
+            if col.is_hidden {
+                return None;
+            }
+        } else {
+            return None;
+        }
+        let expr_display = ExprDisplay {
+            expr: &self.index_item[column_idx],
+            input_schema: &Schema::new(
+                self.primary_table
+                    .columns
+                    .iter()
+                    .map(|col| Field::from(&col.column_desc))
+                    .collect_vec(),
+            ),
+        };
+
+        // TODO(Kexiang): Currently, extra info like ":Int32" introduced by `ExprDisplay` is kept for simplity.
+        // We'd better remove it in the future.
+        Some(expr_display.to_string())
+    }
+
     pub fn display(&self) -> IndexDisplay {
         let index_table = self.index_table.clone();
         let index_columns_with_ordering = index_table

src/frontend/src/catalog/root_catalog.rs

@@ -685,6 +685,20 @@ impl Catalog {
             .ok_or_else(|| CatalogError::NotFound("index", index_name.to_string()))
     }
 
+    pub fn get_index_by_id(
+        &self,
+        db_name: &str,
+        index_id: u32,
+    ) -> CatalogResult<&Arc<IndexCatalog>> {
+        let index_id = IndexId::from(index_id);
+        for schema in self.get_database_by_name(db_name)?.iter_schemas() {
+            if let Some(index) = schema.get_index_by_id(&index_id) {
+                return Ok(index);
+            }
+        }
+        Err(CatalogError::NotFound("index", index_id.to_string()))
+    }
+
     pub fn get_view_by_name<'a>(
         &self,
         db_name: &str,

src/frontend/src/expr/function_impl/mod.rs

@@ -15,3 +15,4 @@
 mod cast_regclass;
 mod col_description;
 pub mod context;
+mod pg_get_indexdef;

src/frontend/src/expr/function_impl/pg_get_indexdef.rs

@@ -0,0 +1,69 @@
+// Copyright 2023 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 std::fmt::Write;
+
+use risingwave_expr::{capture_context, function, ExprError, Result};
+use thiserror_ext::AsReport;
+
+use super::context::{CATALOG_READER, DB_NAME};
+use crate::catalog::CatalogReader;
+
+#[function("pg_get_indexdef(int4) -> varchar")]
+fn pg_get_indexdef(oid: i32, writer: &mut impl Write) -> Result<()> {
+    pg_get_indexdef_impl_captured(oid, 0, writer)
+}
+
+#[function("pg_get_indexdef(int4, int4, boolean) -> varchar")]
+fn pg_get_indexdef_col(
+    oid: i32,
+    column_no: i32,
+    _pretty_bool: bool,
+    writer: &mut impl Write,
+) -> Result<()> {
+    pg_get_indexdef_impl_captured(oid, column_no, writer)
+}
+
+#[capture_context(CATALOG_READER, DB_NAME)]
+fn pg_get_indexdef_impl(
+    catalog: &CatalogReader,
+    db_name: &str,
+    oid: i32,
+    column_no: i32,
+    writer: &mut impl Write,
+) -> Result<()> {
+    let ans = if column_no == 0 {
+        catalog
+            .read_guard()
+            .get_index_by_id(db_name, oid as u32)
+            .map_err(|e| ExprError::InvalidParam {
+                name: "oid",
+                reason: e.to_report_string().into(),
+            })?
+            .index_table
+            .create_sql()
+    } else {
+        catalog
+            .read_guard()
+            .get_index_by_id(db_name, oid as u32)
+            .map_err(|e| ExprError::InvalidParam {
+                name: "oid",
+                reason: e.to_report_string().into(),
+            })?
+            .get_column_def(column_no as usize - 1)
+            .unwrap_or_default()
+    };
+    write!(writer, "{}", ans).unwrap();
+    Ok(())
+}

src/frontend/src/expr/pure.rs

@@ -239,8 +239,9 @@ impl ExprVisitor for ImpureAnalyzer {
             | expr_node::Type::PgSleep
             | expr_node::Type::PgSleepFor
             | expr_node::Type::PgSleepUntil
-            | expr_node::Type::ColDescription
             | expr_node::Type::CastRegclass
+            | expr_node::Type::PgGetIndexdef
+            | expr_node::Type::ColDescription
             | expr_node::Type::MakeTimestamptz => self.impure = true,
         }
     }