This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new fdacb41de [FLINK-40240][runtime] Support TRY_CAST, IFNULL and NULLIF
in YAML Transform (#4492)
fdacb41de is described below
commit fdacb41de9d1a3df587f3adec74eac6e1f42e016
Author: haruki <[email protected]>
AuthorDate: Tue Aug 11 19:42:30 2026 +0800
[FLINK-40240][runtime] Support TRY_CAST, IFNULL and NULLIF in YAML
Transform (#4492)
Co-authored-by: 春栖 <[email protected]>
---
docs/content.zh/docs/core-concept/transform.md | 6 +
docs/content/docs/core-concept/transform.md | 6 +
.../src/test/resources/specs/casting.yaml | 16 ++
.../src/test/resources/specs/condition.yaml | 15 ++
.../runtime/functions/impl/CastingFunctions.java | 114 +++++++++-
.../runtime/functions/impl/LogicalFunctions.java | 38 ++++
.../flink/cdc/runtime/parser/JaninoCompiler.java | 232 +++++++++++++++++----
.../flink/cdc/runtime/parser/TransformParser.java | 15 +-
.../runtime/parser/TransformSqlSyntaxRewriter.java | 210 +++++++++++++++++++
.../parser/metadata/TransformSqlOperatorTable.java | 24 +++
.../parser/metadata/TransformSqlReturnTypes.java | 21 ++
.../functions/impl/CastingFunctionsTest.java | 106 ++++++++++
.../functions/impl/LogicalFunctionsTest.java | 65 ++++++
.../transform/PostTransformOperatorTest.java | 100 +++++++++
.../cdc/runtime/parser/JaninoCompilerTest.java | 26 +++
.../cdc/runtime/parser/TransformParserTest.java | 141 ++++++++++++-
.../parser/TransformSqlSyntaxRewriterTest.java | 77 +++++++
17 files changed, 1157 insertions(+), 55 deletions(-)
diff --git a/docs/content.zh/docs/core-concept/transform.md
b/docs/content.zh/docs/core-concept/transform.md
index d6daab547..545f069de 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( <EXPR> AS <T> )` 语法将任何有效的表达式 `<EXPR>` 转换为特定类型 `<T>`。可能的转换路径如下：
+`TRY_CAST( <EXPR> AS <T> )` 支持与 `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 28bda0e34..5e623b277 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( <EXPR> AS <T> )` syntax to convert any valid expression
`<EXPR>` to a specific type `<T>`. Possible conversion paths are:
+`TRY_CAST( <EXPR> AS <T> )` 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 f5ce0e802..288b8df16 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 d220ffa86..bf2d98fa4 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 d895ea177..127892586 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.LocalDateTime;
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 class CastingFunctions {
}
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 class CastingFunctions {
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 07366bd34..8119e7925 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 class LogicalFunctions {
}
return null;
}
+
+ public static <T> T ifNull(T value, T replacement) {
+ return value != null ? value : replacement;
+ }
+
+ public static <T> 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 6f8d0e464..7dda0a26e 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.schema.Column;
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 @@ public class JaninoCompiler {
private static Java.Rvalue translateSqlBasicCall(Context context,
SqlBasicCall sqlBasicCall) {
String functionName =
sqlBasicCall.getOperator().getName().toUpperCase();
+ Optional<UserDefinedFunctionDescriptor> 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 @@ public class JaninoCompiler {
if (DATE_PART_FUNCTION_UNITS.containsKey(functionName)) {
return generateDatePartFunctionOperation(context, sqlBasicCall,
functionName);
}
-
- List<SqlNode> operandList = sqlBasicCall.getOperandList();
- List<Java.Rvalue> atoms = new ArrayList<>();
- for (SqlNode sqlNode : operandList) {
- translateSqlNodeToAtoms(context, sqlNode, atoms);
+ if (functionName.equals("TRY_CAST")) {
+ return generateTryCastOperation(context, sqlBasicCall);
}
+
+ List<Java.Rvalue> 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 @@ public class JaninoCompiler {
return sqlBasicCallToJaninoRvalue(context, sqlBasicCall,
atoms.toArray(new Java.Rvalue[0]));
}
+ private static List<Java.Rvalue> translateOperands(Context context,
SqlBasicCall sqlBasicCall) {
+ List<Java.Rvalue> 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 @@ public class JaninoCompiler {
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 @@ public class JaninoCompiler {
}
private static boolean isBasicCallNullable(Context context, SqlBasicCall
sqlBasicCall) {
+ if (findUserDefinedFunction(context,
sqlBasicCall.getOperator().getName()).isPresent()) {
+ return true;
+ }
+ if (sqlBasicCall.getOperator().getName().equalsIgnoreCase("IFNULL")) {
+ List<SqlNode> 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 @@ public class JaninoCompiler {
}
List<SqlNode> 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<SqlNode> 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<SqlNode> 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 @@ public class JaninoCompiler {
// 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 @@ public class JaninoCompiler {
new Java.Annotation[0],
canonicalName.split("\\."),
null),
- methodInvocation);
+ expression);
}
}
- return methodInvocation;
+ return expression;
}
private static Java.Rvalue generateOtherFunctionOperation(
@@ -795,27 +853,100 @@ public class JaninoCompiler {
} 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<UserDefinedFunctionDescriptor> 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<UserDefinedFunctionDescriptor>
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 @@ public class JaninoCompiler {
}
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 @@ public class JaninoCompiler {
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<Java.Rvalue> timestampAtoms = new
ArrayList<>(Arrays.asList(atoms));
timestampAtoms.add(
@@ -917,7 +1071,7 @@ public class JaninoCompiler {
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 4912ef5a5..6020245ba 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 class TransformParser {
} else {
List<String> originalColumnNames =
parseColumnNameList(exprNode);
Map<String, String> 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 000000000..7b70679fb
--- /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 096e3fa2b..ff6ca6803 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 class TransformSqlOperatorTable extends
ReflectiveSqlOperatorTable {
// ---------------------
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 class TransformSqlOperatorTable extends
ReflectiveSqlOperatorTable {
// 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 cd4681089..11570041f 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 class TransformSqlReturnTypes {
},
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 000000000..d35f7bfb5
--- /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 000000000..dc2b8f8ff
--- /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 ba837efc0..8bfa4a03e 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.assertj.core.api.Assertions;
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 @@ class PostTransformOperatorTest {
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<PostTransformOperator, Event> 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 @@ class PostTransformOperatorTest {
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 adf0ee159..3380d894a 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 @@ class JaninoCompilerTest {
.isEqualTo(false);
}
+ @Test
+ void testTranslatedNullFunctionsCompileWithJanino() throws
InvocationTargetException {
+ List<Column> columns = List.of(Column.physicalColumn("input_value",
DataTypes.STRING()));
+ Map<String, String> 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 ab9ed2467..8efde2e38 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 @@ class TransformParserTest {
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 @@ class TransformParserTest {
"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 @@ class TransformParserTest {
"Unrecognized projection expression: 1 + 1. Should be
<EXPR> AS <IDENTIFIER>");
}
+ @Test
+ void testNullHelperAndTryCastTypeInference() {
+ List<Column> 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<ProjectionColumn> 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<INT>) AS
unsupported",
+ columns,
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]))
+ .isInstanceOf(RuntimeException.class);
+ }
+
@Test
public void testGenerateProjectionColumnsWithPrecision() {
List<Column> testColumns =
@@ -955,6 +1020,12 @@ class TransformParserTest {
@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 @@ class TransformParserTest {
"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<UserDefinedFunctionDescriptor> 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<Column> columns =
@@ -1216,20 +1321,34 @@ class TransformParserTest {
String expressionExpect,
List<Column> columns,
Map<String, String> 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<Column> columns,
+ Map<String, String> columnNameMap,
+ List<UserDefinedFunctionDescriptor> 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 000000000..7205810e2
--- /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");
+ }
+}
