From 7d043161bd17299fc886bc3e4c19134fb06c2175 Mon Sep 17 00:00:00 2001 From: haruki Date: Tue, 28 Jul 2026 12:59:46 +0800 Subject: [PATCH 1/4] [FLINK-40240][runtime] Support TRY_CAST, IFNULL and NULLIF in YAML Transform --- .../content.zh/docs/core-concept/transform.md | 4 + docs/content/docs/core-concept/transform.md | 4 + .../src/test/resources/specs/casting.yaml | 16 ++ .../src/test/resources/specs/condition.yaml | 15 ++ .../functions/impl/CastingFunctions.java | 50 ++++- .../functions/impl/LogicalFunctions.java | 43 ++++ .../cdc/runtime/parser/JaninoCompiler.java | 170 ++++++++++++-- .../cdc/runtime/parser/TransformParser.java | 15 +- .../parser/TransformSqlSyntaxRewriter.java | 210 ++++++++++++++++++ .../metadata/TransformSqlOperatorTable.java | 24 ++ .../metadata/TransformSqlReturnTypes.java | 21 ++ .../functions/impl/CastingFunctionsTest.java | 93 ++++++++ .../functions/impl/LogicalFunctionsTest.java | 47 ++++ .../transform/PostTransformOperatorTest.java | 8 + .../runtime/parser/JaninoCompilerTest.java | 26 +++ .../runtime/parser/TransformParserTest.java | 73 +++++- .../TransformSqlSyntaxRewriterTest.java | 77 +++++++ 17 files changed, 876 insertions(+), 20 deletions(-) create mode 100644 flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformSqlSyntaxRewriter.java create mode 100644 flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctionsTest.java create mode 100644 flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctionsTest.java create mode 100644 flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformSqlSyntaxRewriterTest.java diff --git a/docs/content.zh/docs/core-concept/transform.md b/docs/content.zh/docs/core-concept/transform.md index d6daab54713..0f9ce4df039 100644 --- a/docs/content.zh/docs/core-concept/transform.md +++ b/docs/content.zh/docs/core-concept/transform.md @@ -244,11 +244,15 @@ Flink CDC 使用 [Calcite](https://calcite.apache.org/) 来解析表达式并且 | CASE WHEN condition1 THEN result1 (WHEN condition2 THEN result2)* (ELSE result_z) END | 嵌套三元表达式 | 当第一个 conditionX 满足时，返回 resultX。如果没有条件满足，如果提供了 result_z 则返回 result_z，否则返回 NULL。 | | COALESCE(value1 [, value2]*) | coalesce(Object... objects) | 返回第一个不为 NULL 的参数。如果所有参数都为 NULL，则返回 NULL。返回类型是所有参数中限制最少的公共类型。如果所有参数都可为空，则返回类型也可为空。 | | IF(condition, true_value, false_value) | condition ? true_value : false_value | 如果条件满足，返回 true_value，否则返回 false_value。例如，IF(5 > 3, 5, 3) 返回 5。 | +| IFNULL(value, replacement) | ifNull(value, replacement) | 当 value 为 NULL 时返回 replacement，否则返回 value。两个参数必须存在公共类型。仅当 replacement 可为 NULL 时，结果才可为 NULL。 | +| NULLIF(value1, value2) | nullIf(value1, value2) | 当 value1 与 value2 相等时返回 NULL，否则返回 value1。返回类型为 value1 对应的可空类型。 | ## 转换函数 你可以使用 `CAST( AS )` 语法将任何有效的表达式 `` 转换为特定类型 ``。可能的转换路径如下： +`TRY_CAST( AS )` 支持与 `CAST` 相同的转换路径。当某个值无法转换时返回 NULL；不支持的源类型到目标类型转换路径会在 pipeline 校验或初始化阶段被拒绝。配置错误、UDF 异常和内部程序错误不会被转换为 NULL。 + | 源类型 | 目标类型 | 说明 | |-------------------------------------|-----------|--------------------------------------------------| | ANY | STRING | 所有类型都可以转换为 STRING。 | diff --git a/docs/content/docs/core-concept/transform.md b/docs/content/docs/core-concept/transform.md index 28bda0e34a9..005f32edbc1 100644 --- a/docs/content/docs/core-concept/transform.md +++ b/docs/content/docs/core-concept/transform.md @@ -245,11 +245,15 @@ Logical functions follow SQL three-valued logic for nullable BOOLEAN values. `AN | CASE WHEN condition1 THEN result1 (WHEN condition2 THEN result2)* (ELSE result_z) END | Nested ternary expression | Returns resultX when the first conditionX is met. When no condition is met, returns result_z if it is provided and returns NULL otherwise. | | COALESCE(value1 [, value2]*) | coalesce(Object... objects) | Returns the first argument that is not NULL.If all arguments are NULL, it returns NULL as well. The return type is the least restrictive, common type of all of its arguments. The return type is nullable if all arguments are nullable as well. | | IF(condition, true_value, false_value) | condition ? true_value : false_value | Returns the true_value if condition is met, otherwise false_value. E.g., IF(5 > 3, 5, 3) returns 5. | +| IFNULL(value, replacement) | ifNull(value, replacement) | Returns `replacement` when `value` is NULL; otherwise, returns `value`. The arguments must have a common type. The result can be NULL only when `replacement` can be NULL. | +| NULLIF(value1, value2) | nullIf(value1, value2) | Returns NULL when `value1` equals `value2`; otherwise, returns `value1`. Its return type is the nullable type of `value1`. | ## Casting Functions You can use `CAST( AS )` syntax to convert any valid expression `` to a specific type ``. Possible conversion paths are: +`TRY_CAST( AS )` supports the same conversion paths as `CAST`. It returns NULL when a value cannot be converted. An unsupported source-to-target type path is rejected during pipeline validation or initialization. Configuration errors, UDF failures, and internal errors are not converted to NULL. + | Source Type | Target Type | Notes | |-------------------------------------|-------------|--------------------------------------------------------------------------------------------| | ANY | STRING | All types can be cast to STRING. | diff --git a/flink-cdc-composer/src/test/resources/specs/casting.yaml b/flink-cdc-composer/src/test/resources/specs/casting.yaml index f5ce0e802a5..288b8df16dc 100644 --- a/flink-cdc-composer/src/test/resources/specs/casting.yaml +++ b/flink-cdc-composer/src/test/resources/specs/casting.yaml @@ -263,6 +263,22 @@ DataChangeEvent{tableId=foo.bar.baz, before=[-1, 1970-01-09T08:57:36.789723456, 1970-01-10T15:49:27.891834561, 1970-01-11T22:41:18.912945612, 1970-01-09T08:57:36.789723456, 1970-01-10T22:49:27.891834561, 1970-01-12T10:41:18.912945612, 1970-01-09T16:57:36.789723456, 1970-01-10T23:49:27.891834561, 1970-01-12T06:41:18.912945612, 2019-12-31T21:48:25], after=[], op=DELETE, meta=()} DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null, null, null, null, null, null, null, null, 2019-12-31T21:48:25], op=INSERT, meta=()} DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null, null, null, null, null, null, null, null, 2019-12-31T21:48:25], after=[], op=DELETE, meta=()} +- do: Try Cast + projection: |- + id_ + TRY_CAST('123' AS INTEGER) AS valid_int + TRY_CAST('FOOBAR' AS INTEGER) AS invalid_int + TRY_CAST('2019-12-31T21:48:25' AS TIMESTAMP(6)) AS valid_timestamp + TRY_CAST('FOOBAR' AS TIMESTAMP(6)) AS invalid_timestamp + IFNULL(TRY_CAST('FOOBAR' AS INTEGER), 17) AS int_fallback + primary-key: id_ + expect: |- + CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`valid_int` INT,`invalid_int` INT,`valid_timestamp` TIMESTAMP(3),`invalid_timestamp` TIMESTAMP(3),`int_fallback` INT NOT NULL}, primaryKeys=id_, options=()} + DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 123, null, 2019-12-31T21:48:25, null, 17], op=INSERT, meta=()} + DataChangeEvent{tableId=foo.bar.baz, before=[1, 123, null, 2019-12-31T21:48:25, null, 17], after=[-1, 123, null, 2019-12-31T21:48:25, null, 17], op=UPDATE, meta=()} + DataChangeEvent{tableId=foo.bar.baz, before=[-1, 123, null, 2019-12-31T21:48:25, null, 17], after=[], op=DELETE, meta=()} + DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, 123, null, 2019-12-31T21:48:25, null, 17], op=INSERT, meta=()} + DataChangeEvent{tableId=foo.bar.baz, before=[0, 123, null, 2019-12-31T21:48:25, null, 17], after=[], op=DELETE, meta=()} - do: Cast To Timestamp Failure projection: |- id_ diff --git a/flink-cdc-composer/src/test/resources/specs/condition.yaml b/flink-cdc-composer/src/test/resources/specs/condition.yaml index d220ffa86ba..bf2d98fa499 100644 --- a/flink-cdc-composer/src/test/resources/specs/condition.yaml +++ b/flink-cdc-composer/src/test/resources/specs/condition.yaml @@ -100,3 +100,18 @@ DataChangeEvent{tableId=foo.bar.baz, before=[-1, 非正, 短], after=[], op=DELETE, meta=()} DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, 非正, 短], op=INSERT, meta=()} DataChangeEvent{tableId=foo.bar.baz, before=[0, 非正, 短], after=[], op=DELETE, meta=()} +- do: IfNull and NullIf Clauses + projection: |- + id_, int_ + IFNULL(int_, 17) AS int_fallback + NULLIF(int_, 4) AS int_nullif + NULLIF(CAST(int_ AS BIGINT), CAST(4 AS INT)) AS bigint_nullif + filter: IFNULL(NULLIF(id_, -999), 0) IS NOT NULL + primary-key: id_ + expect: |- + CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT NULL 'Identifier',`int_` INT,`int_fallback` INT NOT NULL,`int_nullif` INT,`bigint_nullif` BIGINT}, primaryKeys=id_, options=()} + DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 4, 4, null, null], op=INSERT, meta=()} + DataChangeEvent{tableId=foo.bar.baz, before=[1, 4, 4, null, null], after=[-1, -4, -4, -4, -4], op=UPDATE, meta=()} + DataChangeEvent{tableId=foo.bar.baz, before=[-1, -4, -4, -4, -4], after=[], op=DELETE, meta=()} + DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, 17, null, null], op=INSERT, meta=()} + DataChangeEvent{tableId=foo.bar.baz, before=[0, null, 17, null, null], after=[], op=DELETE, meta=()} diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctions.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctions.java index d895ea17722..930faf6a28e 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctions.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctions.java @@ -264,7 +264,51 @@ public static BigDecimal castToBigDecimal(Object object, int precision, int scal } public static LocalDateTime castToTimestamp(Object object, String timezone) { - ZoneId zoneId = ZoneId.of(timezone); + return castToTimestamp(object, ZoneId.of(timezone), false); + } + + public static String tryCastToString(Object object) { + return castToString(object); + } + + public static Boolean tryCastToBoolean(Object object) { + return castToBoolean(object); + } + + public static Byte tryCastToByte(Object object) { + return castToByte(object); + } + + public static Short tryCastToShort(Object object) { + return castToShort(object); + } + + public static Integer tryCastToInteger(Object object) { + return castToInteger(object); + } + + public static Long tryCastToLong(Object object) { + return castToLong(object); + } + + public static Float tryCastToFloat(Object object) { + return castToFloat(object); + } + + public static Double tryCastToDouble(Object object) { + return castToDouble(object); + } + + public static BigDecimal tryCastToBigDecimal(Object object, int precision, int scale) { + return castToBigDecimal(object, precision, scale); + } + + public static LocalDateTime tryCastToTimestamp(Object object, String timezone) { + return castToTimestamp(object, ZoneId.of(timezone), true); + } + + private static LocalDateTime castToTimestamp( + Object object, ZoneId zoneId, boolean returnNullOnFailure) { if (object == null) { return null; } @@ -302,7 +346,9 @@ public static LocalDateTime castToTimestamp(Object object, String timezone) { return ZonedDateTime.parse(stringRep).toLocalDateTime(); } catch (DateTimeParseException ignored) { } - + if (returnNullOnFailure) { + return null; + } throw new IllegalArgumentException( "Unable to parse given string as timestamp: " + stringRep); } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java index 07366bd34e7..273194f260a 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java @@ -17,6 +17,8 @@ package org.apache.flink.cdc.runtime.functions.impl; +import java.math.BigDecimal; +import java.util.Objects; import java.util.function.Supplier; /** Logical built-in functions. */ @@ -102,4 +104,45 @@ public static Object coalesce(Object... objects) { } return null; } + + public static T ifNull(T value, T replacement) { + return value != null ? value : replacement; + } + + public static T nullIf(T value, Object comparison) { + return valuesEqualForNullIf(value, comparison) ? null : value; + } + + private static boolean valuesEqualForNullIf(Object value, Object comparison) { + if (value == null || comparison == null) { + return false; + } + if (!(value instanceof Number) || !(comparison instanceof Number)) { + return Objects.deepEquals(value, comparison); + } + + Number left = (Number) value; + Number right = (Number) comparison; + if (left instanceof Double || right instanceof Double) { + double leftValue = left.doubleValue(); + double rightValue = right.doubleValue(); + if (Double.isFinite(leftValue) && Double.isFinite(rightValue)) { + return BigDecimal.valueOf(leftValue).compareTo(BigDecimal.valueOf(rightValue)) == 0; + } + return Double.compare(leftValue, rightValue) == 0; + } + if (left instanceof Float || right instanceof Float) { + return Float.compare(left.floatValue(), right.floatValue()) == 0; + } + if (left instanceof BigDecimal || right instanceof BigDecimal) { + return toBigDecimal(left).compareTo(toBigDecimal(right)) == 0; + } + return left.longValue() == right.longValue(); + } + + private static BigDecimal toBigDecimal(Number number) { + return number instanceof BigDecimal + ? (BigDecimal) number + : BigDecimal.valueOf(number.longValue()); + } } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java index 6f8d0e464bc..b1f352a260c 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java @@ -25,6 +25,7 @@ import org.apache.flink.cdc.common.source.SupportedMetadataColumn; import org.apache.flink.cdc.common.types.DataType; import org.apache.flink.cdc.common.types.DataTypeRoot; +import org.apache.flink.cdc.common.types.DecimalType; import org.apache.flink.cdc.common.utils.Preconditions; import org.apache.flink.cdc.common.utils.StringUtils; import org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptor; @@ -233,6 +234,9 @@ private static Java.Rvalue translateSqlBasicCall(Context context, SqlBasicCall s if (DATE_PART_FUNCTION_UNITS.containsKey(functionName)) { return generateDatePartFunctionOperation(context, sqlBasicCall, functionName); } + if (functionName.equals("TRY_CAST")) { + return generateTryCastOperation(context, sqlBasicCall); + } List operandList = sqlBasicCall.getOperandList(); List atoms = new ArrayList<>(); @@ -448,6 +452,8 @@ private static Java.Rvalue sqlBasicCallToJaninoRvalue( case LESS_THAN_OR_EQUAL: case GREATER_THAN_OR_EQUAL: return generateCompareOperation(context, sqlBasicCall, atoms); + case NULLIF: + return generateNullIfOperation(context, sqlBasicCall, atoms); case CAST: return generateCastOperation(context, sqlBasicCall, atoms); case TIMESTAMP_DIFF: @@ -558,6 +564,14 @@ private static boolean isIdentifierNullable(Context context, SqlIdentifier sqlId } private static boolean isBasicCallNullable(Context context, SqlBasicCall sqlBasicCall) { + if (sqlBasicCall.getOperator().getName().equalsIgnoreCase("IFNULL")) { + List operands = sqlBasicCall.getOperandList(); + if (operands.size() != 2) { + return true; + } + return isExpressionNullable(context, operands.get(0)) + && isExpressionNullable(context, operands.get(1)); + } switch (sqlBasicCall.getKind()) { case AND: case OR: @@ -636,7 +650,33 @@ private static Java.Rvalue generateCastOperation( } List operandList = sqlBasicCall.getOperandList(); SqlDataTypeSpec sqlDataTypeSpec = (SqlDataTypeSpec) operandList.get(1); - return generateTypeConvertMethod(context, sqlDataTypeSpec, atoms); + return generateTypeConvertMethod(context, sqlDataTypeSpec, atoms, false); + } + + private static Java.Rvalue generateTryCastOperation( + Context context, SqlBasicCall sqlBasicCall) { + List operands = sqlBasicCall.getOperandList(); + if (operands.size() != 1 || !(operands.get(0) instanceof SqlBasicCall)) { + throw new ParseException("Unrecognized expression: " + sqlBasicCall); + } + + SqlBasicCall castCall = (SqlBasicCall) operands.get(0); + List castOperands = castCall.getOperandList(); + if (castCall.getKind() != SqlKind.CAST + || castOperands.size() != 2 + || !(castOperands.get(1) instanceof SqlDataTypeSpec)) { + throw new ParseException("Unrecognized expression: " + sqlBasicCall); + } + + Java.Rvalue castOperand = translateSqlNodeToJaninoRvalue(context, castOperands.get(0)); + if (castOperand == null) { + throw new ParseException("Unrecognized expression: " + sqlBasicCall); + } + return generateTypeConvertMethod( + context, + (SqlDataTypeSpec) castOperands.get(1), + new Java.Rvalue[] {castOperand}, + true); } private static Java.Rvalue generateCompareOperation( @@ -765,6 +805,10 @@ private static Java.Rvalue generateItemAccessOperation( // Get the Java class for the result type and add a cast // Use getCanonicalName() to correctly handle array types (e.g., byte[] instead of "[B") + return castToJavaType(resultType, methodInvocation); + } + + private static Java.Rvalue castToJavaType(DataType resultType, Java.Rvalue expression) { Class javaClass = JavaClassConverter.toJavaClass(resultType); if (javaClass != null && javaClass != Object.class) { String canonicalName = javaClass.getCanonicalName(); @@ -776,10 +820,10 @@ private static Java.Rvalue generateItemAccessOperation( new Java.Annotation[0], canonicalName.split("\\."), null), - methodInvocation); + expression); } } - return methodInvocation; + return expression; } private static Java.Rvalue generateOtherFunctionOperation( @@ -795,6 +839,10 @@ private static Java.Rvalue generateOtherFunctionOperation( } else { throw new ParseException("Unrecognized expression: " + sqlBasicCall); } + } else if (operationName.equals("IFNULL")) { + return generateIfNullOperation(context, sqlBasicCall, atoms); + } else if (operationName.equals("NULLIF")) { + return generateNullIfOperation(context, sqlBasicCall, atoms); } else { Optional udfFunctionOptional = context.udfDescriptors.stream() @@ -819,6 +867,77 @@ private static Java.Rvalue generateOtherFunctionOperation( } } + private static Java.Rvalue generateIfNullOperation( + Context context, SqlBasicCall sqlBasicCall, Java.Rvalue[] atoms) { + if (atoms.length != 2) { + throw new ParseException("Unrecognized expression: " + sqlBasicCall); + } + + DataType resultType = + TransformParser.deduceSubExpressionType( + context.columns, + sqlBasicCall, + context.udfDescriptors, + context.supportedMetadataColumns); + Java.Rvalue[] coercedAtoms = new Java.Rvalue[atoms.length]; + for (int index = 0; index < atoms.length; index++) { + coercedAtoms[index] = generateNumericTypeConvertMethod(resultType, atoms[index]); + } + return castToJavaType(resultType, generateFunctionOperation("ifNull", coercedAtoms)); + } + + private static Java.Rvalue generateNullIfOperation( + Context context, SqlBasicCall sqlBasicCall, Java.Rvalue[] atoms) { + if (atoms.length != 2) { + throw new ParseException("Unrecognized expression: " + sqlBasicCall); + } + Java.Rvalue operation = generateFunctionOperation("nullIf", atoms); + SqlNode value = sqlBasicCall.getOperandList().get(0); + if (value instanceof SqlLiteral && ((SqlLiteral) value).getValue() == null) { + return operation; + } + DataType resultType = + TransformParser.deduceSubExpressionType( + context.columns, + value, + context.udfDescriptors, + context.supportedMetadataColumns); + return castToJavaType(resultType, operation); + } + + private static Java.Rvalue generateNumericTypeConvertMethod( + DataType dataType, Java.Rvalue atom) { + switch (dataType.getTypeRoot()) { + case TINYINT: + return generateFunctionOperation("castToByte", new Java.Rvalue[] {atom}); + case SMALLINT: + return generateFunctionOperation("castToShort", new Java.Rvalue[] {atom}); + case INTEGER: + return generateFunctionOperation("castToInteger", new Java.Rvalue[] {atom}); + case BIGINT: + return generateFunctionOperation("castToLong", new Java.Rvalue[] {atom}); + case FLOAT: + return generateFunctionOperation("castToFloat", new Java.Rvalue[] {atom}); + case DOUBLE: + return generateFunctionOperation("castToDouble", new Java.Rvalue[] {atom}); + case DECIMAL: + DecimalType decimalType = (DecimalType) dataType; + return generateFunctionOperation( + "castToBigDecimal", + new Java.Rvalue[] { + atom, + new Java.AmbiguousName( + Location.NOWHERE, + new String[] {String.valueOf(decimalType.getPrecision())}), + new Java.AmbiguousName( + Location.NOWHERE, + new String[] {String.valueOf(decimalType.getScale())}) + }); + default: + return atom; + } + } + private static Java.Rvalue generateTimezoneFreeTemporalFunctionOperation( Context context, String operationName) { return new Java.MethodInvocation( @@ -865,22 +984,41 @@ private static Java.Rvalue generateTimezoneRequiredTemporalConversionFunctionOpe } private static Java.Rvalue generateTypeConvertMethod( - Context context, SqlDataTypeSpec sqlDataTypeSpec, Java.Rvalue[] atoms) { + Context context, + SqlDataTypeSpec sqlDataTypeSpec, + Java.Rvalue[] atoms, + boolean tryCast) { switch (sqlDataTypeSpec.getTypeName().getSimple().toUpperCase()) { case "BOOLEAN": - return new Java.MethodInvocation(Location.NOWHERE, null, "castToBoolean", atoms); + return new Java.MethodInvocation( + Location.NOWHERE, + null, + tryCast ? "tryCastToBoolean" : "castToBoolean", + atoms); case "TINYINT": - return new Java.MethodInvocation(Location.NOWHERE, null, "castToByte", atoms); + return new Java.MethodInvocation( + Location.NOWHERE, null, tryCast ? "tryCastToByte" : "castToByte", atoms); case "SMALLINT": - return new Java.MethodInvocation(Location.NOWHERE, null, "castToShort", atoms); + return new Java.MethodInvocation( + Location.NOWHERE, null, tryCast ? "tryCastToShort" : "castToShort", atoms); case "INTEGER": - return new Java.MethodInvocation(Location.NOWHERE, null, "castToInteger", atoms); + return new Java.MethodInvocation( + Location.NOWHERE, + null, + tryCast ? "tryCastToInteger" : "castToInteger", + atoms); case "BIGINT": - return new Java.MethodInvocation(Location.NOWHERE, null, "castToLong", atoms); + return new Java.MethodInvocation( + Location.NOWHERE, null, tryCast ? "tryCastToLong" : "castToLong", atoms); case "FLOAT": - return new Java.MethodInvocation(Location.NOWHERE, null, "castToFloat", atoms); + return new Java.MethodInvocation( + Location.NOWHERE, null, tryCast ? "tryCastToFloat" : "castToFloat", atoms); case "DOUBLE": - return new Java.MethodInvocation(Location.NOWHERE, null, "castToDouble", atoms); + return new Java.MethodInvocation( + Location.NOWHERE, + null, + tryCast ? "tryCastToDouble" : "castToDouble", + atoms); case "DECIMAL": int precision = 10; int scale = 0; @@ -904,12 +1042,16 @@ private static Java.Rvalue generateTypeConvertMethod( return new Java.MethodInvocation( Location.NOWHERE, null, - "castToBigDecimal", + tryCast ? "tryCastToBigDecimal" : "castToBigDecimal", newAtoms.toArray(new Java.Rvalue[0])); case "CHAR": case "VARCHAR": case "STRING": - return new Java.MethodInvocation(Location.NOWHERE, null, "castToString", atoms); + return new Java.MethodInvocation( + Location.NOWHERE, + null, + tryCast ? "tryCastToString" : "castToString", + atoms); case "TIMESTAMP": List timestampAtoms = new ArrayList<>(Arrays.asList(atoms)); timestampAtoms.add( @@ -917,7 +1059,7 @@ private static Java.Rvalue generateTypeConvertMethod( return new Java.MethodInvocation( Location.NOWHERE, null, - "castToTimestamp", + tryCast ? "tryCastToTimestamp" : "castToTimestamp", timestampAtoms.toArray(new Java.Rvalue[0])); default: throw new ParseException( 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 4912ef5a5d7..6020245bad6 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 @@ -100,7 +100,7 @@ public class TransformParser { private static SqlParser getCalciteParser(String sql) { return SqlParser.create( - sql, + TransformSqlSyntaxRewriter.rewriteTryCast(sql), SqlParser.Config.DEFAULT .withConformance(SqlConformanceEnum.MYSQL_5) .withCaseSensitive(true) @@ -344,11 +344,20 @@ public static List generateProjectionColumns( } else { List originalColumnNames = parseColumnNameList(exprNode); Map columnNameMap = generateColumnNameMap(originalColumnNames); + DataType dataType = + CalciteDataTypeConverter.convertCalciteRelDataTypeToDataType( + relDataType); + if (exprNode instanceof SqlBasicCall + && ((SqlBasicCall) exprNode) + .getOperator() + .getName() + .equalsIgnoreCase("IFNULL")) { + dataType = dataType.copy(relDataType.isNullable()); + } projectionColumn = ProjectionColumn.ofCalculated( columnName, - CalciteDataTypeConverter.convertCalciteRelDataTypeToDataType( - relDataType), + dataType, exprNode.toString(), JaninoCompiler.translateSqlNodeToJaninoExpression( JaninoCompiler.Context.of( diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformSqlSyntaxRewriter.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformSqlSyntaxRewriter.java new file mode 100644 index 00000000000..7b70679fbed --- /dev/null +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformSqlSyntaxRewriter.java @@ -0,0 +1,210 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.runtime.parser; + +/** Rewrites Transform SQL syntax that is not recognized by the bundled Calcite parser. */ +final class TransformSqlSyntaxRewriter { + + private static final String TRY_CAST = "TRY_CAST"; + + private TransformSqlSyntaxRewriter() {} + + static String rewriteTryCast(String sql) { + return rewriteRange(sql, 0, sql.length()); + } + + private static String rewriteRange(String sql, int start, int end) { + StringBuilder rewritten = new StringBuilder(end - start); + int index = start; + while (index < end) { + int protectedEnd = findProtectedRegionEnd(sql, index, end); + if (protectedEnd > index) { + rewritten.append(sql, index, protectedEnd); + index = protectedEnd; + continue; + } + + if (!matchesTryCast(sql, index, end)) { + rewritten.append(sql.charAt(index)); + index++; + continue; + } + + int openingParenthesis = findOpeningParenthesis(sql, index + TRY_CAST.length(), end); + if (openingParenthesis < 0) { + rewritten.append(sql.charAt(index)); + index++; + continue; + } + + int closingParenthesis = findMatchingParenthesis(sql, openingParenthesis, end); + if (closingParenthesis < 0) { + rewritten.append(sql, index, end); + break; + } + + String operand = rewriteRange(sql, openingParenthesis + 1, closingParenthesis); + rewritten.append(sql, index, openingParenthesis + 1); + if (containsTopLevelAs(operand)) { + rewritten.append("CAST(").append(operand).append(')'); + } else { + rewritten.append(operand); + } + rewritten.append(')'); + index = closingParenthesis + 1; + } + return rewritten.toString(); + } + + private static boolean matchesTryCast(String sql, int index, int end) { + int keywordEnd = index + TRY_CAST.length(); + return keywordEnd <= end + && sql.regionMatches(true, index, TRY_CAST, 0, TRY_CAST.length()) + && (index == 0 || !isIdentifierPart(sql.charAt(index - 1))) + && (keywordEnd == end || !isIdentifierPart(sql.charAt(keywordEnd))); + } + + private static boolean isIdentifierPart(char character) { + return Character.isLetterOrDigit(character) || character == '_' || character == '$'; + } + + private static int findOpeningParenthesis(String sql, int start, int end) { + int index = start; + while (index < end) { + char character = sql.charAt(index); + if (Character.isWhitespace(character)) { + index++; + } else if (isLineCommentStart(sql, index, end)) { + index = findLineCommentEnd(sql, index, end); + } else if (isBlockCommentStart(sql, index, end)) { + index = findBlockCommentEnd(sql, index, end); + } else { + return character == '(' ? index : -1; + } + } + return -1; + } + + private static int findMatchingParenthesis(String sql, int openingParenthesis, int end) { + int depth = 1; + int index = openingParenthesis + 1; + while (index < end) { + int protectedEnd = findProtectedRegionEnd(sql, index, end); + if (protectedEnd > index) { + index = protectedEnd; + continue; + } + + char character = sql.charAt(index); + if (character == '(') { + depth++; + } else if (character == ')' && --depth == 0) { + return index; + } + index++; + } + return -1; + } + + private static boolean containsTopLevelAs(String sql) { + int depth = 0; + int index = 0; + while (index < sql.length()) { + int protectedEnd = findProtectedRegionEnd(sql, index, sql.length()); + if (protectedEnd > index) { + index = protectedEnd; + continue; + } + + char character = sql.charAt(index); + if (character == '(') { + depth++; + } else if (character == ')') { + depth--; + } else if (depth == 0 + && index + 2 <= sql.length() + && sql.regionMatches(true, index, "AS", 0, 2) + && (index == 0 || !isIdentifierPart(sql.charAt(index - 1))) + && (index + 2 == sql.length() || !isIdentifierPart(sql.charAt(index + 2)))) { + return true; + } + index++; + } + return false; + } + + private static int findProtectedRegionEnd(String sql, int index, int end) { + char character = sql.charAt(index); + if (character == '\'' || character == '"' || character == '`') { + return findQuotedRegionEnd(sql, index, end, character); + } + if (isLineCommentStart(sql, index, end)) { + return findLineCommentEnd(sql, index, end); + } + if (isBlockCommentStart(sql, index, end)) { + return findBlockCommentEnd(sql, index, end); + } + return index; + } + + private static int findQuotedRegionEnd(String sql, int start, int end, char quote) { + int index = start + 1; + while (index < end) { + char character = sql.charAt(index); + if (character == '\\' && index + 1 < end) { + index += 2; + } else if (character == quote) { + if (index + 1 < end && sql.charAt(index + 1) == quote) { + index += 2; + } else { + return index + 1; + } + } else { + index++; + } + } + return end; + } + + private static boolean isLineCommentStart(String sql, int index, int end) { + return index + 1 < end && sql.charAt(index) == '-' && sql.charAt(index + 1) == '-'; + } + + private static int findLineCommentEnd(String sql, int start, int end) { + int index = start + 2; + while (index < end && sql.charAt(index) != '\n' && sql.charAt(index) != '\r') { + index++; + } + return index; + } + + private static boolean isBlockCommentStart(String sql, int index, int end) { + return index + 1 < end && sql.charAt(index) == '/' && sql.charAt(index + 1) == '*'; + } + + private static int findBlockCommentEnd(String sql, int start, int end) { + int index = start + 2; + while (index + 1 < end) { + if (sql.charAt(index) == '*' && sql.charAt(index + 1) == '/') { + return index + 2; + } + index++; + } + return end; + } +} diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java index 096e3fa2b0b..ff6ca680305 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java @@ -610,6 +610,22 @@ public SqlSyntax getSyntax() { // --------------------- public static final SqlCaseOperator CASE = SqlStdOperatorTable.CASE; public static final SqlFunction COALESCE = SqlStdOperatorTable.COALESCE; + public static final SqlFunction NULLIF = + new SqlFunction( + "NULLIF", + SqlKind.NULLIF, + ReturnTypes.ARG0_FORCE_NULLABLE, + null, + OperandTypes.COMPARABLE_UNORDERED_COMPARABLE_UNORDERED, + SqlFunctionCategory.SYSTEM); + public static final SqlFunction IFNULL = + new SqlFunction( + "IFNULL", + SqlKind.OTHER_FUNCTION, + TransformSqlReturnTypes.IF_NULL, + null, + OperandTypes.SAME_SAME, + SqlFunctionCategory.SYSTEM); public static final SqlFunction IF = new SqlFunction( "IF", @@ -657,6 +673,14 @@ public SqlSyntax getSyntax() { // Cast Functions // -------------- public static final SqlFunction CAST = SqlStdOperatorTable.CAST; + public static final SqlFunction TRY_CAST = + new SqlFunction( + "TRY_CAST", + SqlKind.OTHER_FUNCTION, + ReturnTypes.ARG0_FORCE_NULLABLE, + null, + OperandTypes.ANY, + SqlFunctionCategory.SYSTEM); // --------------------- // Struct Functions diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlReturnTypes.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlReturnTypes.java index cd468108906..11570041f0b 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlReturnTypes.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlReturnTypes.java @@ -163,6 +163,27 @@ public RelDataType inferReturnType(SqlOperatorBinding opBinding) { }, SqlTypeTransforms.TO_NULLABLE); + /** Return type inference matching Flink SQL's IFNULL function. */ + public static final SqlReturnTypeInference IF_NULL = + opBinding -> { + RelDataType inputType = opBinding.getOperandType(0); + if (!inputType.isNullable()) { + return inputType; + } + + RelDataType replacementType = opBinding.getOperandType(1); + RelDataType commonType = + opBinding + .getTypeFactory() + .leastRestrictive(List.of(inputType, replacementType)); + if (commonType == null) { + return null; + } + return opBinding + .getTypeFactory() + .createTypeWithNullability(commonType, replacementType.isNullable()); + }; + public static final SqlReturnTypeInference NUMERIC_FROM_ARG1_DEFAULT1 = new NumericOrDefaultReturnTypeInference(1, 1); diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctionsTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctionsTest.java new file mode 100644 index 00000000000..ada56c00f15 --- /dev/null +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctionsTest.java @@ -0,0 +1,93 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.runtime.functions.impl; + +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.math.BigDecimal; +import java.time.LocalDateTime; +import java.time.zone.ZoneRulesException; + +class CastingFunctionsTest { + + @Test + void testTryCastSupportedTypes() { + Assertions.assertThat(CastingFunctions.tryCastToBoolean("true")).isTrue(); + Assertions.assertThat(CastingFunctions.tryCastToByte("1")).isEqualTo((byte) 1); + Assertions.assertThat(CastingFunctions.tryCastToShort("2")).isEqualTo((short) 2); + Assertions.assertThat(CastingFunctions.tryCastToInteger("3")).isEqualTo(3); + Assertions.assertThat(CastingFunctions.tryCastToLong("4")).isEqualTo(4L); + Assertions.assertThat(CastingFunctions.tryCastToFloat("5.5")).isEqualTo(5.5f); + Assertions.assertThat(CastingFunctions.tryCastToDouble("6.5")).isEqualTo(6.5d); + Assertions.assertThat(CastingFunctions.tryCastToBigDecimal("7.50", 3, 2)) + .isEqualByComparingTo(new BigDecimal("7.50")); + Assertions.assertThat(CastingFunctions.tryCastToString(8)).isEqualTo("8"); + Assertions.assertThat(CastingFunctions.tryCastToTimestamp("2024-01-02T03:04:05", "UTC")) + .isEqualTo(LocalDateTime.of(2024, 1, 2, 3, 4, 5)); + } + + @Test + void testTryCastInvalidDataReturnsNull() { + Assertions.assertThat(CastingFunctions.tryCastToInteger("invalid")).isNull(); + Assertions.assertThat(CastingFunctions.tryCastToDouble("invalid")).isNull(); + Assertions.assertThat(CastingFunctions.tryCastToBigDecimal("invalid", 10, 2)).isNull(); + Assertions.assertThat(CastingFunctions.tryCastToTimestamp("invalid-timestamp", "UTC")) + .isNull(); + Assertions.assertThat(CastingFunctions.tryCastToInteger(null)).isNull(); + Assertions.assertThat(CastingFunctions.tryCastToTimestamp(null, "UTC")).isNull(); + } + + @Test + void testTryCastDoesNotSuppressNonConversionErrors() { + Assertions.assertThatThrownBy( + () -> + CastingFunctions.tryCastToTimestamp( + "2024-01-02T03:04:05", "Invalid/Timezone")) + .isExactlyInstanceOf(ZoneRulesException.class); + Assertions.assertThatThrownBy( + () -> + CastingFunctions.tryCastToString( + new Object() { + @Override + public String toString() { + throw new IllegalStateException("internal error"); + } + })) + .isExactlyInstanceOf(IllegalStateException.class) + .hasMessage("internal error"); + } + + @Test + void testExistingCastFailureBehaviorIsUnchanged() { + int[] toStringCalls = {0}; + Object invalidTimestamp = + new Object() { + @Override + public String toString() { + toStringCalls[0]++; + return "invalid-timestamp"; + } + }; + Assertions.assertThatThrownBy( + () -> CastingFunctions.castToTimestamp(invalidTimestamp, "UTC")) + .isExactlyInstanceOf(IllegalArgumentException.class) + .hasMessage("Unable to parse given string as timestamp: invalid-timestamp"); + Assertions.assertThat(toStringCalls[0]).isEqualTo(1); + } +} diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctionsTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctionsTest.java new file mode 100644 index 00000000000..cb269eacc33 --- /dev/null +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctionsTest.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.runtime.functions.impl; + +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.math.BigDecimal; + +class LogicalFunctionsTest { + + @Test + void testIfNull() { + Assertions.assertThat(LogicalFunctions.ifNull(1, 2)).isEqualTo(1); + Assertions.assertThat(LogicalFunctions.ifNull(null, 2)).isEqualTo(2); + Assertions.assertThat((Object) LogicalFunctions.ifNull(null, null)).isNull(); + } + + @Test + void testNullIf() { + Assertions.assertThat(LogicalFunctions.nullIf(1, 1)).isNull(); + Assertions.assertThat(LogicalFunctions.nullIf(1, 2)).isEqualTo(1); + Assertions.assertThat((Object) LogicalFunctions.nullIf(null, 1)).isNull(); + Assertions.assertThat(LogicalFunctions.nullIf(1, null)).isEqualTo(1); + Assertions.assertThat(LogicalFunctions.nullIf(1, 1L)).isNull(); + Assertions.assertThat(LogicalFunctions.nullIf(new BigDecimal("1.00"), 1L)).isNull(); + Assertions.assertThat(LogicalFunctions.nullIf(1, 1.0d)).isNull(); + Assertions.assertThat(LogicalFunctions.nullIf(16_777_217, 16_777_216f)).isNull(); + Assertions.assertThat(LogicalFunctions.nullIf(new byte[] {1, 2}, new byte[] {1, 2})) + .isNull(); + } +} 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 ba837efc0b2..8c15a85ff45 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 @@ -2969,6 +2969,14 @@ void testBuildInFunctionTransform() throws Exception { testExpressionConditionTransform("round(3.1415926, 2) = 3.14"); testExpressionConditionTransform("IF(2>0,1,0) = 1"); testExpressionConditionTransform("COALESCE(null,1,2) = 1"); + testExpressionConditionTransform("TRY_CAST('invalid' AS INT) IS NULL"); + testExpressionConditionTransform("TRY_CAST('invalid-timestamp' AS TIMESTAMP) IS NULL"); + testExpressionConditionTransform("IFNULL(TRY_CAST('invalid' AS INT), 42) = 42"); + testExpressionConditionTransform("NULLIF(1, 1) IS NULL"); + testExpressionConditionTransform("NULLIF(1, 2) = 1"); + testExpressionConditionTransform("NULLIF(CAST(1 AS INT), CAST(1 AS BIGINT)) IS NULL"); + testExpressionConditionTransform("NULLIF(NULL, 1) IS NULL"); + testExpressionConditionTransform("NULLIF(1, NULL) = 1"); testExpressionConditionTransform("1 + 1 = 2"); testExpressionConditionTransform("1 - 1 = 0"); testExpressionConditionTransform("1 * 1 = 1"); diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/JaninoCompilerTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/JaninoCompilerTest.java index adf0ee1596f..3380d894a83 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/JaninoCompilerTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/JaninoCompilerTest.java @@ -364,6 +364,32 @@ void testTranslatedNestedExpressionPreservesSemantics() throws InvocationTargetE .isEqualTo(false); } + @Test + void testTranslatedNullFunctionsCompileWithJanino() throws InvocationTargetException { + List columns = List.of(Column.physicalColumn("input_value", DataTypes.STRING())); + Map columnNameMap = Map.of("input_value", "$0"); + + ExpressionEvaluator tryCastEvaluator = + compileTranslatedFilterExpression( + "IFNULL(TRY_CAST(input_value AS INT), 42) = 42", + columns, + columnNameMap, + List.of("$0"), + List.of(String.class)); + Assertions.assertThat(tryCastEvaluator.evaluate(new Object[] {"invalid"})).isEqualTo(true); + Assertions.assertThat(tryCastEvaluator.evaluate(new Object[] {"7"})).isEqualTo(false); + + ExpressionEvaluator nullIfEvaluator = + compileTranslatedFilterExpression( + "NULLIF(CAST(input_value AS INT), CAST(1 AS BIGINT)) IS NULL", + columns, + columnNameMap, + List.of("$0"), + List.of(String.class)); + Assertions.assertThat(nullIfEvaluator.evaluate(new Object[] {"1"})).isEqualTo(true); + Assertions.assertThat(nullIfEvaluator.evaluate(new Object[] {"2"})).isEqualTo(false); + } + @Test void testLargeNumericLiterals() { // Test parsing integer literals 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 ab9ed24670d..e73228fe476 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 @@ -479,7 +479,7 @@ void testTranslateFilterToJaninoExpression() { testFilterExpression( "timestampadd(year, 1, dt)", "timestampadd(\"YEAR\", 1, dt, __time_zone__)"); testFilterExpression("IF(a>b,a,b)", "isTrue(greaterThan(a, b)) ? a : b"); - testFilterExpression("NULLIF(a,b)", "nullif(a, b)"); + testFilterExpression("NULLIF(id,id)", "(java.lang.Integer) nullIf(id, id)"); testFilterExpression("COALESCE(a,b,c)", "coalesce(a, b, c)"); testFilterExpression("id + 2", "id + 2"); testFilterExpression("id - 2", "id - 2"); @@ -541,6 +541,24 @@ void testTranslateFilterToJaninoExpression() { "cast(CURRENT_TIMESTAMP as TIMESTAMP)", "castToTimestamp(currentTimestamp(__epoch_time__), __time_zone__)"); testFilterExpression("cast(dt as TIMESTAMP)", "castToTimestamp(dt, __time_zone__)"); + testFilterExpression("try_cast(id||'0' as int)", "tryCastToInteger(concat(id, \"0\"))"); + testFilterExpression("try_cast(1 as string)", "tryCastToString(1)"); + testFilterExpression("try_cast(1 as boolean)", "tryCastToBoolean(1)"); + testFilterExpression("try_cast(1 as tinyint)", "tryCastToByte(1)"); + testFilterExpression("try_cast(1 as smallint)", "tryCastToShort(1)"); + testFilterExpression("try_cast(1 as bigint)", "tryCastToLong(1)"); + testFilterExpression("try_cast(1 as float)", "tryCastToFloat(1)"); + testFilterExpression("try_cast(1 as double)", "tryCastToDouble(1)"); + testFilterExpression("try_cast(1 as decimal)", "tryCastToBigDecimal(1, 10, 0)"); + testFilterExpression("try_cast(1 as char)", "tryCastToString(1)"); + testFilterExpression("try_cast(1 as varchar)", "tryCastToString(1)"); + testFilterExpression("try_cast(dt as timestamp)", "tryCastToTimestamp(dt, __time_zone__)"); + testFilterExpression( + "try_cast(try_cast(id as int) as varchar)", + "tryCastToString(tryCastToInteger(id))"); + testFilterExpression( + "ifnull(null, 1)", + "(java.lang.Integer) ifNull(castToInteger(null), castToInteger(1))"); testFilterExpression("parse_json(jsonStr)", "parseJson(jsonStr)"); testFilterExpression("try_parse_json(jsonStr)", "tryParseJson(jsonStr)"); } @@ -881,6 +899,53 @@ void testGenerateProjectionColumns() { "Unrecognized projection expression: 1 + 1. Should be AS "); } + @Test + void testNullHelperAndTryCastTypeInference() { + List columns = + List.of( + Column.physicalColumn("text", DataTypes.STRING()), + Column.physicalColumn("nullable_int", DataTypes.INT()), + Column.physicalColumn("not_null_int", DataTypes.INT().notNull()), + Column.physicalColumn("nullable_bigint", DataTypes.BIGINT())); + + List result = + TransformParser.generateProjectionColumns( + "TRY_CAST(text AS INT) AS try_int, " + + "NULLIF(nullable_int, 1) AS nullif_int, " + + "IFNULL(nullable_int, 1) AS ifnull_not_null, " + + "IFNULL(nullable_int, nullable_bigint) AS ifnull_nullable, " + + "IFNULL(not_null_int, nullable_bigint) AS ifnull_input_not_null", + columns, + Collections.emptyList(), + new SupportedMetadataColumn[0]); + + Assertions.assertThat(result) + .extracting(ProjectionColumn::getDataType) + .containsExactly( + DataTypes.INT(), + DataTypes.INT(), + DataTypes.INT().notNull(), + DataTypes.BIGINT(), + DataTypes.INT().notNull()); + + Assertions.assertThatThrownBy( + () -> + TransformParser.generateProjectionColumns( + "IFNULL(nullable_int, DATE '2024-01-01') AS invalid", + columns, + Collections.emptyList(), + new SupportedMetadataColumn[0])) + .hasMessageContaining("Parameters must be of the same type"); + Assertions.assertThatThrownBy( + () -> + TransformParser.generateProjectionColumns( + "TRY_CAST(text AS ARRAY) AS unsupported", + columns, + Collections.emptyList(), + new SupportedMetadataColumn[0])) + .isInstanceOf(RuntimeException.class); + } + @Test public void testGenerateProjectionColumnsWithPrecision() { List testColumns = @@ -955,6 +1020,12 @@ void testGenerateReferencedColumns() { @Test void testTranslateUdfFilterToJaninoExpression() { + testFilterExpressionWithUdf( + "IFNULL(format(id), 'fallback')", + "(java.lang.String) ifNull(__instanceOfFormatFunctionClass.eval(id), \"fallback\")"); + testFilterExpressionWithUdf( + "NULLIF(format(id), '1')", + "(java.lang.String) nullIf(__instanceOfFormatFunctionClass.eval(id), \"1\")"); testFilterExpressionWithUdf( "format(upper(id))", "__instanceOfFormatFunctionClass.eval(upper(id))"); testFilterExpressionWithUdf( diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformSqlSyntaxRewriterTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformSqlSyntaxRewriterTest.java new file mode 100644 index 00000000000..7205810e29b --- /dev/null +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformSqlSyntaxRewriterTest.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.runtime.parser; + +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +class TransformSqlSyntaxRewriterTest { + + @Test + void rewritesTryCastSyntax() { + Assertions.assertThat( + TransformSqlSyntaxRewriter.rewriteTryCast( + "SELECT TRY_CAST(value AS INT) FROM t")) + .isEqualTo("SELECT TRY_CAST(CAST(value AS INT)) FROM t"); + Assertions.assertThat( + TransformSqlSyntaxRewriter.rewriteTryCast( + "SELECT try_cast /* marker */ (value AS DECIMAL(10, 2)) FROM t")) + .isEqualTo("SELECT try_cast /* marker */ (CAST(value AS DECIMAL(10, 2))) FROM t"); + Assertions.assertThat( + TransformSqlSyntaxRewriter.rewriteTryCast( + "SELECT TrY_CaSt(COALESCE(a, CAST(b AS INT)) AS VARCHAR) FROM t")) + .isEqualTo("SELECT TrY_CaSt(CAST(COALESCE(a, CAST(b AS INT)) AS VARCHAR)) FROM t"); + } + + @Test + void rewritesNestedTryCastSyntax() { + Assertions.assertThat( + TransformSqlSyntaxRewriter.rewriteTryCast( + "TRY_CAST(TRY_CAST(value AS INT) AS VARCHAR)")) + .isEqualTo("TRY_CAST(CAST(TRY_CAST(CAST(value AS INT)) AS VARCHAR))"); + } + + @Test + void ignoresTryCastInProtectedRegionsAndSimilarIdentifiers() { + String sql = + "SELECT 'TRY_CAST(a AS INT)', \"TRY_CAST(b AS INT)\", " + + "`TRY_CAST(c AS INT)`, MY_TRY_CAST(d AS INT), TRY_CASTED " + + "-- TRY_CAST(e AS INT)\n" + + "/* TRY_CAST(f AS INT) */ FROM t"; + Assertions.assertThat(TransformSqlSyntaxRewriter.rewriteTryCast(sql)).isEqualTo(sql); + } + + @Test + void preservesQuotedEscapesAndCommentsInsideTryCast() { + String sql = + "TRY_CAST(CASE WHEN value = 'it''s TRY_CAST(x AS INT)' " + + "THEN /* AS VARCHAR */ \"quoted\" ELSE `AS` END AS VARCHAR)"; + Assertions.assertThat(TransformSqlSyntaxRewriter.rewriteTryCast(sql)) + .isEqualTo( + "TRY_CAST(CAST(CASE WHEN value = 'it''s TRY_CAST(x AS INT)' " + + "THEN /* AS VARCHAR */ \"quoted\" ELSE `AS` END AS VARCHAR))"); + } + + @Test + void leavesMalformedTryCastForCalciteToReject() { + Assertions.assertThat(TransformSqlSyntaxRewriter.rewriteTryCast("TRY_CAST(value)")) + .isEqualTo("TRY_CAST(value)"); + Assertions.assertThat(TransformSqlSyntaxRewriter.rewriteTryCast("TRY_CAST(value AS INT")) + .isEqualTo("TRY_CAST(value AS INT"); + } +} From 44251f627bc836ff96dc5228cc05b66e626eaa73 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=A5=E6=A0=96?= Date: Wed, 5 Aug 2026 16:36:14 +0800 Subject: [PATCH 2/4] [FLINK-40240][runtime] Address TRY_CAST and NULLIF review comments --- .../functions/impl/CastingFunctions.java | 64 +++++++++++++++++++ .../functions/impl/LogicalFunctions.java | 15 ++--- .../functions/impl/CastingFunctionsTest.java | 13 ++++ .../functions/impl/LogicalFunctionsTest.java | 20 +++++- .../transform/PostTransformOperatorTest.java | 12 ++++ 5 files changed, 113 insertions(+), 11 deletions(-) diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctions.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctions.java index 930faf6a28e..127892586e7 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctions.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctions.java @@ -30,6 +30,7 @@ import java.time.ZoneId; import java.time.ZonedDateTime; import java.time.format.DateTimeParseException; +import java.util.Locale; /** Casting built-in functions. */ public class CastingFunctions { @@ -272,34 +273,97 @@ public static String tryCastToString(Object object) { } public static Boolean tryCastToBoolean(Object object) { + if (object instanceof String) { + switch (((String) object).toLowerCase(Locale.ROOT)) { + case "t": + case "true": + case "y": + case "yes": + case "1": + return true; + case "f": + case "false": + case "n": + case "no": + case "0": + return false; + default: + return null; + } + } return castToBoolean(object); } public static Byte tryCastToByte(Object object) { + if (object instanceof String) { + try { + return Byte.valueOf(((String) object).trim()); + } catch (NumberFormatException ignored) { + return null; + } + } return castToByte(object); } public static Short tryCastToShort(Object object) { + if (object instanceof String) { + try { + return Short.valueOf(((String) object).trim()); + } catch (NumberFormatException ignored) { + return null; + } + } return castToShort(object); } public static Integer tryCastToInteger(Object object) { + if (object instanceof String) { + try { + return Integer.valueOf(((String) object).trim()); + } catch (NumberFormatException ignored) { + return null; + } + } return castToInteger(object); } public static Long tryCastToLong(Object object) { + if (object instanceof String) { + try { + return Long.valueOf(((String) object).trim()); + } catch (NumberFormatException ignored) { + return null; + } + } return castToLong(object); } public static Float tryCastToFloat(Object object) { + if (object instanceof String) { + try { + return Float.valueOf(((String) object).trim()); + } catch (NumberFormatException ignored) { + return null; + } + } return castToFloat(object); } public static Double tryCastToDouble(Object object) { + if (object instanceof String) { + try { + return Double.valueOf(((String) object).trim()); + } catch (NumberFormatException ignored) { + return null; + } + } return castToDouble(object); } public static BigDecimal tryCastToBigDecimal(Object object, int precision, int scale) { + if (object instanceof String) { + return castToBigDecimal(((String) object).trim(), precision, scale); + } return castToBigDecimal(object, precision, scale); } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java index 273194f260a..8119e792584 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctions.java @@ -123,16 +123,11 @@ private static boolean valuesEqualForNullIf(Object value, Object comparison) { Number left = (Number) value; Number right = (Number) comparison; - if (left instanceof Double || right instanceof Double) { - double leftValue = left.doubleValue(); - double rightValue = right.doubleValue(); - if (Double.isFinite(leftValue) && Double.isFinite(rightValue)) { - return BigDecimal.valueOf(leftValue).compareTo(BigDecimal.valueOf(rightValue)) == 0; - } - return Double.compare(leftValue, rightValue) == 0; - } - if (left instanceof Float || right instanceof Float) { - return Float.compare(left.floatValue(), right.floatValue()) == 0; + if (left instanceof Double + || right instanceof Double + || left instanceof Float + || right instanceof Float) { + return left.doubleValue() == right.doubleValue(); } if (left instanceof BigDecimal || right instanceof BigDecimal) { return toBigDecimal(left).compareTo(toBigDecimal(right)) == 0; diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctionsTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctionsTest.java index ada56c00f15..d35f7bfb542 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctionsTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctionsTest.java @@ -40,10 +40,19 @@ void testTryCastSupportedTypes() { Assertions.assertThat(CastingFunctions.tryCastToString(8)).isEqualTo("8"); Assertions.assertThat(CastingFunctions.tryCastToTimestamp("2024-01-02T03:04:05", "UTC")) .isEqualTo(LocalDateTime.of(2024, 1, 2, 3, 4, 5)); + Assertions.assertThat(CastingFunctions.tryCastToBoolean("yes")).isTrue(); + Assertions.assertThat(CastingFunctions.tryCastToBoolean("no")).isFalse(); + Assertions.assertThat(CastingFunctions.tryCastToByte(128)).isEqualTo((byte) -128); + Assertions.assertThat(CastingFunctions.tryCastToInteger(1.5d)).isEqualTo(1); } @Test void testTryCastInvalidDataReturnsNull() { + Assertions.assertThat(CastingFunctions.tryCastToBoolean("invalid")).isNull(); + Assertions.assertThat(CastingFunctions.tryCastToByte("128")).isNull(); + Assertions.assertThat(CastingFunctions.tryCastToShort("32768")).isNull(); + Assertions.assertThat(CastingFunctions.tryCastToInteger("1.5")).isNull(); + Assertions.assertThat(CastingFunctions.tryCastToLong("9223372036854775808")).isNull(); Assertions.assertThat(CastingFunctions.tryCastToInteger("invalid")).isNull(); Assertions.assertThat(CastingFunctions.tryCastToDouble("invalid")).isNull(); Assertions.assertThat(CastingFunctions.tryCastToBigDecimal("invalid", 10, 2)).isNull(); @@ -75,6 +84,10 @@ public String toString() { @Test void testExistingCastFailureBehaviorIsUnchanged() { + Assertions.assertThat(CastingFunctions.castToBoolean("invalid")).isFalse(); + Assertions.assertThat(CastingFunctions.castToByte("128")).isEqualTo((byte) -128); + Assertions.assertThat(CastingFunctions.castToInteger("1.5")).isEqualTo(1); + int[] toStringCalls = {0}; Object invalidTimestamp = new Object() { diff --git a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctionsTest.java b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctionsTest.java index cb269eacc33..dc2b8f8ffa3 100644 --- a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctionsTest.java +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctionsTest.java @@ -40,8 +40,26 @@ void testNullIf() { Assertions.assertThat(LogicalFunctions.nullIf(1, 1L)).isNull(); Assertions.assertThat(LogicalFunctions.nullIf(new BigDecimal("1.00"), 1L)).isNull(); Assertions.assertThat(LogicalFunctions.nullIf(1, 1.0d)).isNull(); - Assertions.assertThat(LogicalFunctions.nullIf(16_777_217, 16_777_216f)).isNull(); + Assertions.assertThat(LogicalFunctions.nullIf(16_777_217L, 16_777_217f)) + .isEqualTo(16_777_217L); Assertions.assertThat(LogicalFunctions.nullIf(new byte[] {1, 2}, new byte[] {1, 2})) .isNull(); } + + @Test + void testNullIfFloatingPointValues() { + Assertions.assertThat(LogicalFunctions.nullIf(-0.0f, 0.0f)).isNull(); + Assertions.assertThat(LogicalFunctions.nullIf(-0.0d, 0.0d)).isNull(); + Assertions.assertThat(LogicalFunctions.nullIf(Float.NaN, Float.NaN)).isNaN(); + Assertions.assertThat(LogicalFunctions.nullIf(Double.NaN, Double.NaN)).isNaN(); + Assertions.assertThat( + LogicalFunctions.nullIf(Float.POSITIVE_INFINITY, Float.POSITIVE_INFINITY)) + .isNull(); + Assertions.assertThat( + LogicalFunctions.nullIf(Double.POSITIVE_INFINITY, Double.POSITIVE_INFINITY)) + .isNull(); + Assertions.assertThat( + LogicalFunctions.nullIf(Double.POSITIVE_INFINITY, Double.NEGATIVE_INFINITY)) + .isEqualTo(Double.POSITIVE_INFINITY); + } } 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 8c15a85ff45..169b8cdf23d 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 @@ -2970,6 +2970,9 @@ void testBuildInFunctionTransform() throws Exception { testExpressionConditionTransform("IF(2>0,1,0) = 1"); testExpressionConditionTransform("COALESCE(null,1,2) = 1"); testExpressionConditionTransform("TRY_CAST('invalid' AS INT) IS NULL"); + testExpressionConditionTransform("TRY_CAST('invalid' AS BOOLEAN) IS NULL"); + testExpressionConditionTransform("TRY_CAST('128' AS TINYINT) IS NULL"); + testExpressionConditionTransform("TRY_CAST('1.5' AS INT) IS NULL"); testExpressionConditionTransform("TRY_CAST('invalid-timestamp' AS TIMESTAMP) IS NULL"); testExpressionConditionTransform("IFNULL(TRY_CAST('invalid' AS INT), 42) = 42"); testExpressionConditionTransform("NULLIF(1, 1) IS NULL"); @@ -2977,6 +2980,15 @@ void testBuildInFunctionTransform() throws Exception { testExpressionConditionTransform("NULLIF(CAST(1 AS INT), CAST(1 AS BIGINT)) IS NULL"); testExpressionConditionTransform("NULLIF(NULL, 1) IS NULL"); testExpressionConditionTransform("NULLIF(1, NULL) = 1"); + testExpressionConditionTransform( + "NULLIF(CAST(16777217 AS BIGINT), CAST(16777217 AS FLOAT)) " + + "= CAST(16777217 AS BIGINT)"); + testExpressionConditionTransform( + "NULLIF(CAST('-0.0' AS DOUBLE), CAST('0.0' AS DOUBLE)) IS NULL"); + testExpressionConditionTransform( + "NULLIF(CAST('NaN' AS DOUBLE), CAST('NaN' AS DOUBLE)) IS NOT NULL"); + testExpressionConditionTransform( + "NULLIF(CAST('Infinity' AS DOUBLE), CAST('Infinity' AS DOUBLE)) IS NULL"); testExpressionConditionTransform("1 + 1 = 2"); testExpressionConditionTransform("1 - 1 = 0"); testExpressionConditionTransform("1 * 1 = 1"); From 6242b6cf12731253e0624ebc0c03f5785bdfb7e3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=A5=E6=A0=96?= Date: Tue, 11 Aug 2026 14:35:50 +0800 Subject: [PATCH 3/4] [FLINK-40240][runtime] Respect UDF precedence for built-in function names --- .../content.zh/docs/core-concept/transform.md | 2 + docs/content/docs/core-concept/transform.md | 2 + .../cdc/runtime/parser/JaninoCompiler.java | 28 +++++-- .../transform/PostTransformOperatorTest.java | 80 +++++++++++++++++++ .../runtime/parser/TransformParserTest.java | 68 +++++++++++++--- 5 files changed, 162 insertions(+), 18 deletions(-) diff --git a/docs/content.zh/docs/core-concept/transform.md b/docs/content.zh/docs/core-concept/transform.md index 0f9ce4df039..545f069de05 100644 --- a/docs/content.zh/docs/core-concept/transform.md +++ b/docs/content.zh/docs/core-concept/transform.md @@ -247,6 +247,8 @@ Flink CDC 使用 [Calcite](https://calcite.apache.org/) 来解析表达式并且 | IFNULL(value, replacement) | ifNull(value, replacement) | 当 value 为 NULL 时返回 replacement，否则返回 value。两个参数必须存在公共类型。仅当 replacement 可为 NULL 时，结果才可为 NULL。 | | NULLIF(value1, value2) | nullIf(value1, value2) | 当 value1 与 value2 相等时返回 NULL，否则返回 value1。返回类型为 value1 对应的可空类型。 | +`NULLIF` 会对数值参数进行跨数值类型比较。例如，`NULLIF(CAST(1 AS INT), CAST(1 AS BIGINT))` 返回 NULL。该行为与当前 Transform `=` 运算符的类型敏感相等比较不同；对于上述混合类型比较，`=` 返回 FALSE。 + ## 转换函数 你可以使用 `CAST( AS )` 语法将任何有效的表达式 `` 转换为特定类型 ``。可能的转换路径如下： diff --git a/docs/content/docs/core-concept/transform.md b/docs/content/docs/core-concept/transform.md index 005f32edbc1..5e623b27724 100644 --- a/docs/content/docs/core-concept/transform.md +++ b/docs/content/docs/core-concept/transform.md @@ -248,6 +248,8 @@ Logical functions follow SQL three-valued logic for nullable BOOLEAN values. `AN | IFNULL(value, replacement) | ifNull(value, replacement) | Returns `replacement` when `value` is NULL; otherwise, returns `value`. The arguments must have a common type. The result can be NULL only when `replacement` can be NULL. | | NULLIF(value1, value2) | nullIf(value1, value2) | Returns NULL when `value1` equals `value2`; otherwise, returns `value1`. Its return type is the nullable type of `value1`. | +`NULLIF` compares numeric operands across numeric types. For example, `NULLIF(CAST(1 AS INT), CAST(1 AS BIGINT))` returns NULL. This differs from the current type-sensitive equality behavior of the Transform `=` operator, which returns FALSE for the same mixed-type comparison. + ## Casting Functions You can use `CAST( AS )` syntax to convert any valid expression `` to a specific type ``. Possible conversion paths are: diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java index b1f352a260c..acd8f454618 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java @@ -234,7 +234,8 @@ private static Java.Rvalue translateSqlBasicCall(Context context, SqlBasicCall s if (DATE_PART_FUNCTION_UNITS.containsKey(functionName)) { return generateDatePartFunctionOperation(context, sqlBasicCall, functionName); } - if (functionName.equals("TRY_CAST")) { + if (functionName.equals("TRY_CAST") + && !findUserDefinedFunction(context, functionName).isPresent()) { return generateTryCastOperation(context, sqlBasicCall); } @@ -453,6 +454,10 @@ private static Java.Rvalue sqlBasicCallToJaninoRvalue( case GREATER_THAN_OR_EQUAL: return generateCompareOperation(context, sqlBasicCall, atoms); case NULLIF: + if (findUserDefinedFunction(context, sqlBasicCall.getOperator().getName()) + .isPresent()) { + return generateOtherFunctionOperation(context, sqlBasicCall, atoms); + } return generateNullIfOperation(context, sqlBasicCall, atoms); case CAST: return generateCastOperation(context, sqlBasicCall, atoms); @@ -564,7 +569,9 @@ private static boolean isIdentifierNullable(Context context, SqlIdentifier sqlId } private static boolean isBasicCallNullable(Context context, SqlBasicCall sqlBasicCall) { - if (sqlBasicCall.getOperator().getName().equalsIgnoreCase("IFNULL")) { + if (sqlBasicCall.getOperator().getName().equalsIgnoreCase("IFNULL") + && !findUserDefinedFunction(context, sqlBasicCall.getOperator().getName()) + .isPresent()) { List operands = sqlBasicCall.getOperandList(); if (operands.size() != 2) { return true; @@ -829,6 +836,8 @@ private static Java.Rvalue castToJavaType(DataType resultType, Java.Rvalue expre private static Java.Rvalue generateOtherFunctionOperation( Context context, SqlBasicCall sqlBasicCall, Java.Rvalue[] atoms) { String operationName = sqlBasicCall.getOperator().getName().toUpperCase(); + Optional udfFunctionOptional = + findUserDefinedFunction(context, operationName); if (operationName.equals("IF")) { if (atoms.length == 3) { return new Java.ConditionalExpression( @@ -839,15 +848,11 @@ private static Java.Rvalue generateOtherFunctionOperation( } else { throw new ParseException("Unrecognized expression: " + sqlBasicCall); } - } else if (operationName.equals("IFNULL")) { + } else if (operationName.equals("IFNULL") && !udfFunctionOptional.isPresent()) { return generateIfNullOperation(context, sqlBasicCall, atoms); - } else if (operationName.equals("NULLIF")) { + } else if (operationName.equals("NULLIF") && !udfFunctionOptional.isPresent()) { return generateNullIfOperation(context, sqlBasicCall, atoms); } else { - Optional udfFunctionOptional = - context.udfDescriptors.stream() - .filter(e -> e.getName().equalsIgnoreCase(operationName)) - .findFirst(); return udfFunctionOptional .map( udfFunction -> @@ -867,6 +872,13 @@ private static Java.Rvalue generateOtherFunctionOperation( } } + private static Optional findUserDefinedFunction( + Context context, String functionName) { + return context.udfDescriptors.stream() + .filter(udf -> udf.getName().equalsIgnoreCase(functionName)) + .findFirst(); + } + private static Java.Rvalue generateIfNullOperation( Context context, SqlBasicCall sqlBasicCall, Java.Rvalue[] atoms) { if (atoms.length != 2) { 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 169b8cdf23d..8bfa4a03ed9 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 @@ -17,6 +17,7 @@ package org.apache.flink.cdc.runtime.operators.transform; +import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.cdc.common.data.DateData; import org.apache.flink.cdc.common.data.DecimalData; import org.apache.flink.cdc.common.data.GenericArrayData; @@ -45,6 +46,8 @@ import org.junit.jupiter.api.Test; import java.math.BigDecimal; +import java.util.Collections; +import java.util.List; /** Unit tests for the {@link PostTransformOperator}. */ class PostTransformOperatorTest { @@ -469,6 +472,83 @@ void testDataChangeEventTransform() throws Exception { transformFunctionEventEventOperatorTestHarness.close(); } + @Test + void testUdfTakesPrecedenceOverBuiltInFunction() throws Exception { + Schema expectedSchema = + Schema.newBuilder() + .physicalColumn("col1", DataTypes.STRING().notNull()) + .physicalColumn("col2", DataTypes.STRING()) + .physicalColumn("col12", DataTypes.STRING()) + .physicalColumn("udf_ifnull", DataTypes.STRING()) + .physicalColumn("udf_try_cast", DataTypes.STRING()) + .physicalColumn("udf_nullif", DataTypes.STRING()) + .primaryKey("col1") + .build(); + PostTransformOperator transform = + PostTransformOperator.newBuilder() + .addTransform( + CUSTOMERS_TABLEID.identifier(), + "*, CAST(IFNULL(1, 0) AS VARCHAR) AS udf_ifnull, " + + "TRY_CAST(col1) AS udf_try_cast, " + + "NULLIF('%s', col1) AS udf_nullif", + null) + .addUdfFunctions( + List.of( + Tuple3.of( + "ifnull", + "org.apache.flink.cdc.udf.examples.java.ThrottlerFunctionClass", + Collections.emptyMap()), + Tuple3.of( + "try_cast", + "org.apache.flink.cdc.udf.examples.java.TypeOfFunctionClass", + Collections.emptyMap()), + Tuple3.of( + "nullif", + "org.apache.flink.cdc.udf.examples.java.FormatFunctionClass", + Collections.emptyMap()))) + .build(); + RegularEventOperatorTestHarness harness = + RegularEventOperatorTestHarness.with(transform, 1); + BinaryRecordDataGenerator inputGenerator = + new BinaryRecordDataGenerator((RowType) CUSTOMERS_SCHEMA.toRowDataType()); + BinaryRecordDataGenerator outputGenerator = + new BinaryRecordDataGenerator((RowType) expectedSchema.toRowDataType()); + + harness.open(); + harness.getOperator() + .processElement( + new StreamRecord<>( + new CreateTableEvent(CUSTOMERS_TABLEID, CUSTOMERS_SCHEMA))); + Assertions.assertThat(harness.getOutputRecords().poll()) + .isEqualTo( + new StreamRecord<>( + new CreateTableEvent(CUSTOMERS_TABLEID, expectedSchema))); + + DataChangeEvent inputEvent = + DataChangeEvent.insertEvent( + CUSTOMERS_TABLEID, + inputGenerator.generate( + new Object[] { + new BinaryStringData("1"), new BinaryStringData("2"), null + })); + DataChangeEvent expectedEvent = + DataChangeEvent.insertEvent( + CUSTOMERS_TABLEID, + outputGenerator.generate( + new Object[] { + new BinaryStringData("1"), + new BinaryStringData("2"), + null, + new BinaryStringData("throttled_1"), + new BinaryStringData("String: 1"), + new BinaryStringData("1") + })); + harness.getOperator().processElement(new StreamRecord<>(inputEvent)); + Assertions.assertThat(harness.getOutputRecords().poll()) + .isEqualTo(new StreamRecord<>(expectedEvent)); + harness.close(); + } + @Test void testRegexpExtractAllProjection() throws Exception { TableId tableId = TableId.tableId("my_company", "my_branch", "regexp_table"); 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 e73228fe476..8efde2e3813 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 @@ -1068,6 +1068,40 @@ void testTranslateUdfFilterToJaninoExpression() { "greaterThan(__instanceOfAddOneFunctionClass.eval(__instanceOfAddOneFunctionClass.eval(id)), 4) || !valueEquals(__instanceOfTypeOfFunctionClass.eval(id), \"bool\") && !valueEquals(__instanceOfFormatFunctionClass.eval(\"from %s to %s is %s\", \"a\", \"z\", \"lie\"), \"\")"); } + @Test + void testUdfTakesPrecedenceOverBuiltInFunction() { + List udfDescriptors = + Arrays.asList( + new UserDefinedFunctionDescriptor( + "ifnull", + "org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass"), + new UserDefinedFunctionDescriptor( + "try_cast", + "org.apache.flink.cdc.udf.examples.java.TypeOfFunctionClass"), + new UserDefinedFunctionDescriptor( + "nullif", + "org.apache.flink.cdc.udf.examples.java.FormatFunctionClass")); + + testFilterExpressionWithUdf( + "IFNULL(id)", + "__instanceOfAddOneFunctionClass.eval(id)", + DUMMY_COLUMNS, + Collections.emptyMap(), + udfDescriptors); + testFilterExpressionWithUdf( + "TRY_CAST(id)", + "__instanceOfTypeOfFunctionClass.eval(id)", + DUMMY_COLUMNS, + Collections.emptyMap(), + udfDescriptors); + testFilterExpressionWithUdf( + "NULLIF('%s', 'udf')", + "__instanceOfFormatFunctionClass.eval(\"%s\", \"udf\")", + DUMMY_COLUMNS, + Collections.emptyMap(), + udfDescriptors); + } + @Test public void testTranslateUdfFilterToJaninoExpressionWithColumnNameMap() { List columns = @@ -1287,20 +1321,34 @@ private void testFilterExpressionWithUdf( String expressionExpect, List columns, Map columnNameMap) { + testFilterExpressionWithUdf( + expression, + expressionExpect, + columns, + columnNameMap, + Arrays.asList( + new UserDefinedFunctionDescriptor( + "format", + "org.apache.flink.cdc.udf.examples.java.FormatFunctionClass"), + new UserDefinedFunctionDescriptor( + "addone", + "org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass"), + new UserDefinedFunctionDescriptor( + "typeof", + "org.apache.flink.cdc.udf.examples.java.TypeOfFunctionClass"))); + } + + private void testFilterExpressionWithUdf( + String expression, + String expressionExpect, + List columns, + Map columnNameMap, + List udfDescriptors) { String janinoExpression = TransformParser.translateFilterExpressionToJaninoExpression( expression, columns, - Arrays.asList( - new UserDefinedFunctionDescriptor( - "format", - "org.apache.flink.cdc.udf.examples.java.FormatFunctionClass"), - new UserDefinedFunctionDescriptor( - "addone", - "org.apache.flink.cdc.udf.examples.java.AddOneFunctionClass"), - new UserDefinedFunctionDescriptor( - "typeof", - "org.apache.flink.cdc.udf.examples.java.TypeOfFunctionClass")), + udfDescriptors, new SupportedMetadataColumn[0], columnNameMap); Assertions.assertThat(janinoExpression).isEqualTo(expressionExpect); From e307dcfcd0fbf977d3c639d644de9dd351437cff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=A5=E6=A0=96?= Date: Tue, 11 Aug 2026 17:47:31 +0800 Subject: [PATCH 4/4] [FLINK-40240][runtime] Centralize UDF-first function dispatch --- .../cdc/runtime/parser/JaninoCompiler.java | 68 +++++++++---------- 1 file changed, 34 insertions(+), 34 deletions(-) diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java index acd8f454618..7dda0a26e5f 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java @@ -225,6 +225,13 @@ private static Java.Rvalue translateSqlSqlLiteral(Context context, SqlLiteral sq private static Java.Rvalue translateSqlBasicCall(Context context, SqlBasicCall sqlBasicCall) { String functionName = sqlBasicCall.getOperator().getName().toUpperCase(); + Optional udfFunction = + findUserDefinedFunction(context, functionName); + if (udfFunction.isPresent()) { + return generateUserDefinedFunctionOperation( + udfFunction.get(), + translateOperands(context, sqlBasicCall).toArray(new Java.Rvalue[0])); + } if (isIntervalArithmetic(sqlBasicCall)) { return generateIntervalArithmeticOperation(context, sqlBasicCall); } @@ -234,16 +241,11 @@ private static Java.Rvalue translateSqlBasicCall(Context context, SqlBasicCall s if (DATE_PART_FUNCTION_UNITS.containsKey(functionName)) { return generateDatePartFunctionOperation(context, sqlBasicCall, functionName); } - if (functionName.equals("TRY_CAST") - && !findUserDefinedFunction(context, functionName).isPresent()) { + if (functionName.equals("TRY_CAST")) { return generateTryCastOperation(context, sqlBasicCall); } - List operandList = sqlBasicCall.getOperandList(); - List atoms = new ArrayList<>(); - for (SqlNode sqlNode : operandList) { - translateSqlNodeToAtoms(context, sqlNode, atoms); - } + List atoms = translateOperands(context, sqlBasicCall); if (TIMEZONE_FREE_TEMPORAL_FUNCTIONS.contains( sqlBasicCall.getOperator().getName().toUpperCase())) { atoms.add(new Java.AmbiguousName(Location.NOWHERE, new String[] {DEFAULT_EPOCH_TIME})); @@ -258,6 +260,14 @@ private static Java.Rvalue translateSqlBasicCall(Context context, SqlBasicCall s return sqlBasicCallToJaninoRvalue(context, sqlBasicCall, atoms.toArray(new Java.Rvalue[0])); } + private static List translateOperands(Context context, SqlBasicCall sqlBasicCall) { + List atoms = new ArrayList<>(); + for (SqlNode operand : sqlBasicCall.getOperandList()) { + translateSqlNodeToAtoms(context, operand, atoms); + } + return atoms; + } + private static boolean isIntervalArithmetic(SqlBasicCall sqlBasicCall) { if (sqlBasicCall.getKind() != SqlKind.PLUS && sqlBasicCall.getKind() != SqlKind.MINUS) { return false; @@ -454,10 +464,6 @@ private static Java.Rvalue sqlBasicCallToJaninoRvalue( case GREATER_THAN_OR_EQUAL: return generateCompareOperation(context, sqlBasicCall, atoms); case NULLIF: - if (findUserDefinedFunction(context, sqlBasicCall.getOperator().getName()) - .isPresent()) { - return generateOtherFunctionOperation(context, sqlBasicCall, atoms); - } return generateNullIfOperation(context, sqlBasicCall, atoms); case CAST: return generateCastOperation(context, sqlBasicCall, atoms); @@ -569,9 +575,10 @@ private static boolean isIdentifierNullable(Context context, SqlIdentifier sqlId } private static boolean isBasicCallNullable(Context context, SqlBasicCall sqlBasicCall) { - if (sqlBasicCall.getOperator().getName().equalsIgnoreCase("IFNULL") - && !findUserDefinedFunction(context, sqlBasicCall.getOperator().getName()) - .isPresent()) { + if (findUserDefinedFunction(context, sqlBasicCall.getOperator().getName()).isPresent()) { + return true; + } + if (sqlBasicCall.getOperator().getName().equalsIgnoreCase("IFNULL")) { List operands = sqlBasicCall.getOperandList(); if (operands.size() != 2) { return true; @@ -836,8 +843,6 @@ private static Java.Rvalue castToJavaType(DataType resultType, Java.Rvalue expre private static Java.Rvalue generateOtherFunctionOperation( Context context, SqlBasicCall sqlBasicCall, Java.Rvalue[] atoms) { String operationName = sqlBasicCall.getOperator().getName().toUpperCase(); - Optional udfFunctionOptional = - findUserDefinedFunction(context, operationName); if (operationName.equals("IF")) { if (atoms.length == 3) { return new Java.ConditionalExpression( @@ -848,30 +853,25 @@ private static Java.Rvalue generateOtherFunctionOperation( } else { throw new ParseException("Unrecognized expression: " + sqlBasicCall); } - } else if (operationName.equals("IFNULL") && !udfFunctionOptional.isPresent()) { + } else if (operationName.equals("IFNULL")) { return generateIfNullOperation(context, sqlBasicCall, atoms); - } else if (operationName.equals("NULLIF") && !udfFunctionOptional.isPresent()) { + } else if (operationName.equals("NULLIF")) { return generateNullIfOperation(context, sqlBasicCall, atoms); } else { - return udfFunctionOptional - .map( - udfFunction -> - new Java.MethodInvocation( - Location.NOWHERE, - null, - generateInvokeExpression(udfFunction), - atoms)) - .orElseGet( - () -> - new Java.MethodInvocation( - Location.NOWHERE, - null, - StringUtils.convertToCamelCase( - sqlBasicCall.getOperator().getName()), - atoms)); + return new Java.MethodInvocation( + Location.NOWHERE, + null, + StringUtils.convertToCamelCase(sqlBasicCall.getOperator().getName()), + atoms); } } + private static Java.Rvalue generateUserDefinedFunctionOperation( + UserDefinedFunctionDescriptor udfFunction, Java.Rvalue[] atoms) { + return new Java.MethodInvocation( + Location.NOWHERE, null, generateInvokeExpression(udfFunction), atoms); + } + private static Optional findUserDefinedFunction( Context context, String functionName) { return context.udfDescriptors.stream()