From d107204f7af67b3f7e9cc6e72b508ca6a5286d8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=A5=E6=A0=96?= Date: Tue, 11 Aug 2026 21:45:44 +0800 Subject: [PATCH 1/2] [hotfix][runtime] Respect UDF precedence during type validation --- .../cdc/runtime/parser/TransformParser.java | 77 ++++++++++++++++++- .../transform/PostTransformOperatorTest.java | 2 +- 2 files changed, 74 insertions(+), 5 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 index 1020a39705c..33528f54e72 100644 --- 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 @@ -55,8 +55,10 @@ import org.apache.calcite.sql.SqlKind; import org.apache.calcite.sql.SqlNode; import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.SqlOperator; import org.apache.calcite.sql.SqlOperatorTable; import org.apache.calcite.sql.SqlSelect; +import org.apache.calcite.sql.SqlSyntax; import org.apache.calcite.sql.parser.SqlParseException; import org.apache.calcite.sql.parser.SqlParser; import org.apache.calcite.sql.parser.SqlParserPos; @@ -66,9 +68,11 @@ import org.apache.calcite.sql.type.SqlTypeFactoryImpl; import org.apache.calcite.sql.type.SqlTypeName; import org.apache.calcite.sql.util.SqlOperatorTables; +import org.apache.calcite.sql.util.SqlShuttle; import org.apache.calcite.sql.validate.SqlConformance; import org.apache.calcite.sql.validate.SqlConformanceEnum; import org.apache.calcite.sql.validate.SqlDelegatingConformance; +import org.apache.calcite.sql.validate.SqlNameMatcher; import org.apache.calcite.sql.validate.SqlValidator; import org.apache.calcite.sql.validate.SqlValidatorUtil; import org.apache.calcite.sql2rel.SqlToRelConverter; @@ -214,18 +218,22 @@ private static RelNode sqlToRel( TransformSqlOperatorTable transformSqlOperatorTable = TransformSqlOperatorTable.instance(); SqlOperatorTable udfOperatorTable = SqlOperatorTables.of(udfFunctions); SqlOperatorTable aiFunctionOperatorTable = AiFunctionSqlOperatorTable.create(); + // Calcite looks up function candidates again when deriving a call's type. Rebind + // same-name calls and expose only UDF candidates to keep validation consistent with + // UDF-first code generation. SqlValidator validator = SqlValidatorUtil.newValidator( - SqlOperatorTables.chain( - transformSqlOperatorTable, + new UdfFirstSqlOperatorTable( udfOperatorTable, - aiFunctionOperatorTable), + SqlOperatorTables.chain( + transformSqlOperatorTable, aiFunctionOperatorTable)), calciteCatalogReader, factory, SqlValidator.Config.DEFAULT .withIdentifierExpansion(true) .withConformance(SqlConformanceEnum.MYSQL_5)); - SqlNode validateSqlNode = validator.validate(sqlNode); + SqlNode validateSqlNode = + validator.validate(resolveUserDefinedFunctions(sqlNode, udfFunctions)); SqlToRelConverter sqlToRelConverter = new SqlToRelConverter( null, @@ -240,6 +248,67 @@ private static RelNode sqlToRel( return relRoot.rel; } + private static SqlNode resolveUserDefinedFunctions( + SqlNode sqlNode, List udfFunctions) { + return sqlNode.accept( + new SqlShuttle() { + @Override + public SqlNode visit(SqlCall call) { + SqlNode visited = super.visit(call); + if (visited instanceof SqlBasicCall) { + SqlBasicCall basicCall = (SqlBasicCall) visited; + udfFunctions.stream() + .filter( + udf -> + udf.getName() + .equalsIgnoreCase( + basicCall + .getOperator() + .getName())) + .findFirst() + .ifPresent(basicCall::setOperator); + } + return visited; + } + }); + } + + private static final class UdfFirstSqlOperatorTable implements SqlOperatorTable { + private final SqlOperatorTable udfOperatorTable; + private final SqlOperatorTable fallbackOperatorTable; + + private UdfFirstSqlOperatorTable( + SqlOperatorTable udfOperatorTable, SqlOperatorTable fallbackOperatorTable) { + this.udfOperatorTable = udfOperatorTable; + this.fallbackOperatorTable = fallbackOperatorTable; + } + + @Override + public void lookupOperatorOverloads( + SqlIdentifier opName, + @Nullable SqlFunctionCategory category, + SqlSyntax syntax, + List operatorList, + SqlNameMatcher nameMatcher) { + List udfOperators = new ArrayList<>(); + udfOperatorTable.lookupOperatorOverloads( + opName, category, syntax, udfOperators, nameMatcher); + if (udfOperators.isEmpty()) { + fallbackOperatorTable.lookupOperatorOverloads( + opName, category, syntax, operatorList, nameMatcher); + } else { + operatorList.addAll(udfOperators); + } + } + + @Override + public List getOperatorList() { + List operators = new ArrayList<>(udfOperatorTable.getOperatorList()); + operators.addAll(fallbackOperatorTable.getOperatorList()); + return operators; + } + } + public static SqlSelect parseSelect(String statement) { SqlNode sqlNode; try { diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java index 8bfa4a03ed9..2e02b05a3b9 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java @@ -488,7 +488,7 @@ void testUdfTakesPrecedenceOverBuiltInFunction() throws Exception { PostTransformOperator.newBuilder() .addTransform( CUSTOMERS_TABLEID.identifier(), - "*, CAST(IFNULL(1, 0) AS VARCHAR) AS udf_ifnull, " + "*, IFNULL(1, 0) AS udf_ifnull, " + "TRY_CAST(col1) AS udf_try_cast, " + "NULLIF('%s', col1) AS udf_nullif", null) From 7213182d12ccb3960d00a2a88f2c0ccd0179be07 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=A5=E6=A0=96?= Date: Wed, 19 Aug 2026 21:34:50 +0800 Subject: [PATCH 2/2] [hotfix][runtime] Restrict UDF resolution to function calls --- .../cdc/runtime/parser/TransformParser.java | 3 +++ .../runtime/parser/TransformParserTest.java | 22 +++++++++++++++++++ 2 files changed, 25 insertions(+) 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 index 33528f54e72..02691968c27 100644 --- 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 @@ -257,6 +257,9 @@ public SqlNode visit(SqlCall call) { SqlNode visited = super.visit(call); if (visited instanceof SqlBasicCall) { SqlBasicCall basicCall = (SqlBasicCall) visited; + if (basicCall.getOperator().getSyntax().family != SqlSyntax.FUNCTION) { + return visited; + } udfFunctions.stream() .filter( udf -> diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java index 50fcc1e4556..06db8f9f472 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java @@ -1230,6 +1230,28 @@ void testUdfTakesPrecedenceOverBuiltInFunction() { udfDescriptors); } + @Test + void testUdfDoesNotReplaceStructuralOperator() { + List udfDescriptors = + Collections.singletonList( + new UserDefinedFunctionDescriptor( + "as", + "org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass")); + + List projectionColumns = + TransformParser.generateProjectionColumns( + "id AS alias", + DUMMY_COLUMNS, + udfDescriptors, + new SupportedMetadataColumn[0]); + + Assertions.assertThat(projectionColumns).hasSize(1); + ProjectionColumn projectionColumn = projectionColumns.get(0); + Assertions.assertThat(projectionColumn.getColumnName()).isEqualTo("alias"); + Assertions.assertThat(projectionColumn.getDataType()).isEqualTo(DataTypes.INT()); + Assertions.assertThat(projectionColumn.getScriptExpression()).isEqualTo("$0"); + } + @Test public void testTranslateUdfFilterToJaninoExpressionWithColumnNameMap() { List columns =