diff --git a/docs/content.zh/docs/core-concept/transform.md b/docs/content.zh/docs/core-concept/transform.md index d6daab54713..545f069de05 100644 --- a/docs/content.zh/docs/core-concept/transform.md +++ b/docs/content.zh/docs/core-concept/transform.md @@ -244,11 +244,17 @@ 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 对应的可空类型。 | + +`NULLIF` 会对数值参数进行跨数值类型比较。例如，`NULLIF(CAST(1 AS INT), CAST(1 AS BIGINT))` 返回 NULL。该行为与当前 Transform `=` 运算符的类型敏感相等比较不同；对于上述混合类型比较，`=` 返回 FALSE。 ## 转换函数 你可以使用 `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..5e623b27724 100644 --- a/docs/content/docs/core-concept/transform.md +++ b/docs/content/docs/core-concept/transform.md @@ -245,11 +245,17 @@ 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`. | + +`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: +`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..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 { @@ -264,7 +265,114 @@ 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) { + 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); + } + + 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 +410,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..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 @@ -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,40 @@ 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 + || 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; + } + 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..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 @@ -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; @@ -224,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); } @@ -233,12 +241,11 @@ private static Java.Rvalue translateSqlBasicCall(Context context, SqlBasicCall s if (DATE_PART_FUNCTION_UNITS.containsKey(functionName)) { return generateDatePartFunctionOperation(context, sqlBasicCall, functionName); } - - List operandList = sqlBasicCall.getOperandList(); - List atoms = new ArrayList<>(); - for (SqlNode sqlNode : operandList) { - translateSqlNodeToAtoms(context, sqlNode, atoms); + if (functionName.equals("TRY_CAST")) { + return generateTryCastOperation(context, sqlBasicCall); } + + 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})); @@ -253,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; @@ -448,6 +463,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 +575,17 @@ private static boolean isIdentifierNullable(Context context, SqlIdentifier sqlId } private static boolean isBasicCallNullable(Context context, SqlBasicCall sqlBasicCall) { + 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; + } + return isExpressionNullable(context, operands.get(0)) + && isExpressionNullable(context, operands.get(1)); + } switch (sqlBasicCall.getKind()) { case AND: case OR: @@ -636,7 +664,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 +819,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 +834,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,27 +853,100 @@ 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() - .filter(e -> e.getName().equalsIgnoreCase(operationName)) - .findFirst(); - 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() + .filter(udf -> udf.getName().equalsIgnoreCase(functionName)) + .findFirst(); + } + + 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; } } @@ -865,22 +996,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 +1054,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 +1071,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..d35f7bfb542 --- /dev/null +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/CastingFunctionsTest.java @@ -0,0 +1,106 @@ +/* + * 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)); + 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(); + 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() { + 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() { + @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..dc2b8f8ffa3 --- /dev/null +++ b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/LogicalFunctionsTest.java @@ -0,0 +1,65 @@ +/* + * 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_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 ba837efc0b2..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"); @@ -2969,6 +3049,26 @@ 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' 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"); + 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( + "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"); 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..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 @@ -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( @@ -997,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 = @@ -1216,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); 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"); + } +}