feat(catalog): add pg_type.typinput and support dummy casting to regproc

e2e_test/batch/catalog/pg_type.slt.part

@@ -1,20 +1,20 @@
 query ITITT
-SELECT oid, typname, typelem, typnotnull, typtype FROM pg_catalog.pg_type order by oid;
+SELECT oid, typname, typelem, typnotnull, typtype, typinput FROM pg_catalog.pg_type order by oid;
 ----
-16	bool	0	f	b
-17	bytea	0	f	b
-20	int8	0	f	b
-21	int2	0	f	b
-23	int4	0	f	b
-25	text	0	f	b
-700	float4	0	f	b
-701	float8	0	f	b
-1043	varchar	0	f	b
-1082	date	0	f	b
-1083	time	0	f	b
-1114	timestamp	0	f	b
-1184	timestamptz	0	f	b
-1186	interval	0	f	b
-1301	rw_int256	0	f	b
-1700	numeric	0	f	b
-3802	jsonb	0	f	b
+16	bool	0	f	b boolin
+17	bytea	0	f	b byteain
+20	int8	0	f	b int8in
+21	int2	0	f	b int2in
+23	int4	0	f	b int4in
+25	text	0	f	b textin
+700	float4	0	f	b float4in
+701	float8	0	f	b float8in
+1043	varchar	0	f	b varcharin
+1082	date	0	f	b date_in
+1083	time	0	f	b time_in
+1114	timestamp	0	f	b timestamp_in
+1184	timestamptz	0	f	b timestamptz_in
+1186	interval	0	f	b interval_in
+1301	rw_int256	0	f	b rw_int256_in
+1700	numeric	0	f	b numeric_in
+3802	jsonb	0	f	b jsonb_in

src/common/src/types/postgres_type.rs

@@ -18,7 +18,7 @@ use crate::error::ErrorCode;
 /// `DataType` information extracted from PostgreSQL `pg_type`
 ///
 /// ```sql
-/// select oid, typarray, typname, typlen from pg_type
+/// select oid, typarray, typname, typinput, typlen from pg_type
 /// where oid in (16, 21, 23, 20, 1700, 700, 701, 1043, 17, 1082, 1114, 1184, 1083, 1186, 3802);
 /// ```
 ///
@@ -30,21 +30,21 @@ macro_rules! for_all_base_types {
     ($macro:ident $(, $x:tt)*) => {
         $macro! {
             $($x, )*
-            { Boolean     |   16 |     1000 | bool        |      1 }
-            { Bytea       |   17 |     1001 | bytea       |     -1 }
-            { Int64       |   20 |     1016 | int8        |      8 }
-            { Int16       |   21 |     1005 | int2        |      2 }
-            { Int32       |   23 |     1007 | int4        |      4 }
-            { Float32     |  700 |     1021 | float4      |      4 }
-            { Float64     |  701 |     1022 | float8      |      8 }
-            { Varchar     | 1043 |     1015 | varchar     |     -1 }
-            { Date        | 1082 |     1182 | date        |      4 }
-            { Time        | 1083 |     1183 | time        |      8 }
-            { Timestamp   | 1114 |     1115 | timestamp   |      8 }
-            { Timestamptz | 1184 |     1185 | timestamptz |      8 }
-            { Interval    | 1186 |     1187 | interval    |     16 }
-            { Decimal     | 1700 |     1231 | numeric     |     -1 }
-            { Jsonb       | 3802 |     3807 | jsonb       |     -1 }
+            { Boolean     |   16 |     1000 | bool        | boolin         |      1 }
+            { Bytea       |   17 |     1001 | bytea       | byteain        |     -1 }
+            { Int64       |   20 |     1016 | int8        | int8in         |      8 }
+            { Int16       |   21 |     1005 | int2        | int2in         |      2 }
+            { Int32       |   23 |     1007 | int4        | int4in         |      4 }
+            { Float32     |  700 |     1021 | float4      | float4in       |      4 }
+            { Float64     |  701 |     1022 | float8      | float8in       |      8 }
+            { Varchar     | 1043 |     1015 | varchar     | varcharin      |     -1 }
+            { Date        | 1082 |     1182 | date        | date_in        |      4 }
+            { Time        | 1083 |     1183 | time        | time_in        |      8 }
+            { Timestamp   | 1114 |     1115 | timestamp   | timestamp_in   |      8 }
+            { Timestamptz | 1184 |     1185 | timestamptz | timestamptz_in |      8 }
+            { Interval    | 1186 |     1187 | interval    | interval_in    |     16 }
+            { Decimal     | 1700 |     1231 | numeric     | numeric_in     |     -1 }
+            { Jsonb       | 3802 |     3807 | jsonb       | jsonb_in       |     -1 }
         }
     };
 }
