feat(frontend): support functional indexes creation

e2e_test/batch/catalog/pg_index.slt.part

@@ -10,7 +10,7 @@ select ix.indnatts, ix.indkey from pg_catalog.pg_class t
     join pg_catalog.pg_class i on i.oid = ix.indexrelid
     where t.relname = 'tmp' and i.relname = 'tmp_id2_idx';
 ----
-1 {2}
+2 {2,3}
 
 statement ok
 create index tmp_id2_idx_include_id1 on tmp(id2) include(id1);
@@ -21,7 +21,7 @@ select ix.indnatts, ix.indkey from pg_catalog.pg_class t
     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}
+3 {2,3,4}
 
 statement ok
 create index tmp_id1_id2_idx on tmp(id1, id2);
@@ -32,7 +32,7 @@ select ix.indnatts, ix.indkey from pg_catalog.pg_class t
     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,3,4}
 
 statement ok
 drop table tmp;

e2e_test/ddl/alter_table_column.slt

@@ -215,7 +215,7 @@ statement ok
 create table t(id int primary key, a int, b varchar);
 
 statement ok
-create index idx on t(a);
+create index idx on t(a, lower(b));
 
 statement ok
 alter table t add column c int;

src/frontend/src/handler/create_index.rs

@@ -228,7 +228,7 @@ pub(crate) fn gen_create_index_plan(
     index_table_prost.owner = session.user_id();
     index_table_prost.dependent_relations = vec![table.id.table_id];
 
-    // FIXME: why telemetry need these information?
+    // FIXME: why sqlalchemy need these information?
     let original_columns = index_table
         .columns
         .iter()

src/meta/src/manager/catalog/mod.rs

@@ -90,11 +90,12 @@ macro_rules! commit_meta {
 }
 pub(crate) use commit_meta;
 use risingwave_common::util::column_index_mapping::ColIndexMapping;
-use risingwave_pb::expr::expr_node::RexNode;
 use risingwave_pb::meta::relation::RelationInfo;
 use risingwave_pb::meta::{CreatingJobInfo, Relation, RelationGroup};
 
-use crate::manager::catalog::utils::{alter_relation_rename, alter_relation_rename_refs};
+use crate::manager::catalog::utils::{
+    alter_relation_rename, alter_relation_rename_refs, ReplaceTableExprRewriter,
+};
 
 pub type CatalogManagerRef<S> = Arc<CatalogManager<S>>;
 
@@ -1920,28 +1921,16 @@ where
 
         let mut updated_indexes = vec![];
 
+        let expr_rewriter = ReplaceTableExprRewriter {
+            table_col_index_mapping: table_col_index_mapping.clone(),
+        };
+
         for index_id in &index_ids {
             let mut index = indexes.get_mut(*index_id).unwrap();
             index
                 .index_item
                 .iter_mut()
-                .for_each(|x| match x.rex_node.as_mut().unwrap() {
-                    RexNode::InputRef(input_col_idx) => {
-                        *input_col_idx =
-                            table_col_index_mapping.map(*input_col_idx as usize) as u32;
-                        assert_eq!(
-                            x.return_type,
-                            table.columns[*input_col_idx as usize]
-                                .column_desc
-                                .clone()
-                                .unwrap()
-                                .column_type
-                        );
-                    }
-                    RexNode::FuncCall(_) => unimplemented!(),
-                    _ => unreachable!(),
-                });
-
+                .for_each(|x| expr_rewriter.rewrite_expr(x));
             updated_indexes.push(indexes.get(index_id).cloned().unwrap());
         }
 

src/meta/src/manager/catalog/utils.rs

@@ -13,6 +13,9 @@
 // limitations under the License.
 
 use itertools::Itertools;
+use risingwave_common::util::column_index_mapping::ColIndexMapping;
+use risingwave_pb::expr::expr_node::RexNode;
+use risingwave_pb::expr::{ExprNode, FunctionCall, UserDefinedFunction};
 use risingwave_sqlparser::ast::{
     Array, CreateSink, CreateSinkStatement, CreateSourceStatement, Distinct, Expr, Function,
     FunctionArg, FunctionArgExpr, Ident, ObjectName, Query, SelectItem, SetExpr, Statement,
@@ -312,6 +315,37 @@ impl QueryRewriter<'_> {
     }
 }
 
+pub struct ReplaceTableExprRewriter {
+    pub table_col_index_mapping: ColIndexMapping,
+}
+
+impl ReplaceTableExprRewriter {
+    pub fn rewrite_expr(&self, expr: &mut ExprNode) {
+        let rex_node = expr.rex_node.as_mut().unwrap();
+        match rex_node {
+            RexNode::InputRef(input_col_idx) => {
+                *input_col_idx = self.table_col_index_mapping.map(*input_col_idx as usize) as u32
+            }
+            RexNode::Constant(_) => {}
+            RexNode::Udf(udf) => self.rewrite_udf(udf),
+            RexNode::FuncCall(function_call) => self.rewrite_function_call(function_call),
+        }
+    }
+
+    fn rewrite_udf(&self, udf: &mut UserDefinedFunction) {
+        udf.children
+            .iter_mut()
+            .for_each(|expr| self.rewrite_expr(expr));
+    }
+
+    fn rewrite_function_call(&self, function_call: &mut FunctionCall) {
+        function_call
+            .children
+            .iter_mut()
+            .for_each(|expr| self.rewrite_expr(expr));
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use crate::manager::catalog::utils::{alter_relation_rename, alter_relation_rename_refs};