feat(expr): support pg_get_indexdef

e2e_test/batch/catalog/sysinfo.slt.part

@@ -12,3 +12,26 @@ 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;

proto/expr.proto

@@ -271,6 +271,8 @@ message ExprNode {
     // Adminitration functions
     COL_DESCRIPTION = 2100;
     CAST_REGCLASS = 2101;
+    // System information functions
+    PG_GET_INDEXDEF = 2400;
   }
   Type function_type = 1;
   data.DataType return_type = 3;

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

@@ -999,6 +999,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/root_catalog.rs

@@ -685,6 +685,20 @@ impl Catalog {
             .ok_or_else(|| CatalogError::NotFound("index", index_name.to_string()))
     }
 
+    pub fn get_table_by_index_id(
+        &self,
+        db_name: &str,
+        index_id: u32,
+    ) -> CatalogResult<&Arc<TableCatalog>> {
+        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.index_table);
+            }
+        }
+        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,49 @@
+// 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, writer)
+}
+
+#[capture_context(CATALOG_READER, DB_NAME)]
+fn pg_get_indexdef_impl(
+    catalog: &CatalogReader,
+    db_name: &str,
+    oid: i32,
+    writer: &mut impl Write,
+) -> Result<()> {
+    write!(
+        writer,
+        "{}",
+        catalog
+            .read_guard()
+            .get_table_by_index_id(db_name, oid as u32)
+            .map_err(|e| ExprError::InvalidParam {
+                name: "oid",
+                reason: e.to_report_string().into(),
+            })?
+            .create_sql()
+    )
+    .unwrap();
+    Ok(())
+}

src/frontend/src/expr/pure.rs

@@ -240,6 +240,7 @@ impl ExprVisitor for ImpureAnalyzer {
             | expr_node::Type::PgSleepUntil
             | expr_node::Type::ColDescription
             | expr_node::Type::CastRegclass
+            | expr_node::Type::PgGetIndexdef
             | expr_node::Type::MakeTimestamptz => self.impure = true,
         }
     }