Fix SqlCase in DDL parsing

apache · May 11, 2024 · 9b7c313 · 9b7c313
1 parent ef2c654
commit 9b7c313
Showing 1 changed file with 20 additions and 14 deletions.
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
@@ -46,7 +46,6 @@
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.SqlSelect;
-import org.apache.calcite.sql.fun.SqlCase;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.parser.SqlParseException;
 import org.apache.calcite.sql.parser.SqlParser;
@@ -413,12 +412,12 @@ private static List<String> parseColumnNameList(SqlNode sqlNode) {
             SqlIdentifier sqlIdentifier = (SqlIdentifier) sqlNode;
             String columnName = sqlIdentifier.names.get(sqlIdentifier.names.size() - 1);
             columnNameList.add(columnName);
-        } else if (sqlNode instanceof SqlBasicCall) {
-            SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlNode;
-            findSqlIdentifier(sqlBasicCall.getOperandList(), columnNameList);
-        } else if (sqlNode instanceof SqlCase) {
-            SqlCase sqlCase = (SqlCase) sqlNode;
-            findSqlIdentifier(sqlCase.getWhenOperands().getList(), columnNameList);
+        } else if (sqlNode instanceof SqlCall) {
+            SqlCall sqlCall = (SqlCall) sqlNode;
+            findSqlIdentifier(sqlCall.getOperandList(), columnNameList);
+        } else if (sqlNode instanceof SqlNodeList) {
+            SqlNodeList sqlNodeList = (SqlNodeList) sqlNode;
+            findSqlIdentifier(sqlNodeList.getList(), columnNameList);
         }
         return columnNameList;
     }
@@ -429,13 +428,12 @@ private static void findSqlIdentifier(List<SqlNode> sqlNodes, List<String> colum
                 SqlIdentifier sqlIdentifier = (SqlIdentifier) sqlNode;
                 String columnName = sqlIdentifier.names.get(sqlIdentifier.names.size() - 1);
                 columnNameList.add(columnName);
-            } else if (sqlNode instanceof SqlBasicCall) {
-                SqlBasicCall sqlBasicCall = (SqlBasicCall) sqlNode;
-                findSqlIdentifier(sqlBasicCall.getOperandList(), columnNameList);
-            } else if (sqlNode instanceof SqlCase) {
-                SqlCase sqlCase = (SqlCase) sqlNode;
-                SqlNodeList whenOperands = sqlCase.getWhenOperands();
-                findSqlIdentifier(whenOperands.getList(), columnNameList);
+            } else if (sqlNode instanceof SqlCall) {
+                SqlCall sqlCall = (SqlCall) sqlNode;
+                findSqlIdentifier(sqlCall.getOperandList(), columnNameList);
+            } else if (sqlNode instanceof SqlNodeList) {
+                SqlNodeList sqlNodeList = (SqlNodeList) sqlNode;
+                findSqlIdentifier(sqlNodeList.getList(), columnNameList);
             }
         }
     }
@@ -509,6 +507,14 @@ public static SqlNode rewriteExpression(SqlNode sqlNode, Map<String, SqlNode> re
                 }
             }
             return sqlIdentifier;
+        } else if (sqlNode instanceof SqlNodeList) {
+            SqlNodeList sqlNodeList = (SqlNodeList) sqlNode;
+            IntStream.range(0, sqlNodeList.size())
+                    .forEach(
+                            i ->
+                                    sqlNodeList.set(
+                                            i, rewriteExpression(sqlNodeList.get(i), replaceMap)));
+            return sqlNodeList;
         } else {
             return sqlNode;
         }