From 43873097636aeded748cf4d88a2d0592226c3436 Mon Sep 17 00:00:00 2001
From: Jia Fan <fanjiaeminem@qq.com>
Date: Thu, 29 Jan 2026 15:45:33 +0800
Subject: [PATCH 1/4] [FLINK-38996][transform] Enhance error messages for
 projection and filtering expressions

---
 .../src/test/resources/specs/casting.yaml     |  2 +-
 .../src/test/resources/specs/logical.yaml     |  2 +-
 .../operators/transform/ProjectionColumn.java |  4 ++
 .../transform/ProjectionColumnProcessor.java  |  5 ++-
 .../TransformExpressionCompiler.java          |  6 ++-
 .../transform/TransformExpressionKey.java     | 37 ++++++++++++++++---
 .../transform/TransformFilterProcessor.java   |  8 +++-
 .../transform/PostTransformOperatorTest.java  |  4 ++
 8 files changed, 57 insertions(+), 11 deletions(-)

diff --git a/flink-cdc-composer/src/test/resources/specs/casting.yaml b/flink-cdc-composer/src/test/resources/specs/casting.yaml
index 84a68e788c5..f5ce0e802a5 100644
--- a/flink-cdc-composer/src/test/resources/specs/casting.yaml
+++ b/flink-cdc-composer/src/test/resources/specs/casting.yaml
@@ -268,4 +268,4 @@
     id_
     CAST('FOOBAR' AS TIMESTAMP(6)) AS comp_11
   primary-key: id_
-  expect-error: Failed to evaluate projection expression `castToTimestamp("FOOBAR", __time_zone__)` for column `comp_11` in table `foo.bar.baz`.
+  expect-error: Failed to evaluate projection expression `CAST('FOOBAR' AS TIMESTAMP(6))` for column `comp_11` in table `foo.bar.baz`.
diff --git a/flink-cdc-composer/src/test/resources/specs/logical.yaml b/flink-cdc-composer/src/test/resources/specs/logical.yaml
index cc825f7a69c..4f09aa82301 100644
--- a/flink-cdc-composer/src/test/resources/specs/logical.yaml
+++ b/flink-cdc-composer/src/test/resources/specs/logical.yaml
@@ -45,4 +45,4 @@
     bool_ IS FALSE AS comp_8
     bool_ IS NOT FALSE AS comp_9
   primary-key: id_
-  expect-error: 'java.lang.RuntimeException: Failed to evaluate projection expression `$0 || false` for column `comp_2` in table `foo.bar.baz`.'
+  expect-error: 'java.lang.RuntimeException: Failed to evaluate projection expression ``TB`.`bool_` OR FALSE` for column `comp_2` in table `foo.bar.baz`.'
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumn.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumn.java
index 5c3b17a4396..fdbaf267851 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumn.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumn.java
@@ -87,6 +87,10 @@ public DataType getDataType() {
         return column.getType();
     }
 
+    public String getExpression() {
+        return expression;
+    }
+
     public String getScriptExpression() {
         return scriptExpression;
     }
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
index 8db8efc3126..03db01be5e6 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/ProjectionColumnProcessor.java
@@ -89,10 +89,12 @@ public Object evaluate(Object[] rowData, TransformContext context) {
             throw new RuntimeException(
                     String.format(
                             "Failed to evaluate projection expression `%s` for column `%s` in table `%s`.\n"
+                                    + "\tCompiled expression: %s\n"
                                     + "\tColumn name map: {%s}",
-                            projectionColumn.getScriptExpression(),
+                            projectionColumn.getExpression(),
                             projectionColumn.getColumnName(),
                             tableInfo.getName(),
+                            projectionColumn.getScriptExpression(),
                             projectionColumn.getColumnNameMapAsString()),
                     e);
         }
