fix empty generated expression script

Author: 春栖

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionCompiler.java

@@ -76,16 +76,17 @@ public static TransformExpressionEvaluator compileExpression(
                         // Result type
                         scriptEvaluator.setReturnType(key.getReturnClass());
 
+                        String fullScript = key.getFullScript();
                         if (LOG.isDebugEnabled()) {
-                            LOG.debug("Going to evaluate script: {}", key.getFullScript());
+                            LOG.debug("Going to evaluate script: {}", fullScript);
                             LOG.debug("  - Argument names: {}", argumentNames);
                             LOG.debug("  - Argument types: {}", argumentClasses);
                             LOG.debug("  - Returns: {}", key.getReturnClass());
                         }
 
                         try {
                             // Compile
-                            scriptEvaluator.cook(key.getFullScript());
+                            scriptEvaluator.cook(fullScript);
                         } catch (CompileException e) {
                             throw new InvalidProgramException(
                                     String.format(

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionKey.java

@@ -78,6 +78,7 @@ public String getCompiledScript() {
     }
 
     public String getFullScript() {
+        validateGeneratedExpression();
         return JaninoCompiler.loadSystemFunction(getCompiledScript());
     }
 
@@ -97,6 +98,13 @@ public Map<String, String> getColumnNameMap() {
         return Collections.unmodifiableMap(columnNameMap);
     }
 
+    private void validateGeneratedExpression() {
+        if (generatedExpression.getResultTerm().trim().isEmpty()) {
+            throw new IllegalArgumentException(
+                    "Generated expression result term must not be empty.");
+        }
+    }
+
     public static TransformExpressionKey of(
             @Nullable String originalExpression,
             String compiledExpression,

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

@@ -24,6 +24,7 @@
 import org.apache.flink.cdc.common.types.variant.BinaryVariantInternalBuilder;
 import org.apache.flink.cdc.common.types.variant.Variant;
 import org.apache.flink.cdc.common.types.variant.VariantTypeException;
+import org.apache.flink.cdc.runtime.operators.transform.TransformExpressionKey;
 
 import org.assertj.core.api.Assertions;
 import org.codehaus.commons.compiler.CompileException;
@@ -339,6 +340,22 @@ void testGeneratedLogicalExpressionCompilesWithJanino() throws Exception {
                 .isNull();
     }
 
+    @Test
+    void testTransformExpressionKeyRejectsEmptyResultTermForFullScript() {
+        TransformExpressionKey key =
+                TransformExpressionKey.of(
+                        null,
+                        GeneratedExpression.fromExpression("", Boolean.class),
+                        Collections.emptyList(),
+                        Collections.emptyList(),
+                        Collections.emptyMap());
+
+        Assertions.assertThatCode(key::toString).doesNotThrowAnyException();
+        Assertions.assertThatThrownBy(key::getFullScript)
+                .isExactlyInstanceOf(IllegalArgumentException.class)
+                .hasMessage("Generated expression result term must not be empty.");
+    }
+
     @Test
     void testLargeNumericLiterals() {
         // Test parsing integer literals