Skip to content

[hotfix][runtime] Respect UDF precedence during type validation #4505

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lvyanquan merged 2 commits into from
Aug 20, 2026
Merged

[hotfix][runtime] Respect UDF precedence during type validation #4505

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d107204
[hotfix][runtime] Respect UDF precedence during type validation
Aug 11, 2026
7213182
[hotfix][runtime] Restrict UDF resolution to function calls
Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
80 changes: 76 additions & 4 deletions flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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,
Expand All @@ -240,6 +248,70 @@ private static RelNode sqlToRel(
return relRoot.rel;
}

private static SqlNode resolveUserDefinedFunctions(
SqlNode sqlNode, List<SqlFunction> 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;
if (basicCall.getOperator().getSyntax().family != SqlSyntax.FUNCTION) {
return visited;
}
udfFunctions.stream()

@lvyanquan lvyanquan Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

resolveUserDefinedFunctions matches every SqlBasicCall solely by operator name, including structural operators. Since UDF names are not restricted, registering a UDF named as causes id AS alias to have its AS operator replaced by the UDF, so Calcite validates it as a function call rather than an alias expression.
Please restrict rebinding to actual function-call syntax and leave structural/special operators such as AS intact.

      if (basicCall.getOperator().getSyntax().family
              != SqlSyntax.FUNCTION) {
          return visited;
      }

.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<SqlOperator> operatorList,
SqlNameMatcher nameMatcher) {
List<SqlOperator> 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<SqlOperator> getOperatorList() {
List<SqlOperator> operators = new ArrayList<>(udfOperatorTable.getOperatorList());
operators.addAll(fallbackOperatorTable.getOperatorList());
return operators;
}
}

public static SqlSelect parseSelect(String statement) {
SqlNode sqlNode;
try {
Expand Down
2 changes: 1 addition & 1 deletion ...test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
22 changes: 22 additions & 0 deletions flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1230,6 +1230,28 @@ void testUdfTakesPrecedenceOverBuiltInFunction() {
udfDescriptors);
}

@Test
void testUdfDoesNotReplaceStructuralOperator() {
List<UserDefinedFunctionDescriptor> udfDescriptors =
Collections.singletonList(
new UserDefinedFunctionDescriptor(
"as",
"org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass"));

List<ProjectionColumn> 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<Column> columns =
Expand Down
Loading