address review comments

Author: 春栖

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java

@@ -920,8 +920,10 @@ private GeneratedExpression translate(SqlNode sqlNode, Class<?> resultClass) {
                 return translateSqlCase((SqlCase) sqlNode, resultClass);
             }
             Java.Rvalue rvalue = translateSqlNodeToJaninoRvalue(context, sqlNode);
-            return GeneratedExpression.fromExpression(
-                    rvalue == null ? "" : rvalue.toString(), resultClass);
+            if (rvalue == null) {
+                throw new ParseException("Unrecognized expression: " + sqlNode);
+            }
+            return GeneratedExpression.fromExpression(rvalue.toString(), resultClass);
         }
 
         private GeneratedExpression translateSqlBasicCall(

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java

@@ -605,11 +605,11 @@ public static GeneratedExpression translateFilterExpressionToGeneratedExpression
             SupportedMetadataColumn[] supportedMetadataColumns,
             Map<String, String> columnNameMap) {
         if (isNullOrWhitespaceOnly(filterExpression)) {
-            return GeneratedExpression.fromExpression("", Boolean.class);
+            return GeneratedExpression.fromExpression("Boolean.TRUE", Boolean.class);
         }
         SqlSelect sqlSelect = TransformParser.parseFilterExpression(filterExpression);
         if (!sqlSelect.hasWhere()) {
-            return GeneratedExpression.fromExpression("", Boolean.class);
+            return GeneratedExpression.fromExpression("Boolean.TRUE", Boolean.class);
         }
         SqlNode where = sqlSelect.getWhere();
         return JaninoCompiler.translateSqlNodeToGeneratedExpression(

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/JaninoCompilerTest.java

@@ -17,6 +17,7 @@
 
 package org.apache.flink.cdc.runtime.parser;
 
+import org.apache.flink.api.common.io.ParseException;
 import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.cdc.common.schema.Column;
 import org.apache.flink.cdc.common.source.SupportedMetadataColumn;
@@ -26,6 +27,7 @@
 import org.apache.flink.cdc.common.types.variant.VariantTypeException;
 import org.apache.flink.cdc.runtime.operators.transform.TransformExpressionKey;
 
+import org.apache.calcite.sql.SqlNodeList;
 import org.assertj.core.api.Assertions;
 import org.codehaus.commons.compiler.CompileException;
 import org.codehaus.commons.compiler.Location;
@@ -356,6 +358,22 @@ void testTransformExpressionKeyRejectsEmptyResultTermForFullScript() {
                 .hasMessage("Generated expression result term must not be empty.");
     }
 
+    @Test
+    void testTranslateSqlNodeToGeneratedExpressionRejectsUnrecognizedSqlNode() {
+        Assertions.assertThatThrownBy(
+                        () ->
+                                JaninoCompiler.translateSqlNodeToGeneratedExpression(
+                                        JaninoCompiler.Context.of(
+                                                Collections.emptyList(),
+                                                Collections.emptyMap(),
+                                                Collections.emptyList(),
+                                                new SupportedMetadataColumn[0]),
+                                        SqlNodeList.EMPTY,
+                                        Boolean.class))
+                .isExactlyInstanceOf(ParseException.class)
+                .hasMessageStartingWith("Unrecognized expression:");
+    }
+
     @Test
     void testLargeNumericLiterals() {
         // Test parsing integer literals

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java

@@ -452,6 +452,21 @@ void testTranslateLogicalFilterToGeneratedExpressionByNullability() {
         Assertions.assertThat(generatedExpression.asScript()).endsWith("return result$0;");
     }
 
+    @Test
+    void testTranslateEmptyFilterToGeneratedExpression() {
+        GeneratedExpression generatedExpression =
+                TransformParser.translateFilterExpressionToGeneratedExpression(
+                        "",
+                        DUMMY_COLUMNS,
+                        Collections.emptyList(),
+                        new SupportedMetadataColumn[0],
+                        Collections.emptyMap());
+
+        Assertions.assertThat(generatedExpression.getCode()).isEmpty();
+        Assertions.assertThat(generatedExpression.getResultTerm()).isEqualTo("Boolean.TRUE");
+        Assertions.assertThat(generatedExpression.asScript()).isEqualTo("return Boolean.TRUE;");
+    }
+
     @Test
     public void testTranslateItemAccessToJaninoExpression() {
         // Test collection access functions (ARRAY, MAP) with proper column schema