[FLINK-40240][runtime] Support TRY_CAST, IFNULL and NULLIF in YAML Transform

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。                               |

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.                                                           |

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_

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=()}

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);
     }

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> 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());
+    }
 }