@@ -168,6 +170,7 @@ private TransformExpressionKey generateTransformExpressionKey() {
         paramTypes.add(Long.class);
 
         return TransformExpressionKey.of(
+                projectionColumn.getExpression(),
                 JaninoCompiler.loadSystemFunction(scriptExpression),
                 argumentNames,
                 paramTypes,
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java
index 4c6cb71e1f6..bc496faf92a 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java
@@ -77,7 +77,11 @@ public static ExpressionEvaluator compileExpression(
                         } catch (CompileException e) {
                             throw new InvalidProgramException(
                                     String.format(
-                                            "Expression cannot be compiled. This is a bug. Please file an issue.\n\tExpression: %s\n\tColumn name map: {%s}",
+                                            "Expression cannot be compiled. This is a bug. Please file an issue.\n"
+                                                    + "\tOriginal expression: %s\n"
+                                                    + "\tCompiled expression: %s\n"
+                                                    + "\tColumn name map: {%s}",
+                                            key.getOriginalExpression(),
                                             key.getExpression(),
                                             TransformException.prettyPrintColumnNameMap(
                                                     key.getColumnNameMap())),
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionKey.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionKey.java
index 6ece097fc9c..75cb13b665c 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionKey.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionKey.java
@@ -17,6 +17,8 @@
 
 package org.apache.flink.cdc.runtime.operators.transform;
 
+import javax.annotation.Nullable;
+
 import java.io.Serializable;
 import java.util.Collections;
 import java.util.List;
@@ -29,7 +31,8 @@
  * <p>A transform expression key contains:
  *
  * <ul>
- *   <li>expression: a string for the transformation expression.
+ *   <li>originalExpression: a string for the original transformation expression input by users.
+ *   <li>expression: a string for the compiled transformation expression.
  *   <li>argumentNames: a list for the argument names in expression.
  *   <li>argumentClasses: a list for the argument classes in expression.
  *   <li>returnClass: a class for the return class in expression
@@ -39,6 +42,7 @@
  */
 public class TransformExpressionKey implements Serializable {
     private static final long serialVersionUID = 1L;
+    @Nullable private final String originalExpression;
     private final String expression;
     private final List<String> argumentNames;
     private final List<Class<?>> argumentClasses;
@@ -46,11 +50,13 @@ public class TransformExpressionKey implements Serializable {
     private final Map<String, String> columnNameMap;
 
     private TransformExpressionKey(
+            @Nullable String originalExpression,
             String expression,
             List<String> argumentNames,
             List<Class<?>> argumentClasses,
             Class<?> returnClass,
             Map<String, String> columnNameMap) {
+        this.originalExpression = originalExpression;
         this.expression = expression;
         this.argumentNames = argumentNames;
         this.argumentClasses = argumentClasses;
@@ -58,6 +64,11 @@ private TransformExpressionKey(
         this.columnNameMap = columnNameMap;
     }
 
+    @Nullable
+    public String getOriginalExpression() {
+        return originalExpression;
+    }
+
     public String getExpression() {
         return expression;
     }
@@ -79,13 +90,19 @@ public Map<String, String> getColumnNameMap() {
     }
 
     public static TransformExpressionKey of(
+            @Nullable String originalExpression,
             String expression,
             List<String> argumentNames,
             List<Class<?>> argumentClasses,
             Class<?> returnClass,
             Map<String, String> columnNameMap) {
         return new TransformExpressionKey(
-                expression, argumentNames, argumentClasses, returnClass, columnNameMap);
+                originalExpression,
+                expression,
+                argumentNames,
+                argumentClasses,
+                returnClass,
+                columnNameMap);
     }
 
     @Override
@@ -97,7 +114,8 @@ public boolean equals(Object o) {
             return false;
         }
         TransformExpressionKey that = (TransformExpressionKey) o;
-        return expression.equals(that.expression)
+        return Objects.equals(originalExpression, that.originalExpression)
+                && expression.equals(that.expression)
                 && argumentNames.equals(that.argumentNames)
                 && argumentClasses.equals(that.argumentClasses)
                 && returnClass.equals(that.returnClass)
@@ -106,13 +124,22 @@ public boolean equals(Object o) {
 
     @Override
     public int hashCode() {
-        return Objects.hash(expression, argumentNames, argumentClasses, returnClass, columnNameMap);
+        return Objects.hash(
+                originalExpression,
+                expression,
+                argumentNames,
+                argumentClasses,
+                returnClass,
+                columnNameMap);
     }
 
     @Override
     public String toString() {
         return "TransformExpressionKey{"
-                + "expression='"
+                + "originalExpression='"
+                + originalExpression
+                + '\''
+                + ", expression='"
                 + expression
                 + '\''
                 + ", argumentNames="
diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
index 483b07973e9..73d93faca78 100644
--- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
+++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilterProcessor.java
@@ -112,10 +112,13 @@ public boolean test(Object[] preRow, Object[] postRow, TransformContext context)
         } catch (InvocationTargetException e) {
             throw new RuntimeException(
                     String.format(
-                            "Failed to evaluate filtering expression `%s` for table `%s`.\n"
+                            "Failed to evaluate filtering expression for table `%s`.\n"
+                                    + "\tOriginal expression: %s\n"
+                                    + "\tCompiled expression: %s\n"
                                     + "\tColumn name map: {%s}",
-                            transformFilter.getScriptExpression(),
                             tableInfo.getName(),
+                            transformFilter.getExpression(),
+                            transformFilter.getScriptExpression(),
                             transformFilter.getColumnNameMapAsString()),
                     e);
         }
@@ -206,6 +209,7 @@ private TransformExpressionKey generateTransformExpressionKey() {
         args.f1.add(Long.class);
 
         return TransformExpressionKey.of(
+                transformFilter.getExpression(),
                 JaninoCompiler.loadSystemFunction(transformFilter.getScriptExpression()),
                 args.f0,
                 args.f1,
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 ed49bb3e373..679c3b512aa 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
@@ -2765,6 +2765,10 @@ void testCastErrorTransform() throws Exception {
                                 + "\tcolumns={`col1` STRING NOT NULL,`castInt` INT,`castBoolean` BOOLEAN,`castTinyint` TINYINT,`castSmallint` SMALLINT,`castBigint` BIGINT,`castFloat` FLOAT,`castDouble` DOUBLE,`castChar` STRING,`castVarchar` STRING,`castDecimal` DECIMAL(4, 2),`castTimestamp` TIMESTAMP(3)}, primaryKeys=col1, options=().")
                 .cause()
                 .hasRootCauseInstanceOf(DateTimeParseException.class)
+                .hasMessageContaining(
+                        "Failed to evaluate projection expression `CAST(`TB`.`castFloat` AS TIMESTAMP(3))` for column `castTimestamp` in table `my_company.my_branch.data_cast`.\n"
+                                + "\tCompiled expression: castToTimestamp($0, __time_zone__)\n"
+                                + "\tColumn name map: {$0 -> castFloat}")
                 .hasRootCauseMessage("Text '1.0' could not be parsed at index 0");
         transformFunctionEventEventOperatorTestHarness.close();
     }

From a11d9fb4ee375a93bb14583853e4a2377183c47f Mon Sep 17 00:00:00 2001
From: Jia Fan <fanjiaeminem@qq.com>
Date: Thu, 29 Jan 2026 16:31:47 +0800
Subject: [PATCH 2/4] updte

---
 .../composer/flink/FlinkPipelineTransformITCase.java  | 11 +++++++----
 1 file changed, 7 insertions(+), 4 deletions(-)

diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
index 7390cffe62c..4a78071423a 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
@@ -2736,10 +2736,10 @@ void testTransformErrorMessage() {
                 .cause()
                 .isExactlyInstanceOf(FlinkRuntimeException.class)
                 .hasMessage(
-                        "Failed to compile expression TransformExpressionKey{expression='import static org.apache.flink.cdc.runtime.functions.SystemFunctionUtils.*;greaterThan($0, 0)', argumentNames=[__time_zone__, __epoch_time__], argumentClasses=[class java.lang.String, class java.lang.Long], returnClass=class java.lang.Boolean, columnNameMap={id1=$0}}")
+                        "Failed to compile expression TransformExpressionKey{originalExpression='id1 > 0', expression='import static org.apache.flink.cdc.runtime.functions.SystemFunctionUtils.*;greaterThan($0, 0)', argumentNames=[__time_zone__, __epoch_time__], argumentClasses=[class java.lang.String, class java.lang.Long], returnClass=class java.lang.Boolean, columnNameMap={id1=$0}}")
                 .cause()
                 .hasMessageContaining(
-                        "Expression: import static org.apache.flink.cdc.runtime.functions.SystemFunctionUtils.*;greaterThan($0, 0)")
+                        "Compiled expression: import static org.apache.flink.cdc.runtime.functions.SystemFunctionUtils.*;greaterThan($0, 0)")
                 .hasMessageContaining("Column name map: {$0 -> id1}")
                 .rootCause()
                 .isExactlyInstanceOf(CompileException.class)
@@ -2782,7 +2782,7 @@ void testTransformErrorMessage() {
                 .cause()
                 .isExactlyInstanceOf(RuntimeException.class)
                 .hasMessageContaining(
-                        "Failed to evaluate projection expression `castToInteger($0) + 1` for column `new_name` in table `default_namespace.default_schema.mytable1`")
+                        "Failed to evaluate projection expression `CAST(`TB`.`name` AS INTEGER) + 1` for column `new_name` in table `default_namespace.default_schema.mytable1`")
                 .hasMessageContaining("Column name map: {$0 -> name}");
 
         // Unsupported operations in filter rule
@@ -2803,7 +2803,10 @@ void testTransformErrorMessage() {
                 .cause()
                 .isExactlyInstanceOf(RuntimeException.class)
                 .hasMessageContaining(
-                        "Failed to evaluate filtering expression `greaterThan($0 + 1, 0)` for table `default_namespace.default_schema.mytable1`")
+                        "Failed to evaluate filtering expression for table `default_namespace.default_schema.mytable1`.\n" +
+                                "\tOriginal expression: name + 1 > 0\n" +
+                                "\tCompiled expression: greaterThan($0 + 1, 0)\n" +
+                                "\tColumn name map: {$0 -> name}")
                 .hasMessageContaining("Column name map: {$0 -> name}")
                 .rootCause()
                 .isExactlyInstanceOf(RuntimeException.class)

From a24d9fe5d4cf2476b99ee495e97cd3bd979a66ad Mon Sep 17 00:00:00 2001
From: Jia Fan <fanjiaeminem@qq.com>
Date: Thu, 29 Jan 2026 16:32:01 +0800
Subject: [PATCH 3/4] update

---
 .../cdc/composer/flink/FlinkPipelineTransformITCase.java  | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
index 4a78071423a..c69ad55a73d 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
@@ -2803,10 +2803,10 @@ void testTransformErrorMessage() {
                 .cause()
                 .isExactlyInstanceOf(RuntimeException.class)
                 .hasMessageContaining(
-                        "Failed to evaluate filtering expression for table `default_namespace.default_schema.mytable1`.\n" +
-                                "\tOriginal expression: name + 1 > 0\n" +
-                                "\tCompiled expression: greaterThan($0 + 1, 0)\n" +
-                                "\tColumn name map: {$0 -> name}")
+                        "Failed to evaluate filtering expression for table `default_namespace.default_schema.mytable1`.\n"
+                                + "\tOriginal expression: name + 1 > 0\n"
+                                + "\tCompiled expression: greaterThan($0 + 1, 0)\n"
+                                + "\tColumn name map: {$0 -> name}")
                 .hasMessageContaining("Column name map: {$0 -> name}")
                 .rootCause()
                 .isExactlyInstanceOf(RuntimeException.class)

From a43d5bbd211f08ec42aa016f294b6dc42279177c Mon Sep 17 00:00:00 2001
From: Jia Fan <fanjiaeminem@qq.com>
Date: Fri, 30 Jan 2026 14:28:55 +0800
Subject: [PATCH 4/4] update

---
 .../cdc/composer/flink/FlinkPipelineTransformITCase.java     | 4 ++--
 .../operators/transform/PostTransformOperatorTest.java       | 5 ++---
 2 files changed, 4 insertions(+), 5 deletions(-)

diff --git a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
index b6d02724b87..c81b1ba463b 100644
--- a/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
+++ b/flink-cdc-composer/src/test/java/org/apache/flink/cdc/composer/flink/FlinkPipelineTransformITCase.java
@@ -2737,12 +2737,12 @@ void testTransformErrorMessage() {
                 .cause()
                 .isExactlyInstanceOf(FlinkRuntimeException.class)
                 .hasMessage(
-                        "Failed to compile expression TransformExpressionKey{expression='"
+                        "Failed to compile expression TransformExpressionKey{originalExpression='id1 > 0', expression='"
                                 + JaninoCompiler.LOAD_MODULES_EXPRESSION
                                 + "greaterThan($0, 0)', argumentNames=[__time_zone__, __epoch_time__], argumentClasses=[class java.lang.String, class java.lang.Long], returnClass=class java.lang.Boolean, columnNameMap={id1=$0}}")
                 .cause()
                 .hasMessageContaining(
-                        "Expression: "
+                        "Compiled expression: "
                                 + JaninoCompiler.LOAD_MODULES_EXPRESSION
                                 + "greaterThan($0, 0)")
                 .hasMessageContaining("Column name map: {$0 -> id1}")
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 83033412ace..e50ed0ca7f8 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
@@ -2767,13 +2767,12 @@ void testCastErrorTransform() throws Exception {
                                 + "\tcolumns={`col1` STRING NOT NULL,`castInt` INT,`castBoolean` BOOLEAN,`castTinyint` TINYINT,`castSmallint` SMALLINT,`castBigint` BIGINT,`castFloat` FLOAT,`castDouble` DOUBLE,`castChar` STRING,`castVarchar` STRING,`castDecimal` DECIMAL(4, 2),`castTimestamp` TIMESTAMP(3)}, primaryKeys=col1, options=().")
                 .cause()
                 .hasRootCauseInstanceOf(IllegalArgumentException.class)
-                .hasRootCauseMessage("Unable to parse given string as timestamp: 1.0");
-                .hasRootCauseInstanceOf(DateTimeParseException.class)
                 .hasMessageContaining(
                         "Failed to evaluate projection expression `CAST(`TB`.`castFloat` AS TIMESTAMP(3))` for column `castTimestamp` in table `my_company.my_branch.data_cast`.\n"
                                 + "\tCompiled expression: castToTimestamp($0, __time_zone__)\n"
                                 + "\tColumn name map: {$0 -> castFloat}")
-                .hasRootCauseMessage("Text '1.0' could not be parsed at index 0");
+                .hasRootCauseMessage("Unable to parse given string as timestamp: 1.0");
+
         transformFunctionEventEventOperatorTestHarness.close();
     }