@@ -53,7 +53,7 @@ macro_rules! for_all_base_types {
 impl DataType {
     pub fn type_len(&self) -> i16 {
         macro_rules! impl_type_len {
-            ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $len:literal } )*) => {
+            ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $input:ident | $len:literal } )*) => {
                 match self {
                     $(
                     DataType::$enum => $len,
@@ -75,7 +75,7 @@ impl DataType {
     //  For Numeric(aka Decimal): oid = 1700, array_type_oid = 1231
     pub fn from_oid(oid: i32) -> crate::error::Result<Self> {
         macro_rules! impl_from_oid {
-            ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $len:literal } )*) => {
+            ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $input:ident | $len:literal } )*) => {
                 match oid {
                     $(
                     $oid => Ok(DataType::$enum),
@@ -95,7 +95,7 @@ impl DataType {
 
     pub fn to_oid(&self) -> i32 {
         macro_rules! impl_to_oid {
-            ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $len:literal } )*) => {
+            ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $input:ident | $len:literal } )*) => {
                 match self {
                     $(
                     DataType::$enum => $oid,
@@ -121,7 +121,7 @@ impl DataType {
 
     pub fn pg_name(&self) -> &'static str {
         macro_rules! impl_pg_name {
-            ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $len:literal } )*) => {
+            ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $input:ident | $len:literal } )*) => {
                 match self {
                     $(
                     DataType::$enum => stringify!($name),

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

@@ -23,3 +23,8 @@
   expected_outputs:
   - batch_plan
   - logical_plan
+- sql: |
+    select 'boolin'::regproc
+  expected_outputs:
+  - logical_plan
+  - batch_plan

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

@@ -2,16 +2,16 @@
 - sql: |
     select * from pg_catalog.pg_type
   logical_plan: |-
-    LogicalProject { exprs: [rw_types.id, rw_types.name, 0:Int32, 0:Int32, false:Boolean, 0:Int32, -1:Int32, 0:Int32, 0:Int32, rw_schemas.id, 'b':Varchar, 0:Int32, null:Varchar, null:Varchar, null:Int32] }
+    LogicalProject { exprs: [rw_types.id, rw_types.name, 0:Int32, 0:Int32, rw_types.input_oid, false:Boolean, 0:Int32, -1:Int32, 0:Int32, 0:Int32, rw_schemas.id, 'b':Varchar, 0:Int32, null:Varchar, null:Varchar, null:Int32] }
     └─LogicalShare { id: 3 }
-      └─LogicalProject { exprs: [rw_types.id, rw_types.name, 0:Int32, 0:Int32, false:Boolean, 0:Int32, -1:Int32, 0:Int32, 0:Int32, rw_schemas.id, 'b':Varchar, 0:Int32, null:Varchar, null:Varchar, null:Int32] }
+      └─LogicalProject { exprs: [rw_types.id, rw_types.name, 0:Int32, 0:Int32, rw_types.input_oid, false:Boolean, 0:Int32, -1:Int32, 0:Int32, 0:Int32, rw_schemas.id, 'b':Varchar, 0:Int32, null:Varchar, null:Varchar, null:Int32] }
         └─LogicalJoin { type: Inner, on: (rw_schemas.name = 'pg_catalog':Varchar), output: all }
-          ├─LogicalScan { table: rw_types, columns: [rw_types.id, rw_types.name] }
+          ├─LogicalScan { table: rw_types, columns: [rw_types.id, rw_types.name, rw_types.input_oid] }
           └─LogicalScan { table: rw_schemas, columns: [rw_schemas.id, rw_schemas.name, rw_schemas.owner, rw_schemas.acl] }
   batch_plan: |-
-    BatchProject { exprs: [rw_types.id, rw_types.name, 0:Int32, 0:Int32, false:Boolean, 0:Int32, -1:Int32, 0:Int32, 0:Int32, rw_schemas.id, 'b':Varchar, 0:Int32, null:Varchar, null:Varchar, null:Int32] }
+    BatchProject { exprs: [rw_types.id, rw_types.name, 0:Int32, 0:Int32, rw_types.input_oid, false:Boolean, 0:Int32, -1:Int32, 0:Int32, 0:Int32, rw_schemas.id, 'b':Varchar, 0:Int32, null:Varchar, null:Varchar, null:Int32] }
     └─BatchNestedLoopJoin { type: Inner, predicate: true, output: all }
-      ├─BatchScan { table: rw_types, columns: [rw_types.id, rw_types.name], distribution: Single }
+      ├─BatchScan { table: rw_types, columns: [rw_types.id, rw_types.name, rw_types.input_oid], distribution: Single }
       └─BatchProject { exprs: [rw_schemas.id] }
         └─BatchFilter { predicate: (rw_schemas.name = 'pg_catalog':Varchar) }
           └─BatchScan { table: rw_schemas, columns: [rw_schemas.id, rw_schemas.name], distribution: Single }
@@ -224,3 +224,9 @@
     LogicalProject { exprs: [2:Int32] }
     └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
   batch_plan: 'BatchValues { rows: [[2:Int32]] }'
+- sql: |
+    select 'boolin'::regproc
+  logical_plan: |-
+    LogicalProject { exprs: ['boolin':Varchar] }
+    └─LogicalValues { rows: [[]], schema: Schema { fields: [] } }
+  batch_plan: 'BatchValues { rows: [[''boolin'':Varchar]] }'

src/frontend/src/binder/expr/mod.rs

@@ -550,6 +550,18 @@ impl Binder {
                 self.resolve_regclass(class_name)
                     .map(|id| ExprImpl::literal_int(id as i32))
             }
+            AstDataType::Regproc => {
+                let lhs = self.bind_expr_inner(expr)?;
+                let lhs_ty = lhs.return_type();
+                if lhs_ty == DataType::Varchar {
+                    // FIXME: Currently, we only allow VARCHAR to be casted to Regproc.
+                    // FIXME: Check whether it's a valid proc
+                    // FIXME: The return type should be casted to Regproc, but we don't have this type.
+                    Ok(lhs)
+                } else {
+                    Err(ErrorCode::BindError(format!("Can't cast {} to regproc", lhs_ty)).into())
+                }
+            }
             _ => self.bind_cast_inner(expr, bind_data_type(&data_type)?),
         }
     }
@@ -655,6 +667,7 @@ pub fn bind_data_type(data_type: &AstDataType) -> Result<DataType> {
         }
         AstDataType::Bytea => DataType::Bytea,
         AstDataType::Regclass
+        | AstDataType::Regproc
         | AstDataType::Uuid
         | AstDataType::Custom(_)
         | AstDataType::Decimal(_, _)

src/frontend/src/binder/select.rs

@@ -852,6 +852,7 @@ fn data_type_to_alias(data_type: &AstDataType) -> Option<String> {
         }
         AstDataType::Interval => "interval".to_string(),
         AstDataType::Regclass => "regclass".to_string(),
+        AstDataType::Regproc => "regproc".to_string(),
         AstDataType::Text => "text".to_string(),
         AstDataType::Bytea => "bytea".to_string(),
         AstDataType::Array(ty) => return data_type_to_alias(ty),

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

@@ -31,6 +31,8 @@ pub static PG_TYPE: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
         (DataType::Int32, "typelem"),
         // 0
         (DataType::Int32, "typarray"),
+        // FIXME: Should be regproc type
+        (DataType::Varchar, "typinput"),
         // false
         (DataType::Boolean, "typnotnull"),
         // 0
@@ -58,6 +60,7 @@ pub static PG_TYPE: LazyLock<BuiltinView> = LazyLock::new(|| BuiltinView {
                 t.name AS typname, \
                 0 AS typelem, \
                 0 AS typarray, \
+                t.input_oid AS typinput, \
                 false AS typnotnull, \
                 0 AS typbasetype, \
                 -1 AS typtypmod, \

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

@@ -24,36 +24,41 @@ use risingwave_common::types::{DataType, ScalarImpl};
 use crate::catalog::system_catalog::{BuiltinTable, SysCatalogReaderImpl};
 
 macro_rules! impl_pg_type_data {
-    ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $len:literal } )*) => {
+    ($( { $enum:ident | $oid:literal | $oid_array:literal | $name:ident | $input:ident | $len:literal } )*) => {
         &[
             $(
-            ($oid, stringify!($name)),
+            ($oid, stringify!($name), stringify!($input)),
             )*
             // Note: rw doesn't support `text` type, returning it is just a workaround to be compatible
             // with PostgreSQL.
-            (25, "text"),
-            (1301, "rw_int256"),
+            (25, "text", "textin"),
+            (1301, "rw_int256", "rw_int256_in"),
         ]
     }
 }
-pub const RW_TYPE_DATA: &[(i32, &str)] = for_all_base_types! { impl_pg_type_data };
+pub const RW_TYPE_DATA: &[(i32, &str, &str)] = for_all_base_types! { impl_pg_type_data };
 
 /// `rw_types` stores all supported types in the database.
 pub static RW_TYPES: LazyLock<BuiltinTable> = LazyLock::new(|| BuiltinTable {
     name: "rw_types",
     schema: RW_CATALOG_SCHEMA_NAME,
-    columns: &[(DataType::Int32, "id"), (DataType::Varchar, "name")],
+    columns: &[
+        (DataType::Int32, "id"),
+        (DataType::Varchar, "name"),
+        (DataType::Varchar, "input_oid"),
+    ],
     pk: &[0],
 });
 
 impl SysCatalogReaderImpl {
     pub fn read_rw_types(&self) -> Result<Vec<OwnedRow>> {
         Ok(RW_TYPE_DATA
             .iter()
-            .map(|(id, name)| {
+            .map(|(id, name, input)| {
                 OwnedRow::new(vec![
                     Some(ScalarImpl::Int32(*id)),
                     Some(ScalarImpl::Utf8(name.to_string().into())),
+                    Some(ScalarImpl::Utf8(input.to_string().into())),
                 ])
             })
             .collect_vec())

src/meta/src/manager/env.rs

@@ -25,9 +25,9 @@ use crate::manager::{
     NotificationManagerRef,
 };
 use crate::model::ClusterId;
+use crate::storage::MetaStoreRef;
 #[cfg(any(test, feature = "test"))]
-use crate::storage::MemStore;
-use crate::storage::{MetaStoreBoxExt, MetaStoreRef};
+use crate::storage::{MemStore, MetaStoreBoxExt};
 use crate::MetaResult;
 
 /// [`MetaSrvEnv`] is the global environment in Meta service. The instance will be shared by all

src/sqlparser/src/ast/data_type.rs

@@ -56,6 +56,8 @@ pub enum DataType {
     Interval,
     /// Regclass used in postgresql serial
     Regclass,
+    /// Regproc used in postgresql function
+    Regproc,
     /// Text
     Text,
     /// Bytea
@@ -97,6 +99,7 @@ impl fmt::Display for DataType {
             }
             DataType::Interval => write!(f, "INTERVAL"),
             DataType::Regclass => write!(f, "REGCLASS"),
+            DataType::Regproc => write!(f, "REGPROC"),
             DataType::Text => write!(f, "TEXT"),
             DataType::Bytea => write!(f, "BYTEA"),
             DataType::Array(ty) => write!(f, "{}[]", ty),

src/sqlparser/src/keywords.rs

@@ -395,6 +395,7 @@ define_keywords!(
     REFERENCING,
     REGCLASS,
     REGISTRY,
+    REGPROC,
     REGR_AVGX,
     REGR_AVGY,
     REGR_COUNT,

src/sqlparser/src/parser.rs

@@ -3285,6 +3285,7 @@ impl Parser {
                 // parse_interval_literal for a taste.
                 Keyword::INTERVAL => Ok(DataType::Interval),
                 Keyword::REGCLASS => Ok(DataType::Regclass),
+                Keyword::REGPROC => Ok(DataType::Regproc),
                 Keyword::TEXT => {
                     if self.consume_token(&Token::LBracket) {
                         // Note: this is postgresql-specific