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 9a774642e2f..23c60fdb4e1 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 @@ -2656,9 +2656,9 @@ void testTransformErrorMessage() { .cause() .isExactlyInstanceOf(FlinkRuntimeException.class) .hasMessage( - "Failed to compile expression TransformExpressionKey{originalExpression='id1 > 0', compiledExpression='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', compiledScript='return 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("Compiled expression: greaterThan($0, 0)") + .hasMessageContaining("Compiled script: return greaterThan($0, 0);") .hasMessageContaining("Column name map: {$0 -> id1}") .rootCause() .isExactlyInstanceOf(CompileException.class) @@ -2724,7 +2724,7 @@ void testTransformErrorMessage() { .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" + + "\tCompiled script: return greaterThan($0 + 1, 0);\n" + "\tColumn name map: {$0 -> name}") .rootCause() .isExactlyInstanceOf(RuntimeException.class) 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 fdbaf267851..df655794267 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 @@ -17,10 +17,12 @@ package org.apache.flink.cdc.runtime.operators.transform; +import org.apache.flink.cdc.common.converter.JavaClassConverter; import org.apache.flink.cdc.common.schema.Column; import org.apache.flink.cdc.common.types.DataType; import org.apache.flink.cdc.common.utils.StringUtils; import org.apache.flink.cdc.runtime.operators.transform.exceptions.TransformException; +import org.apache.flink.cdc.runtime.parser.GeneratedExpression; import java.io.Serializable; import java.util.ArrayList; @@ -39,8 +41,7 @@ * @@ -49,19 +50,19 @@ public class ProjectionColumn implements Serializable { private static final long serialVersionUID = 1L; private final Column column; private final String expression; - private final String scriptExpression; + private final GeneratedExpression generatedExpression; private final List originalColumnNames; private final Map columnNameMap; public ProjectionColumn( Column column, String expression, - String scriptExpression, + GeneratedExpression generatedExpression, List originalColumnNames, Map columnNameMap) { this.column = column; this.expression = expression; - this.scriptExpression = scriptExpression; + this.generatedExpression = generatedExpression; this.originalColumnNames = originalColumnNames; this.columnNameMap = columnNameMap; } @@ -70,7 +71,7 @@ public ProjectionColumn copy() { return new ProjectionColumn( column.copy(column.getName()), expression, - scriptExpression, + generatedExpression, new ArrayList<>(originalColumnNames), new HashMap<>(columnNameMap)); } @@ -92,7 +93,15 @@ public String getExpression() { } public String getScriptExpression() { - return scriptExpression; + return getCompiledScript(); + } + + public GeneratedExpression getGeneratedExpression() { + return generatedExpression; + } + + public String getCompiledScript() { + return generatedExpression.asScript(); } public List getOriginalColumnNames() { @@ -108,7 +117,7 @@ public String getColumnNameMapAsString() { } public boolean isValidTransformedProjectionColumn() { - return !StringUtils.isNullOrWhitespaceOnly(scriptExpression); + return !StringUtils.isNullOrWhitespaceOnly(generatedExpression.getResultTerm()); } /** @@ -120,7 +129,12 @@ public static ProjectionColumn ofForwarded(Column column, String mappedColumnNam String name = column.getName(); Map columnNameMap = Collections.singletonMap(name, mappedColumnName); return new ProjectionColumn( - column, name, mappedColumnName, Collections.singletonList(name), columnNameMap); + column, + name, + GeneratedExpression.fromExpression( + mappedColumnName, JavaClassConverter.toJavaClass(column.getType())), + Collections.singletonList(name), + columnNameMap); } /** @@ -136,7 +150,8 @@ public static ProjectionColumn ofAliased( return new ProjectionColumn( column.copy(newName), originalName, - mappedColumnName, + GeneratedExpression.fromExpression( + mappedColumnName, JavaClassConverter.toJavaClass(column.getType())), Collections.singletonList(originalName), columnNameMap); } @@ -150,13 +165,13 @@ public static ProjectionColumn ofCalculated( String columnName, DataType dataType, String expression, - String scriptExpression, + GeneratedExpression generatedExpression, List originalColumnNames, Map columnNameMap) { return new ProjectionColumn( Column.physicalColumn(columnName, dataType), expression, - scriptExpression, + generatedExpression, originalColumnNames, columnNameMap); } @@ -169,8 +184,8 @@ public String toString() { + ", expression='" + expression + '\'' - + ", scriptExpression='" - + scriptExpression + + ", compiledScript='" + + getCompiledScript().replace("\n", "\\n") + '\'' + ", originalColumnNames=" + originalColumnNames 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 db5c37ca258..cbf9af7161b 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 @@ -22,8 +22,6 @@ import org.apache.flink.cdc.common.source.SupportedMetadataColumn; import org.apache.flink.cdc.runtime.parser.JaninoCompiler; -import org.codehaus.janino.ExpressionEvaluator; - import java.lang.reflect.InvocationTargetException; import java.util.ArrayList; import java.util.LinkedHashSet; @@ -45,7 +43,7 @@ public class ProjectionColumnProcessor { private final TransformExpressionKey transformExpressionKey; private final Map supportedMetadataColumns; private final List udfFunctionInstances; - private final ExpressionEvaluator expressionEvaluator; + private final TransformExpressionEvaluator expressionEvaluator; public ProjectionColumnProcessor( PostTransformChangeInfo tableInfo, @@ -89,12 +87,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" + + "\tCompiled script: %s\n" + "\tColumn name map: {%s}", projectionColumn.getExpression(), projectionColumn.getColumnName(), tableInfo.getName(), - projectionColumn.getScriptExpression(), + projectionColumn.getCompiledScript(), projectionColumn.getColumnNameMapAsString()), e); } @@ -170,10 +168,9 @@ private TransformExpressionKey generateTransformExpressionKey() { return TransformExpressionKey.of( projectionColumn.getExpression(), - projectionColumn.getScriptExpression(), + projectionColumn.getGeneratedExpression(), argumentNames, paramTypes, - JavaClassConverter.toJavaClass(projectionColumn.getDataType()), columnNameMap); } } 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 1085a7df82d..061b298a72f 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 @@ -25,7 +25,7 @@ import org.apache.flink.shaded.guava31.com.google.common.cache.CacheBuilder; import org.codehaus.commons.compiler.CompileException; -import org.codehaus.janino.ExpressionEvaluator; +import org.codehaus.janino.ScriptEvaluator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -40,8 +40,8 @@ public class TransformExpressionCompiler { private static final Logger LOG = LoggerFactory.getLogger(TransformExpressionCompiler.class); - static final Cache COMPILED_EXPRESSION_CACHE = - CacheBuilder.newBuilder().softValues().build(); + static final Cache + COMPILED_EXPRESSION_CACHE = CacheBuilder.newBuilder().softValues().build(); /** Triggers internal garbage collection of expired cache entries. */ public static void cleanUp() { @@ -51,14 +51,14 @@ public static void cleanUp() { COMPILED_EXPRESSION_CACHE.invalidateAll(); } - /** Compiles an expression code to a janino {@link ExpressionEvaluator}. */ - public static ExpressionEvaluator compileExpression( + /** Compiles an expression script to a janino {@link ScriptEvaluator}. */ + public static TransformExpressionEvaluator compileExpression( TransformExpressionKey key, List udfDescriptors) { try { return COMPILED_EXPRESSION_CACHE.get( key, () -> { - ExpressionEvaluator expressionEvaluator = new ExpressionEvaluator(); + ScriptEvaluator scriptEvaluator = new ScriptEvaluator(); List argumentNames = new ArrayList<>(key.getArgumentNames()); List> argumentClasses = new ArrayList<>(key.getArgumentClasses()); @@ -69,15 +69,16 @@ public static ExpressionEvaluator compileExpression( } // Input args - expressionEvaluator.setParameters( + scriptEvaluator.setParameters( argumentNames.toArray(new String[0]), argumentClasses.toArray(new Class[0])); // Result type - expressionEvaluator.setExpressionType(key.getReturnClass()); + scriptEvaluator.setReturnType(key.getReturnClass()); + String fullScript = key.getFullScript(); if (LOG.isDebugEnabled()) { - LOG.debug("Going to evaluate expression: {}", key.getFullExpression()); + LOG.debug("Going to evaluate script: {}", fullScript); LOG.debug(" - Argument names: {}", argumentNames); LOG.debug(" - Argument types: {}", argumentClasses); LOG.debug(" - Returns: {}", key.getReturnClass()); @@ -85,21 +86,21 @@ public static ExpressionEvaluator compileExpression( try { // Compile - expressionEvaluator.cook(key.getFullExpression()); + scriptEvaluator.cook(fullScript); } catch (CompileException e) { throw new InvalidProgramException( String.format( "Expression cannot be compiled. This is a bug. Please file an issue.\n" + "\tOriginal expression: %s\n" - + "\tCompiled expression: %s\n" + + "\tCompiled script: %s\n" + "\tColumn name map: {%s}", key.getOriginalExpression(), - key.getCompiledExpression(), + key.getCompiledScript(), TransformException.prettyPrintColumnNameMap( key.getColumnNameMap())), e); } - return expressionEvaluator; + return new TransformExpressionEvaluator(scriptEvaluator); }); } catch (Exception e) { throw new FlinkRuntimeException("Failed to compile expression " + key, e); diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionEvaluator.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionEvaluator.java new file mode 100644 index 00000000000..1a002b9925a --- /dev/null +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformExpressionEvaluator.java @@ -0,0 +1,37 @@ +/* + * 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.operators.transform; + +import org.codehaus.janino.ScriptEvaluator; + +import java.lang.reflect.InvocationTargetException; +import java.util.Objects; + +/** Evaluates a compiled transform expression script. */ +public class TransformExpressionEvaluator { + + private final ScriptEvaluator scriptEvaluator; + + TransformExpressionEvaluator(ScriptEvaluator scriptEvaluator) { + this.scriptEvaluator = Objects.requireNonNull(scriptEvaluator); + } + + public Object evaluate(Object[] params) throws InvocationTargetException { + return scriptEvaluator.evaluate(params); + } +} 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 5a15662e551..5d1cf489b79 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,7 @@ package org.apache.flink.cdc.runtime.operators.transform; +import org.apache.flink.cdc.runtime.parser.GeneratedExpression; import org.apache.flink.cdc.runtime.parser.JaninoCompiler; import javax.annotation.Nullable; @@ -34,7 +35,7 @@ * *
    *
  • originalExpression: a string for the original transformation expression input by users. - *
  • expression: a string for the compiled transformation expression. + *
  • generatedExpression: statement-level code generated from the original expression. *
  • argumentNames: a list for the argument names in expression. *
  • argumentClasses: a list for the argument classes in expression. *
  • returnClass: a class for the return class in expression @@ -45,24 +46,21 @@ public class TransformExpressionKey implements Serializable { private static final long serialVersionUID = 1L; @Nullable private final String originalExpression; - private final String compiledExpression; + private final GeneratedExpression generatedExpression; private final List argumentNames; private final List> argumentClasses; - private final Class returnClass; private final Map columnNameMap; private TransformExpressionKey( @Nullable String originalExpression, - String compiledExpression, + GeneratedExpression generatedExpression, List argumentNames, List> argumentClasses, - Class returnClass, Map columnNameMap) { this.originalExpression = originalExpression; - this.compiledExpression = compiledExpression; + this.generatedExpression = generatedExpression; this.argumentNames = argumentNames; this.argumentClasses = argumentClasses; - this.returnClass = returnClass; this.columnNameMap = columnNameMap; } @@ -71,12 +69,17 @@ public String getOriginalExpression() { return originalExpression; } - public String getCompiledExpression() { - return compiledExpression; + public GeneratedExpression getGeneratedExpression() { + return generatedExpression; } - public String getFullExpression() { - return JaninoCompiler.loadSystemFunction(compiledExpression); + public String getCompiledScript() { + return generatedExpression.asScript(); + } + + public String getFullScript() { + validateGeneratedExpression(); + return JaninoCompiler.loadSystemFunction(getCompiledScript()); } public List getArgumentNames() { @@ -88,13 +91,20 @@ public List> getArgumentClasses() { } public Class getReturnClass() { - return returnClass; + return generatedExpression.getResultClass(); } public Map 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, @@ -104,10 +114,23 @@ public static TransformExpressionKey of( Map columnNameMap) { return new TransformExpressionKey( originalExpression, - compiledExpression, + GeneratedExpression.fromExpression(compiledExpression, returnClass), + argumentNames, + argumentClasses, + columnNameMap); + } + + public static TransformExpressionKey of( + @Nullable String originalExpression, + GeneratedExpression generatedExpression, + List argumentNames, + List> argumentClasses, + Map columnNameMap) { + return new TransformExpressionKey( + originalExpression, + generatedExpression, argumentNames, argumentClasses, - returnClass, columnNameMap); } @@ -121,10 +144,9 @@ public boolean equals(Object o) { } TransformExpressionKey that = (TransformExpressionKey) o; return Objects.equals(originalExpression, that.originalExpression) - && compiledExpression.equals(that.compiledExpression) + && generatedExpression.equals(that.generatedExpression) && argumentNames.equals(that.argumentNames) && argumentClasses.equals(that.argumentClasses) - && returnClass.equals(that.returnClass) && columnNameMap.equals(that.columnNameMap); } @@ -132,10 +154,9 @@ public boolean equals(Object o) { public int hashCode() { return Objects.hash( originalExpression, - compiledExpression, + generatedExpression, argumentNames, argumentClasses, - returnClass, columnNameMap); } @@ -145,15 +166,15 @@ public String toString() { + "originalExpression='" + originalExpression + '\'' - + ", compiledExpression='" - + compiledExpression + + ", compiledScript='" + + getCompiledScript().replace("\n", "\\n") + '\'' + ", argumentNames=" + argumentNames + ", argumentClasses=" + argumentClasses + ", returnClass=" - + returnClass + + getReturnClass() + ", columnNameMap=" + columnNameMap + '}'; diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilter.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilter.java index a9ea00dd072..a6b725e1d7c 100644 --- a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilter.java +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/transform/TransformFilter.java @@ -33,8 +33,6 @@ * *
      *
    • expression: a string for filter expression split from the user-defined filter. - *
    • scriptExpression: a string for filter script expression compiled from the column - * expression. *
    • columnNames: a list for recording the name of all columns used by the filter expression. *
    */ 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 56ce510c114..1a50f534d17 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 @@ -21,11 +21,10 @@ import org.apache.flink.cdc.common.converter.JavaClassConverter; import org.apache.flink.cdc.common.schema.Column; import org.apache.flink.cdc.common.source.SupportedMetadataColumn; +import org.apache.flink.cdc.runtime.parser.GeneratedExpression; import org.apache.flink.cdc.runtime.parser.JaninoCompiler; import org.apache.flink.cdc.runtime.parser.TransformParser; -import org.codehaus.janino.ExpressionEvaluator; - import java.lang.reflect.InvocationTargetException; import java.util.ArrayList; import java.util.HashMap; @@ -49,7 +48,7 @@ public class TransformFilterProcessor { private final Map supportedMetadataColumns; private final TransformExpressionKey transformExpressionKey; - private final ExpressionEvaluator expressionEvaluator; + private final TransformExpressionEvaluator expressionEvaluator; protected TransformFilterProcessor( boolean isNoOp, @@ -72,7 +71,6 @@ protected TransformFilterProcessor( } else { this.transformExpressionKey = generateTransformExpressionKey( - tableInfo.getPreTransformedSchema().getColumns(), udfDescriptors, supportedMetadataColumns .values() @@ -122,14 +120,14 @@ public boolean test(Object[] preRow, Object[] postRow, TransformContext context) String.format( "Failed to evaluate filtering expression for table `%s`.\n" + "\tOriginal expression: %s\n" - + "\tCompiled expression: %s\n" + + "\tCompiled script: %s\n" + "\tColumn name map: {%s}", tableInfo.getName(), transformExpressionKey != null ? transformExpressionKey.getOriginalExpression() : "", transformExpressionKey != null - ? transformExpressionKey.getCompiledExpression() + ? transformExpressionKey.getCompiledScript() : "", transformFilter.getColumnNameMapAsString()), e); @@ -142,15 +140,7 @@ private Tuple2, List>> generateArguments(boolean mapColumn List> argTypes = new ArrayList<>(); String expression = transformFilter.getExpression(); - // Post-transformed columns comes in priority - List columns = new ArrayList<>(tableInfo.getPostTransformedSchema().getColumns()); - { - Set existingColumnNames = - new HashSet<>(tableInfo.getPostTransformedSchema().getColumnNames()); - tableInfo.getPreTransformedSchema().getColumns().stream() - .filter(col -> !existingColumnNames.contains(col.getName())) - .forEach(columns::add); - } + List columns = getAvailableColumns(); Map columnNameMap = transformFilter.getColumnNameMap(); LinkedHashSet columnNames = new LinkedHashSet<>(transformFilter.getColumnNames()); @@ -187,6 +177,17 @@ private Tuple2, List>> generateArguments(boolean mapColumn return Tuple2.of(argNames, argTypes); } + private List getAvailableColumns() { + // Post-transformed columns come in priority. + List columns = new ArrayList<>(tableInfo.getPostTransformedSchema().getColumns()); + Set existingColumnNames = + new HashSet<>(tableInfo.getPostTransformedSchema().getColumnNames()); + tableInfo.getPreTransformedSchema().getColumns().stream() + .filter(column -> !existingColumnNames.contains(column.getName())) + .forEach(columns::add); + return columns; + } + private Object[] generateParams(Object[] preRow, Object[] postRow, TransformContext context) { List params = new ArrayList<>(); @@ -213,7 +214,6 @@ private Object[] generateParams(Object[] preRow, Object[] postRow, TransformCont } private TransformExpressionKey generateTransformExpressionKey( - List columns, List udfDescriptors, SupportedMetadataColumn[] supportedMetadataColumns) { Tuple2, List>> args = generateArguments(true); @@ -223,20 +223,19 @@ private TransformExpressionKey generateTransformExpressionKey( args.f0.add(JaninoCompiler.DEFAULT_EPOCH_TIME); args.f1.add(Long.class); - String scriptExpression = - TransformParser.translateFilterExpressionToJaninoExpression( + GeneratedExpression generatedExpression = + TransformParser.translateFilterExpressionToGeneratedExpression( transformFilter.getExpression(), - columns, + getAvailableColumns(), udfDescriptors, supportedMetadataColumns, transformFilter.getColumnNameMap()); return TransformExpressionKey.of( transformFilter.getExpression(), - scriptExpression, + generatedExpression, args.f0, args.f1, - Boolean.class, transformFilter.getColumnNameMap()); } } diff --git a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/GeneratedExpression.java b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/GeneratedExpression.java new file mode 100644 index 00000000000..365393e7cac --- /dev/null +++ b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/GeneratedExpression.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.parser; + +import java.io.Serializable; +import java.util.Objects; + +/** Statement-level Java code generated from a transform expression. */ +public class GeneratedExpression implements Serializable { + private static final long serialVersionUID = 1L; + + private final String code; + private final String resultTerm; + private final Class resultClass; + + private GeneratedExpression(String code, String resultTerm, Class resultClass) { + this.code = Objects.requireNonNull(code); + this.resultTerm = Objects.requireNonNull(resultTerm); + this.resultClass = Objects.requireNonNull(resultClass); + } + + public String getCode() { + return code; + } + + public String getResultTerm() { + return resultTerm; + } + + public Class getResultClass() { + return resultClass; + } + + public String asScript() { + StringBuilder script = new StringBuilder(); + appendCode(script, code); + script.append("return ").append(resultTerm).append(";"); + return script.toString(); + } + + public static GeneratedExpression of(String code, String resultTerm, Class resultClass) { + return new GeneratedExpression(code, resultTerm, resultClass); + } + + public static GeneratedExpression fromExpression(String resultTerm, Class resultClass) { + return of("", resultTerm, resultClass); + } + + private static void appendCode(StringBuilder builder, String code) { + if (code.isEmpty()) { + return; + } + builder.append(code); + if (code.charAt(code.length() - 1) != '\n') { + builder.append('\n'); + } + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + GeneratedExpression that = (GeneratedExpression) o; + return code.equals(that.code) + && resultTerm.equals(that.resultTerm) + && resultClass.equals(that.resultClass); + } + + @Override + public int hashCode() { + return Objects.hash(code, resultTerm, resultClass); + } + + @Override + public String toString() { + return "GeneratedExpression{" + + "code='" + + code + + '\'' + + ", resultTerm='" + + resultTerm + + '\'' + + ", resultClass=" + + resultClass + + '}'; + } +} 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 909c628fd3c..24aadd6f885 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.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; @@ -35,6 +36,7 @@ import org.apache.calcite.sql.SqlCharStringLiteral; import org.apache.calcite.sql.SqlDataTypeSpec; import org.apache.calcite.sql.SqlIdentifier; +import org.apache.calcite.sql.SqlKind; import org.apache.calcite.sql.SqlLiteral; import org.apache.calcite.sql.SqlNode; import org.apache.calcite.sql.SqlNodeList; @@ -47,6 +49,7 @@ import org.codehaus.janino.ExpressionEvaluator; import org.codehaus.janino.Java; +import java.math.BigDecimal; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -136,6 +139,17 @@ public static String translateSqlNodeToJaninoExpression(Context context, SqlNode return ""; } + public static GeneratedExpression translateSqlNodeToGeneratedExpression( + Context context, SqlNode transform, Class resultClass) { + return new GeneratedExpressionGenerator(context).translate(transform, resultClass); + } + + public static GeneratedExpression translateSqlNodeToGeneratedExpression( + Context context, SqlNode transform) { + return translateSqlNodeToGeneratedExpression( + context, transform, deduceGeneratedExpressionClass(context, transform)); + } + public static Java.Rvalue translateSqlNodeToJaninoRvalue(Context context, SqlNode transform) { if (transform instanceof SqlIdentifier) { return translateSqlIdentifier(context, (SqlIdentifier) transform); @@ -298,6 +312,7 @@ private static Java.Rvalue sqlBasicCallToJaninoRvalue( case CEIL: case FLOOR: case TRIM: + case COALESCE: case OTHER_FUNCTION: return generateOtherFunctionOperation(context, sqlBasicCall, atoms); case PLUS: @@ -496,6 +511,103 @@ private static Java.Rvalue generateEqualsOperation( Location.NOWHERE, null, StringUtils.convertToCamelCase("VALUE_EQUALS"), atoms); } + private static Class deduceGeneratedExpressionClass(Context context, SqlNode sqlNode) { + if (sqlNode instanceof SqlIdentifier) { + return deduceIdentifierClass(context, (SqlIdentifier) sqlNode); + } + if (sqlNode instanceof SqlLiteral) { + return deduceLiteralClass((SqlLiteral) sqlNode); + } + if (sqlNode instanceof SqlBasicCall && isBooleanResultCall((SqlBasicCall) sqlNode)) { + return Boolean.class; + } + try { + return JavaClassConverter.toJavaClass( + TransformParser.deduceSubExpressionType( + context.columns, + sqlNode, + context.udfDescriptors, + context.supportedMetadataColumns)); + } catch (RuntimeException e) { + return Object.class; + } + } + + private static Class deduceIdentifierClass(Context context, SqlIdentifier sqlIdentifier) { + String columnName = sqlIdentifier.names.get(sqlIdentifier.names.size() - 1); + for (Column column : context.columns) { + if (column.getName().equals(columnName)) { + return JavaClassConverter.toJavaClass(column.getType()); + } + } + for (SupportedMetadataColumn metadataColumn : context.supportedMetadataColumns) { + if (metadataColumn.getName().equals(columnName)) { + return metadataColumn.getJavaClass(); + } + } + Optional> metadataColumnClass = + MetadataColumns.METADATA_COLUMNS.stream() + .filter(column -> column.f0.equals(columnName)) + .findFirst() + .map(column -> (Class) column.f2); + return metadataColumnClass.orElse(Object.class); + } + + private static Class deduceLiteralClass(SqlLiteral sqlLiteral) { + if (sqlLiteral.getValue() == null) { + return Object.class; + } + if (sqlLiteral instanceof SqlCharStringLiteral) { + return String.class; + } + if (sqlLiteral instanceof SqlNumericLiteral) { + SqlNumericLiteral numericLiteral = (SqlNumericLiteral) sqlLiteral; + if (numericLiteral.isInteger()) { + long longValue = numericLiteral.longValue(true); + if (longValue > Integer.MAX_VALUE || longValue < Integer.MIN_VALUE) { + return Long.class; + } + return Integer.class; + } + return Double.class; + } + if (sqlLiteral.getTypeName() == SqlTypeName.BOOLEAN) { + return Boolean.class; + } + return Object.class; + } + + private static boolean isBooleanResultCall(SqlBasicCall sqlBasicCall) { + switch (sqlBasicCall.getKind()) { + case AND: + case OR: + case NOT: + case EQUALS: + case NOT_EQUALS: + case IS_DISTINCT_FROM: + case IS_NOT_DISTINCT_FROM: + case IS_NULL: + case IS_NOT_NULL: + case IS_FALSE: + case IS_NOT_TRUE: + case IS_TRUE: + case IS_NOT_FALSE: + case IS_UNKNOWN: + case BETWEEN: + case IN: + case NOT_IN: + case LIKE: + case SIMILAR: + case LESS_THAN: + case GREATER_THAN: + case LESS_THAN_OR_EQUAL: + case GREATER_THAN_OR_EQUAL: + return true; + default: + return false; + } + } + private static Java.Rvalue generateCastOperation( Context context, SqlBasicCall sqlBasicCall, Java.Rvalue[] atoms) { if (atoms.length != 1) { @@ -804,6 +916,418 @@ private static String generateInvokeExpression(UserDefinedFunctionDescriptor udf } } + private static class GeneratedExpressionGenerator { + private final Context context; + private int termId; + + private GeneratedExpressionGenerator(Context context) { + this.context = context; + } + + private GeneratedExpression translate(SqlNode sqlNode, Class resultClass) { + if (sqlNode instanceof SqlBasicCall) { + return translateSqlBasicCall((SqlBasicCall) sqlNode, resultClass); + } + if (sqlNode instanceof SqlCase) { + return translateSqlCase((SqlCase) sqlNode, resultClass); + } + Java.Rvalue rvalue = translateSqlNodeToJaninoRvalue(context, sqlNode); + if (rvalue == null) { + throw new ParseException("Unrecognized expression: " + sqlNode); + } + return GeneratedExpression.fromExpression(rvalue.toString(), resultClass); + } + + private GeneratedExpression translateSqlBasicCall( + SqlBasicCall sqlBasicCall, Class resultClass) { + switch (sqlBasicCall.getKind()) { + case AND: + return translateLogicalBinaryOperation(sqlBasicCall, true); + case OR: + return translateLogicalBinaryOperation(sqlBasicCall, false); + case OTHER_FUNCTION: + if (sqlBasicCall.getOperator().getName().equalsIgnoreCase("IF")) { + return translateIf(sqlBasicCall, resultClass); + } + return translateGenericBasicCall(sqlBasicCall, resultClass); + default: + return translateGenericBasicCall(sqlBasicCall, resultClass); + } + } + + private GeneratedExpression translateLogicalBinaryOperation( + SqlBasicCall sqlBasicCall, boolean isAnd) { + List operands = sqlBasicCall.getOperandList(); + if (operands.size() != 2) { + throw new ParseException("Unrecognized expression: " + sqlBasicCall); + } + + GeneratedExpression left = translate(operands.get(0), Boolean.class); + GeneratedExpression right = translate(operands.get(1), Boolean.class); + + String resultTerm = newTerm("result"); + String leftTerm = newTerm("left"); + String rightTerm = newTerm("right"); + String shortCircuitValue = isAnd ? "FALSE" : "TRUE"; + String functionName = isAnd ? "and" : "or"; + + StringBuilder code = new StringBuilder(); + appendCode(code, left.getCode()); + code.append("Boolean ").append(resultTerm).append(";\n"); + code.append("Boolean ") + .append(leftTerm) + .append(" = ") + .append(left.getResultTerm()) + .append(";\n"); + code.append("if (Boolean.") + .append(shortCircuitValue) + .append(".equals(") + .append(leftTerm) + .append(")) {\n"); + code.append(resultTerm).append(" = Boolean.").append(shortCircuitValue).append(";\n"); + code.append("} else {\n"); + appendCode(code, right.getCode()); + code.append("Boolean ") + .append(rightTerm) + .append(" = ") + .append(right.getResultTerm()) + .append(";\n"); + code.append(resultTerm) + .append(" = ") + .append(functionName) + .append("(") + .append(leftTerm) + .append(", ") + .append(rightTerm) + .append(");\n"); + code.append("}\n"); + + return GeneratedExpression.of(code.toString(), resultTerm, Boolean.class); + } + + private GeneratedExpression translateIf(SqlBasicCall sqlBasicCall, Class resultClass) { + List operands = sqlBasicCall.getOperandList(); + if (operands.size() != 3) { + throw new ParseException("Unrecognized expression: " + sqlBasicCall); + } + + GeneratedExpression condition = translate(operands.get(0), Boolean.class); + GeneratedExpression thenExpression = translateConditionalResult(operands.get(1)); + GeneratedExpression elseExpression = translateConditionalResult(operands.get(2)); + + String resultTerm = newTerm("result"); + String conditionTerm = newTerm("condition"); + StringBuilder code = new StringBuilder(); + appendCode(code, condition.getCode()); + code.append(className(resultClass)).append(" ").append(resultTerm).append(";\n"); + code.append("Boolean ") + .append(conditionTerm) + .append(" = ") + .append(condition.getResultTerm()) + .append(";\n"); + code.append("if (isTrue(").append(conditionTerm).append(")) {\n"); + appendCode(code, thenExpression.getCode()); + appendConditionalResultAssignment( + code, resultTerm, thenExpression, resultClass, sqlBasicCall); + code.append("} else {\n"); + appendCode(code, elseExpression.getCode()); + appendConditionalResultAssignment( + code, resultTerm, elseExpression, resultClass, sqlBasicCall); + code.append("}\n"); + + return GeneratedExpression.of(code.toString(), resultTerm, resultClass); + } + + private GeneratedExpression translateSqlCase(SqlCase sqlCase, Class resultClass) { + String resultTerm = newTerm("result"); + StringBuilder code = new StringBuilder(); + code.append(className(resultClass)).append(" ").append(resultTerm).append(";\n"); + appendCaseBranch( + code, + resultTerm, + sqlCase, + sqlCase.getWhenOperands(), + sqlCase.getThenOperands(), + sqlCase.getElseOperand(), + 0, + resultClass); + return GeneratedExpression.of(code.toString(), resultTerm, resultClass); + } + + private void appendCaseBranch( + StringBuilder code, + String resultTerm, + SqlCase sqlCase, + SqlNodeList whenOperands, + SqlNodeList thenOperands, + SqlNode elseOperand, + int index, + Class resultClass) { + if (index >= whenOperands.size()) { + GeneratedExpression elseExpression = + elseOperand == null + ? GeneratedExpression.fromExpression("null", resultClass) + : translateConditionalResult(elseOperand); + appendCode(code, elseExpression.getCode()); + appendConditionalResultAssignment( + code, resultTerm, elseExpression, resultClass, sqlCase); + return; + } + + GeneratedExpression whenExpression = translate(whenOperands.get(index), Boolean.class); + GeneratedExpression thenExpression = + translateConditionalResult(thenOperands.get(index)); + String conditionTerm = newTerm("condition"); + appendCode(code, whenExpression.getCode()); + code.append("Boolean ") + .append(conditionTerm) + .append(" = ") + .append(whenExpression.getResultTerm()) + .append(";\n"); + code.append("if (isTrue(").append(conditionTerm).append(")) {\n"); + appendCode(code, thenExpression.getCode()); + appendConditionalResultAssignment( + code, resultTerm, thenExpression, resultClass, sqlCase); + code.append("} else {\n"); + appendCaseBranch( + code, + resultTerm, + sqlCase, + whenOperands, + thenOperands, + elseOperand, + index + 1, + resultClass); + code.append("}\n"); + } + + private GeneratedExpression translateConditionalResult(SqlNode sqlNode) { + return translate(sqlNode, deduceGeneratedExpressionClass(context, sqlNode)); + } + + private void appendConditionalResultAssignment( + StringBuilder code, + String resultTerm, + GeneratedExpression value, + Class resultClass, + SqlNode conditionalExpression) { + code.append(resultTerm) + .append(" = ") + .append(coerceConditionalResult(value, resultClass, conditionalExpression)) + .append(";\n"); + } + + private String coerceConditionalResult( + GeneratedExpression value, Class resultClass, SqlNode conditionalExpression) { + if ("null".equals(value.getResultTerm()) + || resultClass.isAssignableFrom(value.getResultClass())) { + return value.getResultTerm(); + } + return coerceRvalueToResultClass( + toRvalue(value), resultClass, conditionalExpression, false) + .toString(); + } + + private Java.Rvalue coerceRvalueToResultClass( + Java.Rvalue valueRvalue, + Class resultClass, + SqlNode resultExpression, + boolean castReferenceType) { + if (resultClass == Byte.class) { + return generateFunctionOperation("castToByte", new Java.Rvalue[] {valueRvalue}); + } + if (resultClass == Short.class) { + return generateFunctionOperation("castToShort", new Java.Rvalue[] {valueRvalue}); + } + if (resultClass == Integer.class) { + return generateFunctionOperation("castToInteger", new Java.Rvalue[] {valueRvalue}); + } + if (resultClass == Long.class) { + return generateFunctionOperation("castToLong", new Java.Rvalue[] {valueRvalue}); + } + if (resultClass == Float.class) { + return generateFunctionOperation("castToFloat", new Java.Rvalue[] {valueRvalue}); + } + if (resultClass == Double.class) { + return generateFunctionOperation("castToDouble", new Java.Rvalue[] {valueRvalue}); + } + if (resultClass == BigDecimal.class) { + DataType resultType = + TransformParser.deduceSubExpressionType( + context.columns, + resultExpression, + context.udfDescriptors, + context.supportedMetadataColumns); + if (!(resultType instanceof DecimalType)) { + throw new ParseException( + "Unable to deduce decimal type for expression: " + resultExpression); + } + DecimalType decimalType = (DecimalType) resultType; + return generateFunctionOperation( + "castToBigDecimal", + new Java.Rvalue[] { + valueRvalue, + new Java.AmbiguousName( + Location.NOWHERE, + new String[] {String.valueOf(decimalType.getPrecision())}), + new Java.AmbiguousName( + Location.NOWHERE, + new String[] {String.valueOf(decimalType.getScale())}) + }); + } + if (resultClass == String.class) { + return generateFunctionOperation("castToString", new Java.Rvalue[] {valueRvalue}); + } + if (resultClass == Boolean.class) { + return generateFunctionOperation("castToBoolean", new Java.Rvalue[] {valueRvalue}); + } + if (castReferenceType && resultClass != Object.class) { + String canonicalName = resultClass.getCanonicalName(); + if (canonicalName != null) { + return new Java.Cast( + Location.NOWHERE, + new Java.ReferenceType( + Location.NOWHERE, + new Java.Annotation[0], + canonicalName.split("\\."), + null), + valueRvalue); + } + } + return valueRvalue; + } + + private GeneratedExpression translateGenericBasicCall( + SqlBasicCall sqlBasicCall, Class resultClass) { + List atoms = new ArrayList<>(); + List operands = sqlBasicCall.getOperandList(); + for (int index = 0; index < operands.size(); index++) { + SqlNode sqlNode = operands.get(index); + // CAST type metadata is consumed by generateCastOperation and is not a runtime + // operand. + if (sqlBasicCall.getKind() == SqlKind.CAST + && index == 1 + && sqlNode instanceof SqlDataTypeSpec) { + continue; + } + translateSqlNodeToGeneratedAtoms(sqlNode, atoms); + } + if (TIMEZONE_FREE_TEMPORAL_FUNCTIONS.contains( + sqlBasicCall.getOperator().getName().toUpperCase())) { + atoms.add( + GeneratedOperand.synthetic( + GeneratedExpression.fromExpression( + DEFAULT_EPOCH_TIME, Long.class))); + } else if (TIMEZONE_REQUIRED_TEMPORAL_FUNCTIONS.contains( + sqlBasicCall.getOperator().getName().toUpperCase())) { + atoms.add( + GeneratedOperand.synthetic( + GeneratedExpression.fromExpression( + DEFAULT_EPOCH_TIME, Long.class))); + atoms.add( + GeneratedOperand.synthetic( + GeneratedExpression.fromExpression( + DEFAULT_TIME_ZONE, String.class))); + } else if (TIMEZONE_REQUIRED_TEMPORAL_CONVERSION_FUNCTIONS.contains( + sqlBasicCall.getOperator().getName().toUpperCase())) { + atoms.add( + GeneratedOperand.synthetic( + GeneratedExpression.fromExpression( + DEFAULT_TIME_ZONE, String.class))); + } + + StringBuilder code = new StringBuilder(); + Java.Rvalue[] rvalues = new Java.Rvalue[atoms.size()]; + for (int i = 0; i < atoms.size(); i++) { + GeneratedOperand operand = atoms.get(i); + GeneratedExpression atom = operand.generatedExpression; + appendCode(code, atom.getCode()); + if (operand.sqlNode instanceof SqlBasicCall && hasGeneratedCodeAfter(atoms, i)) { + String operandTerm = newTerm("operand"); + code.append(className(atom.getResultClass())) + .append(" ") + .append(operandTerm) + .append(" = ") + .append(atom.getResultTerm()) + .append(";\n"); + rvalues[i] = + new Java.AmbiguousName(Location.NOWHERE, new String[] {operandTerm}); + } else { + rvalues[i] = toRvalue(atom); + } + } + Java.Rvalue rvalue = sqlBasicCallToJaninoRvalue(context, sqlBasicCall, rvalues); + if (sqlBasicCall.getKind() == SqlKind.COALESCE) { + rvalue = coerceRvalueToResultClass(rvalue, resultClass, sqlBasicCall, true); + } + return GeneratedExpression.of(code.toString(), rvalue.toString(), resultClass); + } + + private void translateSqlNodeToGeneratedAtoms( + SqlNode sqlNode, List atoms) { + if (sqlNode instanceof SqlNodeList) { + for (SqlNode node : (SqlNodeList) sqlNode) { + translateSqlNodeToGeneratedAtoms(node, atoms); + } + return; + } + atoms.add( + new GeneratedOperand( + sqlNode, + translate(sqlNode, deduceGeneratedExpressionClass(context, sqlNode)))); + } + + private static boolean hasGeneratedCodeAfter( + List operands, int currentIndex) { + for (int index = currentIndex + 1; index < operands.size(); index++) { + if (!operands.get(index).generatedExpression.getCode().isEmpty()) { + return true; + } + } + return false; + } + + private static class GeneratedOperand { + private final SqlNode sqlNode; + private final GeneratedExpression generatedExpression; + + private GeneratedOperand(SqlNode sqlNode, GeneratedExpression generatedExpression) { + this.sqlNode = sqlNode; + this.generatedExpression = generatedExpression; + } + + private static GeneratedOperand synthetic(GeneratedExpression generatedExpression) { + return new GeneratedOperand(null, generatedExpression); + } + } + + private String newTerm(String prefix) { + return prefix + "$" + termId++; + } + + private static Java.Rvalue toRvalue(GeneratedExpression generatedExpression) { + return new Java.AmbiguousName( + Location.NOWHERE, new String[] {generatedExpression.getResultTerm()}); + } + + private static void appendCode(StringBuilder builder, String code) { + if (code.isEmpty()) { + return; + } + builder.append(code); + if (code.charAt(code.length() - 1) != '\n') { + builder.append('\n'); + } + } + + private static String className(Class clazz) { + if (clazz.getCanonicalName() != null) { + return clazz.getCanonicalName(); + } + return clazz.getName(); + } + } + /** Contextual information for {@link JaninoCompiler}. */ public static class Context { 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 4912ef5a5d7..ac6a2a78a06 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 @@ -18,6 +18,7 @@ package org.apache.flink.cdc.runtime.parser; import org.apache.flink.api.common.io.ParseException; +import org.apache.flink.cdc.common.converter.JavaClassConverter; import org.apache.flink.cdc.common.schema.Column; import org.apache.flink.cdc.common.source.SupportedMetadataColumn; import org.apache.flink.cdc.common.types.DataType; @@ -344,19 +345,22 @@ public static List generateProjectionColumns( } else { List originalColumnNames = parseColumnNameList(exprNode); Map columnNameMap = generateColumnNameMap(originalColumnNames); + DataType projectionType = + CalciteDataTypeConverter.convertCalciteRelDataTypeToDataType( + relDataType); projectionColumn = ProjectionColumn.ofCalculated( columnName, - CalciteDataTypeConverter.convertCalciteRelDataTypeToDataType( - relDataType), + projectionType, exprNode.toString(), - JaninoCompiler.translateSqlNodeToJaninoExpression( + JaninoCompiler.translateSqlNodeToGeneratedExpression( JaninoCompiler.Context.of( columns, columnNameMap, udfDescriptors, supportedMetadataColumns), - exprNode), + exprNode, + JavaClassConverter.toJavaClass(projectionType)), originalColumnNames, columnNameMap); } @@ -550,7 +554,12 @@ public static ProjectionColumn resolveProjectionColumnFromIdentifier( CalciteDataTypeConverter.convertCalciteRelDataTypeToDataType(relDataType) .notNull(), identifier, - columnNameMap.get(identifier), + GeneratedExpression.fromExpression( + columnNameMap.get(identifier), + JavaClassConverter.toJavaClass( + CalciteDataTypeConverter.convertCalciteRelDataTypeToDataType( + relDataType) + .notNull())), Collections.singletonList(identifier), columnNameMap); } @@ -589,6 +598,27 @@ public static String translateFilterExpressionToJaninoExpression( where); } + public static GeneratedExpression translateFilterExpressionToGeneratedExpression( + String filterExpression, + List columns, + List udfDescriptors, + SupportedMetadataColumn[] supportedMetadataColumns, + Map columnNameMap) { + if (isNullOrWhitespaceOnly(filterExpression)) { + return GeneratedExpression.fromExpression("Boolean.TRUE", Boolean.class); + } + SqlSelect sqlSelect = TransformParser.parseFilterExpression(filterExpression); + if (!sqlSelect.hasWhere()) { + return GeneratedExpression.fromExpression("Boolean.TRUE", Boolean.class); + } + SqlNode where = sqlSelect.getWhere(); + return JaninoCompiler.translateSqlNodeToGeneratedExpression( + JaninoCompiler.Context.of( + columns, columnNameMap, udfDescriptors, supportedMetadataColumns), + where, + Boolean.class); + } + public static List parseComputedColumnNames( String projection, SupportedMetadataColumn[] supportedMetadataColumns) { List columnNames = new ArrayList<>(); 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 57563b28060..2b69bbe06ab 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 @@ -2133,6 +2133,66 @@ void testNullCastTransform() throws Exception { transformFunctionEventEventOperatorTestHarness.close(); } + @Test + void testNullableLogicalTransformUsesStatementLevelCodegen() throws Exception { + TableId tableId = TableId.tableId("my_company", "my_branch", "nullable_logical"); + Schema schema = + Schema.newBuilder() + .physicalColumn("id", DataTypes.STRING().notNull()) + .physicalColumn("nullable_bool", DataTypes.BOOLEAN()) + .primaryKey("id") + .build(); + Schema expectedSchema = + Schema.newBuilder() + .physicalColumn("id", DataTypes.STRING().notNull()) + .physicalColumn("and_false", DataTypes.BOOLEAN()) + .physicalColumn("and_true", DataTypes.BOOLEAN()) + .physicalColumn("or_true", DataTypes.BOOLEAN()) + .physicalColumn("or_false", DataTypes.BOOLEAN()) + .primaryKey("id") + .build(); + PostTransformOperator transform = + PostTransformOperator.newBuilder() + .addTransform( + tableId.identifier(), + "id" + + ", nullable_bool and false as and_false" + + ", nullable_bool and true as and_true" + + ", nullable_bool or true as or_true" + + ", nullable_bool or false as or_false", + "nullable_bool or true") + .build(); + RegularEventOperatorTestHarness + transformFunctionEventEventOperatorTestHarness = + RegularEventOperatorTestHarness.with(transform, 1); + transformFunctionEventEventOperatorTestHarness.open(); + + BinaryRecordDataGenerator recordDataGenerator = + new BinaryRecordDataGenerator(((RowType) schema.toRowDataType())); + BinaryRecordDataGenerator expectedRecordDataGenerator = + new BinaryRecordDataGenerator(((RowType) expectedSchema.toRowDataType())); + DataChangeEvent insertEvent = + DataChangeEvent.insertEvent( + tableId, + recordDataGenerator.generate( + new Object[] {new BinaryStringData("1"), null})); + DataChangeEvent insertEventExpect = + DataChangeEvent.insertEvent( + tableId, + expectedRecordDataGenerator.generate( + new Object[] {new BinaryStringData("1"), false, null, true, null})); + + transform.processElement(new StreamRecord<>(new CreateTableEvent(tableId, schema))); + Assertions.assertThat( + transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll()) + .isEqualTo(new StreamRecord<>(new CreateTableEvent(tableId, expectedSchema))); + transform.processElement(new StreamRecord<>(insertEvent)); + Assertions.assertThat( + transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll()) + .isEqualTo(new StreamRecord<>(insertEventExpect)); + transformFunctionEventEventOperatorTestHarness.close(); + } + @Test void testCastTransform() throws Exception { PostTransformOperator transform = @@ -2640,7 +2700,7 @@ void testCastErrorTransform() throws Exception { .hasRootCauseInstanceOf(IllegalArgumentException.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" + + "\tCompiled script: return castToTimestamp($0, __time_zone__);\n" + "\tColumn name map: {$0 -> castFloat}") .hasRootCauseMessage("Unable to parse given string as timestamp: 1.0"); @@ -2971,7 +3031,7 @@ void testReduceSchemaTransform() throws Exception { .addTransform( REDUCE_TABLEID.identifier(), "id, upper(id) as uid, age + 1 as newage, lower(ref1) as ref1, 17 as seventeen", - "newage > 17 and ref2 > 17") + "IF(TRUE, newage, 0) + 1 > 18 and ref2 > 17") .addTimezone("GMT") .build(); RegularEventOperatorTestHarness 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 8298d384ada..709b61bfc13 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 @@ -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; @@ -24,7 +25,16 @@ 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.common.udf.UserDefinedFunction; +import org.apache.flink.cdc.runtime.operators.transform.TransformExpressionKey; +import org.apache.flink.cdc.runtime.operators.transform.UserDefinedFunctionDescriptor; +import org.apache.calcite.sql.SqlDynamicParam; +import org.apache.calcite.sql.SqlLiteral; +import org.apache.calcite.sql.SqlNode; +import org.apache.calcite.sql.SqlNodeList; +import org.apache.calcite.sql.fun.SqlStdOperatorTable; +import org.apache.calcite.sql.parser.SqlParserPos; import org.assertj.core.api.Assertions; import org.codehaus.commons.compiler.CompileException; import org.codehaus.commons.compiler.Location; @@ -32,6 +42,7 @@ import org.codehaus.janino.Java; import org.codehaus.janino.Parser; import org.codehaus.janino.Scanner; +import org.codehaus.janino.ScriptEvaluator; import org.codehaus.janino.Unparser; import org.junit.jupiter.api.Test; @@ -48,6 +59,14 @@ /** Unit tests for the {@link JaninoCompiler}. */ class JaninoCompilerTest { + public static class CounterFunction implements UserDefinedFunction { + private int counter; + + public String eval() { + return String.valueOf(++counter); + } + } + @Test void testJaninoParser() throws CompileException, IOException, InvocationTargetException { String expression = "1==2"; @@ -287,6 +306,195 @@ void testTranslatedLogicalExpressionCompilesWithJanino() throws InvocationTarget Assertions.assertThat(nullableOrEvaluator.evaluate(new Object[] {null})).isNull(); } + @Test + void testGeneratedLogicalExpressionCompilesWithJanino() throws Exception { + List columns = List.of(Column.physicalColumn("nullable_bool", DataTypes.BOOLEAN())); + Map columnNameMap = Map.of("nullable_bool", "$0"); + List columnNames = List.of("$0"); + List> columnTypes = List.of(Boolean.class); + + GeneratedExpression nullAndFalse = + translateGeneratedFilterExpression( + "nullable_bool and false", columns, columnNameMap); + Assertions.assertThat(nullAndFalse.asScript()) + .contains("if (Boolean.FALSE.equals") + .doesNotContain("new java.util.function.Supplier"); + Assertions.assertThat( + compileGeneratedFilterExpression(nullAndFalse, columnNames, columnTypes) + .evaluate(new Object[] {null})) + .isEqualTo(false); + + GeneratedExpression nullAndTrue = + translateGeneratedFilterExpression( + "nullable_bool and true", columns, columnNameMap); + Assertions.assertThat(nullAndTrue.asScript()) + .contains("if (Boolean.FALSE.equals") + .doesNotContain("new java.util.function.Supplier"); + Assertions.assertThat( + compileGeneratedFilterExpression(nullAndTrue, columnNames, columnTypes) + .evaluate(new Object[] {null})) + .isNull(); + + GeneratedExpression nullOrTrue = + translateGeneratedFilterExpression("nullable_bool or true", columns, columnNameMap); + Assertions.assertThat(nullOrTrue.asScript()) + .contains("if (Boolean.TRUE.equals") + .doesNotContain("new java.util.function.Supplier"); + Assertions.assertThat( + compileGeneratedFilterExpression(nullOrTrue, columnNames, columnTypes) + .evaluate(new Object[] {null})) + .isEqualTo(true); + + GeneratedExpression nullOrFalse = + translateGeneratedFilterExpression( + "nullable_bool or false", columns, columnNameMap); + Assertions.assertThat(nullOrFalse.asScript()) + .contains("if (Boolean.TRUE.equals") + .doesNotContain("new java.util.function.Supplier"); + Assertions.assertThat( + compileGeneratedFilterExpression(nullOrFalse, columnNames, columnTypes) + .evaluate(new Object[] {null})) + .isNull(); + } + + @Test + void testGeneratedConditionalExpressionWithNumericTypeCoercion() throws Exception { + for (String expression : + List.of( + "IF(TRUE, 1, CAST(2 AS BIGINT)) > 0", + "CASE WHEN TRUE THEN 1 ELSE CAST(2 AS BIGINT) END > 0", + "IF(TRUE, CAST(1 AS TINYINT), 2) > 0", + "IF(TRUE, CAST(1 AS FLOAT), CAST(2 AS DOUBLE)) > 0", + "IF(TRUE, 1, CAST(2 AS DECIMAL(10, 2))) > 0", + "CASE WHEN TRUE THEN IF(TRUE, 1, CAST(2 AS BIGINT)) " + + "ELSE CAST(3 AS BIGINT) END > 0", + "IF(FALSE, 1, CAST(NULL AS BIGINT)) IS NULL")) { + GeneratedExpression generatedExpression = + translateGeneratedFilterExpression( + expression, Collections.emptyList(), Collections.emptyMap()); + + Assertions.assertThat( + compileGeneratedFilterExpression( + generatedExpression, + Collections.emptyList(), + Collections.emptyList()) + .evaluate(new Object[0])) + .as(expression) + .isEqualTo(true); + } + } + + @Test + void testGeneratedConditionalExpressionWithObjectReturningFunction() throws Exception { + for (String expression : + List.of( + "COALESCE(1, 2) = IF(TRUE, 1, 2)", + "IF(TRUE, COALESCE('a', 'b'), 'c') = 'a'")) { + GeneratedExpression generatedExpression = + translateGeneratedFilterExpression( + expression, Collections.emptyList(), Collections.emptyMap()); + + Assertions.assertThat( + compileGeneratedFilterExpression( + generatedExpression, + Collections.emptyList(), + Collections.emptyList()) + .evaluate(new Object[0])) + .as(expression) + .isEqualTo(true); + } + } + + @Test + void testGeneratedFunctionOperandsPreserveEvaluationOrder() throws Exception { + UserDefinedFunctionDescriptor descriptor = + new UserDefinedFunctionDescriptor("counter", CounterFunction.class.getName()); + GeneratedExpression generatedExpression = + TransformParser.translateFilterExpressionToGeneratedExpression( + "CONCAT(counter(), IF(TRUE, counter(), 'x')) = '12'", + Collections.emptyList(), + List.of(descriptor), + new SupportedMetadataColumn[0], + Collections.emptyMap()); + + Assertions.assertThat( + compileGeneratedFilterExpression( + generatedExpression, + List.of("__instanceOf" + descriptor.getClassName()), + List.of(CounterFunction.class)) + .evaluate(new Object[] {new CounterFunction()})) + .isEqualTo(true); + } + + @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 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 testTranslateSqlNodeToGeneratedExpressionRejectsUnrecognizedOperand() { + SqlNode expression = + SqlStdOperatorTable.COALESCE.createCall( + SqlParserPos.ZERO, + SqlLiteral.createExactNumeric("1", SqlParserPos.ZERO), + new SqlDynamicParam(0, SqlParserPos.ZERO)); + + Assertions.assertThatThrownBy( + () -> + JaninoCompiler.translateSqlNodeToGeneratedExpression( + JaninoCompiler.Context.of( + Collections.emptyList(), + Collections.emptyMap(), + Collections.emptyList(), + new SupportedMetadataColumn[0]), + expression, + Integer.class)) + .isExactlyInstanceOf(ParseException.class) + .hasMessage("Unrecognized expression: ?"); + } + + @Test + void testTranslateSqlNodeToGeneratedExpressionHandlesCastTypeMetadata() throws Exception { + GeneratedExpression generatedExpression = + translateGeneratedFilterExpression( + "CAST('true' AS BOOLEAN)", Collections.emptyList(), Collections.emptyMap()); + + Assertions.assertThat( + compileGeneratedFilterExpression( + generatedExpression, + Collections.emptyList(), + Collections.emptyList()) + .evaluate(new Object[0])) + .isEqualTo(true); + } + @Test void testLargeNumericLiterals() { // Test parsing integer literals @@ -396,4 +604,27 @@ private static ExpressionEvaluator compileTranslatedFilterExpression( columnTypes, Boolean.class); } + + private static GeneratedExpression translateGeneratedFilterExpression( + String expression, List columns, Map columnNameMap) { + return TransformParser.translateFilterExpressionToGeneratedExpression( + expression, + columns, + Collections.emptyList(), + new SupportedMetadataColumn[0], + columnNameMap); + } + + private static ScriptEvaluator compileGeneratedFilterExpression( + GeneratedExpression generatedExpression, + List columnNames, + List> columnTypes) + throws CompileException { + ScriptEvaluator scriptEvaluator = new ScriptEvaluator(); + scriptEvaluator.setParameters( + columnNames.toArray(new String[0]), columnTypes.toArray(new Class[0])); + scriptEvaluator.setReturnType(Boolean.class); + scriptEvaluator.cook(JaninoCompiler.loadSystemFunction(generatedExpression.asScript())); + return scriptEvaluator; + } } 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 9943ed069f5..64ecc32efa6 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 @@ -430,6 +430,39 @@ void testTranslateLogicalFilterToJaninoExpressionByNullability() { columns); } + @Test + void testTranslateLogicalFilterToGeneratedExpressionByNullability() { + List columns = List.of(Column.physicalColumn("nullable_bool", DataTypes.BOOLEAN())); + GeneratedExpression generatedExpression = + TransformParser.translateFilterExpressionToGeneratedExpression( + "nullable_bool and false", + columns, + Collections.emptyList(), + new SupportedMetadataColumn[0], + Collections.emptyMap()); + + Assertions.assertThat(generatedExpression.getCode()) + .contains("Boolean result$") + .contains("if (Boolean.FALSE.equals") + .doesNotContain("new java.util.function.Supplier"); + 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 @@ -561,13 +594,13 @@ void testGenerateProjectionColumns() { List expected = Arrays.asList( - "ProjectionColumn{column=`id` INT 'id', expression='id', scriptExpression='$0', originalColumnNames=[id], columnNameMap={id=$0}}", - "ProjectionColumn{column=`name` STRING, expression='UPPER(`TB`.`name`)', scriptExpression='upper($0)', originalColumnNames=[name], columnNameMap={name=$0}}", - "ProjectionColumn{column=`newage` INT, expression='`TB`.`age` + 1', scriptExpression='$0 + 1', originalColumnNames=[age], columnNameMap={age=$0}}", - "ProjectionColumn{column=`newCreateTime` TIMESTAMP(3) 'newCreateTime', expression='createTime', scriptExpression='$0', originalColumnNames=[createTime], columnNameMap={createTime=$0}}", - "ProjectionColumn{column=`newAddress` VARCHAR(50) 'newAddress', expression='address', scriptExpression='$0', originalColumnNames=[address], columnNameMap={address=$0}}", - "ProjectionColumn{column=`deposits` DECIMAL(10, 2) 'deposit', expression='deposit', scriptExpression='$0', originalColumnNames=[deposit], columnNameMap={deposit=$0}}", - "ProjectionColumn{column=`bmi` DOUBLE, expression='`TB`.`weight` / (`TB`.`height` * `TB`.`height`)', scriptExpression='$0 / $1 * $1', originalColumnNames=[weight, height, height], columnNameMap={weight=$0, height=$1}}"); + "ProjectionColumn{column=`id` INT 'id', expression='id', compiledScript='return $0;', originalColumnNames=[id], columnNameMap={id=$0}}", + "ProjectionColumn{column=`name` STRING, expression='UPPER(`TB`.`name`)', compiledScript='return upper($0);', originalColumnNames=[name], columnNameMap={name=$0}}", + "ProjectionColumn{column=`newage` INT, expression='`TB`.`age` + 1', compiledScript='return $0 + 1;', originalColumnNames=[age], columnNameMap={age=$0}}", + "ProjectionColumn{column=`newCreateTime` TIMESTAMP(3) 'newCreateTime', expression='createTime', compiledScript='return $0;', originalColumnNames=[createTime], columnNameMap={createTime=$0}}", + "ProjectionColumn{column=`newAddress` VARCHAR(50) 'newAddress', expression='address', compiledScript='return $0;', originalColumnNames=[address], columnNameMap={address=$0}}", + "ProjectionColumn{column=`deposits` DECIMAL(10, 2) 'deposit', expression='deposit', compiledScript='return $0;', originalColumnNames=[deposit], columnNameMap={deposit=$0}}", + "ProjectionColumn{column=`bmi` DOUBLE, expression='`TB`.`weight` / (`TB`.`height` * `TB`.`height`)', compiledScript='return $0 / $1 * $1;', originalColumnNames=[weight, height, height], columnNameMap={weight=$0, height=$1}}"); Assertions.assertThat(result).hasToString("[" + String.join(", ", expected) + "]"); List metadataResult = @@ -579,18 +612,18 @@ void testGenerateProjectionColumns() { List metadataExpected = Arrays.asList( - "ProjectionColumn{column=`id` INT 'id', expression='id', scriptExpression='$0', originalColumnNames=[id], columnNameMap={id=$0}}", - "ProjectionColumn{column=`name` STRING 'name', expression='name', scriptExpression='$0', originalColumnNames=[name], columnNameMap={name=$0}}", - "ProjectionColumn{column=`age` INT 'age', expression='age', scriptExpression='$0', originalColumnNames=[age], columnNameMap={age=$0}}", - "ProjectionColumn{column=`createTime` TIMESTAMP(3) 'newCreateTime', expression='createTime', scriptExpression='$0', originalColumnNames=[createTime], columnNameMap={createTime=$0}}", - "ProjectionColumn{column=`address` VARCHAR(50) 'newAddress', expression='address', scriptExpression='$0', originalColumnNames=[address], columnNameMap={address=$0}}", - "ProjectionColumn{column=`deposit` DECIMAL(10, 2) 'deposit', expression='deposit', scriptExpression='$0', originalColumnNames=[deposit], columnNameMap={deposit=$0}}", - "ProjectionColumn{column=`weight` DOUBLE 'weight', expression='weight', scriptExpression='$0', originalColumnNames=[weight], columnNameMap={weight=$0}}", - "ProjectionColumn{column=`height` DOUBLE 'height', expression='height', scriptExpression='$0', originalColumnNames=[height], columnNameMap={height=$0}}", - "ProjectionColumn{column=`op_type` STRING NOT NULL, expression='__data_event_type__', scriptExpression='$0', originalColumnNames=[__data_event_type__], columnNameMap={__data_event_type__=$0}}", - "ProjectionColumn{column=`__namespace_name__` STRING NOT NULL, expression='__namespace_name__', scriptExpression='$0', originalColumnNames=[__namespace_name__], columnNameMap={__namespace_name__=$0}}", - "ProjectionColumn{column=`__schema_name__` STRING NOT NULL, expression='__schema_name__', scriptExpression='$0', originalColumnNames=[__schema_name__], columnNameMap={__schema_name__=$0}}", - "ProjectionColumn{column=`__table_name__` STRING NOT NULL, expression='__table_name__', scriptExpression='$0', originalColumnNames=[__table_name__], columnNameMap={__table_name__=$0}}"); + "ProjectionColumn{column=`id` INT 'id', expression='id', compiledScript='return $0;', originalColumnNames=[id], columnNameMap={id=$0}}", + "ProjectionColumn{column=`name` STRING 'name', expression='name', compiledScript='return $0;', originalColumnNames=[name], columnNameMap={name=$0}}", + "ProjectionColumn{column=`age` INT 'age', expression='age', compiledScript='return $0;', originalColumnNames=[age], columnNameMap={age=$0}}", + "ProjectionColumn{column=`createTime` TIMESTAMP(3) 'newCreateTime', expression='createTime', compiledScript='return $0;', originalColumnNames=[createTime], columnNameMap={createTime=$0}}", + "ProjectionColumn{column=`address` VARCHAR(50) 'newAddress', expression='address', compiledScript='return $0;', originalColumnNames=[address], columnNameMap={address=$0}}", + "ProjectionColumn{column=`deposit` DECIMAL(10, 2) 'deposit', expression='deposit', compiledScript='return $0;', originalColumnNames=[deposit], columnNameMap={deposit=$0}}", + "ProjectionColumn{column=`weight` DOUBLE 'weight', expression='weight', compiledScript='return $0;', originalColumnNames=[weight], columnNameMap={weight=$0}}", + "ProjectionColumn{column=`height` DOUBLE 'height', expression='height', compiledScript='return $0;', originalColumnNames=[height], columnNameMap={height=$0}}", + "ProjectionColumn{column=`op_type` STRING NOT NULL, expression='__data_event_type__', compiledScript='return $0;', originalColumnNames=[__data_event_type__], columnNameMap={__data_event_type__=$0}}", + "ProjectionColumn{column=`__namespace_name__` STRING NOT NULL, expression='__namespace_name__', compiledScript='return $0;', originalColumnNames=[__namespace_name__], columnNameMap={__namespace_name__=$0}}", + "ProjectionColumn{column=`__schema_name__` STRING NOT NULL, expression='__schema_name__', compiledScript='return $0;', originalColumnNames=[__schema_name__], columnNameMap={__schema_name__=$0}}", + "ProjectionColumn{column=`__table_name__` STRING NOT NULL, expression='__table_name__', compiledScript='return $0;', originalColumnNames=[__table_name__], columnNameMap={__table_name__=$0}}"); Assertions.assertThat(metadataResult) .map(ProjectionColumn::toString) .containsExactlyElementsOf(metadataExpected); @@ -632,15 +665,15 @@ public void testGenerateProjectionColumnsWithPrecision() { List expected = Arrays.asList( - "ProjectionColumn{column=`id` INT 'id', expression='id', scriptExpression='$0', originalColumnNames=[id], columnNameMap={id=$0}}", - "ProjectionColumn{column=`name2` STRING, expression='UPPER(`TB`.`name`)', scriptExpression='upper($0)', originalColumnNames=[name], columnNameMap={name=$0}}", - "ProjectionColumn{column=`sex2` STRING, expression='UPPER(`TB`.`sex`)', scriptExpression='upper($0)', originalColumnNames=[sex], columnNameMap={sex=$0}}", - "ProjectionColumn{column=`address2` BINARY(50), expression='CASE WHEN `TB`.`address` IS NOT NULL THEN `TB`.`address` ELSE `TB`.`address` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)', originalColumnNames=[address, address, address], columnNameMap={address=$0}}", - "ProjectionColumn{column=`phone2` VARBINARY(50), expression='CASE WHEN `TB`.`phone` IS NOT NULL THEN `TB`.`phone` ELSE `TB`.`phone` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)', originalColumnNames=[phone, phone, phone], columnNameMap={phone=$0}}", - "ProjectionColumn{column=`deposit2` DECIMAL(10, 2), expression='CASE WHEN `TB`.`deposit` IS NOT NULL THEN `TB`.`deposit` ELSE `TB`.`deposit` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)', originalColumnNames=[deposit, deposit, deposit], columnNameMap={deposit=$0}}", - "ProjectionColumn{column=`birthday2` TIMESTAMP(3), expression='CASE WHEN `TB`.`birthday` IS NOT NULL THEN `TB`.`birthday` ELSE `TB`.`birthday` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)', originalColumnNames=[birthday, birthday, birthday], columnNameMap={birthday=$0}}", - "ProjectionColumn{column=`birthday_ltz2` TIMESTAMP_LTZ(3), expression='CASE WHEN `TB`.`birthday_ltz` IS NOT NULL THEN `TB`.`birthday_ltz` ELSE `TB`.`birthday_ltz` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)', originalColumnNames=[birthday_ltz, birthday_ltz, birthday_ltz], columnNameMap={birthday_ltz=$0}}", - "ProjectionColumn{column=`update_time2` TIME(3), expression='CASE WHEN `TB`.`update_time` IS NOT NULL THEN `TB`.`update_time` ELSE `TB`.`update_time` END', scriptExpression='(isTrue(isNotNull($0)) ? $0 : $0)', originalColumnNames=[update_time, update_time, update_time], columnNameMap={update_time=$0}}"); + "ProjectionColumn{column=`id` INT 'id', expression='id', compiledScript='return $0;', originalColumnNames=[id], columnNameMap={id=$0}}", + "ProjectionColumn{column=`name2` STRING, expression='UPPER(`TB`.`name`)', compiledScript='return upper($0);', originalColumnNames=[name], columnNameMap={name=$0}}", + "ProjectionColumn{column=`sex2` STRING, expression='UPPER(`TB`.`sex`)', compiledScript='return upper($0);', originalColumnNames=[sex], columnNameMap={sex=$0}}", + "ProjectionColumn{column=`address2` BINARY(50), expression='CASE WHEN `TB`.`address` IS NOT NULL THEN `TB`.`address` ELSE `TB`.`address` END', compiledScript='byte[] result$0;\\nBoolean condition$1 = isNotNull($0);\\nif (isTrue(condition$1)) {\\nresult$0 = $0;\\n} else {\\nresult$0 = $0;\\n}\\nreturn result$0;', originalColumnNames=[address, address, address], columnNameMap={address=$0}}", + "ProjectionColumn{column=`phone2` VARBINARY(50), expression='CASE WHEN `TB`.`phone` IS NOT NULL THEN `TB`.`phone` ELSE `TB`.`phone` END', compiledScript='byte[] result$0;\\nBoolean condition$1 = isNotNull($0);\\nif (isTrue(condition$1)) {\\nresult$0 = $0;\\n} else {\\nresult$0 = $0;\\n}\\nreturn result$0;', originalColumnNames=[phone, phone, phone], columnNameMap={phone=$0}}", + "ProjectionColumn{column=`deposit2` DECIMAL(10, 2), expression='CASE WHEN `TB`.`deposit` IS NOT NULL THEN `TB`.`deposit` ELSE `TB`.`deposit` END', compiledScript='java.math.BigDecimal result$0;\\nBoolean condition$1 = isNotNull($0);\\nif (isTrue(condition$1)) {\\nresult$0 = $0;\\n} else {\\nresult$0 = $0;\\n}\\nreturn result$0;', originalColumnNames=[deposit, deposit, deposit], columnNameMap={deposit=$0}}", + "ProjectionColumn{column=`birthday2` TIMESTAMP(3), expression='CASE WHEN `TB`.`birthday` IS NOT NULL THEN `TB`.`birthday` ELSE `TB`.`birthday` END', compiledScript='java.time.LocalDateTime result$0;\\nBoolean condition$1 = isNotNull($0);\\nif (isTrue(condition$1)) {\\nresult$0 = $0;\\n} else {\\nresult$0 = $0;\\n}\\nreturn result$0;', originalColumnNames=[birthday, birthday, birthday], columnNameMap={birthday=$0}}", + "ProjectionColumn{column=`birthday_ltz2` TIMESTAMP_LTZ(3), expression='CASE WHEN `TB`.`birthday_ltz` IS NOT NULL THEN `TB`.`birthday_ltz` ELSE `TB`.`birthday_ltz` END', compiledScript='java.time.Instant result$0;\\nBoolean condition$1 = isNotNull($0);\\nif (isTrue(condition$1)) {\\nresult$0 = $0;\\n} else {\\nresult$0 = $0;\\n}\\nreturn result$0;', originalColumnNames=[birthday_ltz, birthday_ltz, birthday_ltz], columnNameMap={birthday_ltz=$0}}", + "ProjectionColumn{column=`update_time2` TIME(3), expression='CASE WHEN `TB`.`update_time` IS NOT NULL THEN `TB`.`update_time` ELSE `TB`.`update_time` END', compiledScript='java.time.LocalTime result$0;\\nBoolean condition$1 = isNotNull($0);\\nif (isTrue(condition$1)) {\\nresult$0 = $0;\\n} else {\\nresult$0 = $0;\\n}\\nreturn result$0;', originalColumnNames=[update_time, update_time, update_time], columnNameMap={update_time=$0}}"); Assertions.assertThat(result).hasToString("[" + String.join(", ", expected) + "]"); } @@ -842,15 +875,15 @@ public void testProjectionColumnsWithColumnNameMap() { List expected = Arrays.asList( - "ProjectionColumn{column=`a` INT 'a', expression='a', scriptExpression='$0', originalColumnNames=[a], columnNameMap={a=$0}}", - "ProjectionColumn{column=`b` INT 'b', expression='b', scriptExpression='$0', originalColumnNames=[b], columnNameMap={b=$0}}", - "ProjectionColumn{column=`c` INT, expression='`TB`.`a` - `TB`.`b`', scriptExpression='$0 - $1', originalColumnNames=[a, b], columnNameMap={a=$0, b=$1}}", - "ProjectionColumn{column=`a-b` DOUBLE '`a-b`', expression='a-b', scriptExpression='$0', originalColumnNames=[a-b], columnNameMap={a-b=$0}}", - "ProjectionColumn{column=`d` DOUBLE '`a-b`', expression='a-b', scriptExpression='$0', originalColumnNames=[a-b], columnNameMap={a-b=$0}}", - "ProjectionColumn{column=`e` DOUBLE, expression='`TB`.`a-b` - 1', scriptExpression='$0 - 1', originalColumnNames=[a-b], columnNameMap={a-b=$0}}", - "ProjectionColumn{column=`f` DOUBLE, expression='`TB`.`a` - `TB`.`b` + `TB`.`a-b`', scriptExpression='$0 - $1 + $2', originalColumnNames=[a, b, a-b], columnNameMap={a=$0, b=$1, a-b=$2}}", - "ProjectionColumn{column=`test-meta-col` INT NOT NULL, expression='test-meta-col', scriptExpression='$0', originalColumnNames=[test-meta-col], columnNameMap={test-meta-col=$0}}", - "ProjectionColumn{column=`g` INT, expression='`TB`.`test-meta-col` - `TB`.`a` - `TB`.`b`', scriptExpression='$0 - $1 - $2', originalColumnNames=[test-meta-col, a, b], columnNameMap={a=$1, b=$2, test-meta-col=$0}}"); + "ProjectionColumn{column=`a` INT 'a', expression='a', compiledScript='return $0;', originalColumnNames=[a], columnNameMap={a=$0}}", + "ProjectionColumn{column=`b` INT 'b', expression='b', compiledScript='return $0;', originalColumnNames=[b], columnNameMap={b=$0}}", + "ProjectionColumn{column=`c` INT, expression='`TB`.`a` - `TB`.`b`', compiledScript='return $0 - $1;', originalColumnNames=[a, b], columnNameMap={a=$0, b=$1}}", + "ProjectionColumn{column=`a-b` DOUBLE '`a-b`', expression='a-b', compiledScript='return $0;', originalColumnNames=[a-b], columnNameMap={a-b=$0}}", + "ProjectionColumn{column=`d` DOUBLE '`a-b`', expression='a-b', compiledScript='return $0;', originalColumnNames=[a-b], columnNameMap={a-b=$0}}", + "ProjectionColumn{column=`e` DOUBLE, expression='`TB`.`a-b` - 1', compiledScript='return $0 - 1;', originalColumnNames=[a-b], columnNameMap={a-b=$0}}", + "ProjectionColumn{column=`f` DOUBLE, expression='`TB`.`a` - `TB`.`b` + `TB`.`a-b`', compiledScript='return $0 - $1 + $2;', originalColumnNames=[a, b, a-b], columnNameMap={a=$0, b=$1, a-b=$2}}", + "ProjectionColumn{column=`test-meta-col` INT NOT NULL, expression='test-meta-col', compiledScript='return $0;', originalColumnNames=[test-meta-col], columnNameMap={test-meta-col=$0}}", + "ProjectionColumn{column=`g` INT, expression='`TB`.`test-meta-col` - `TB`.`a` - `TB`.`b`', compiledScript='return $0 - $1 - $2;', originalColumnNames=[test-meta-col, a, b], columnNameMap={a=$1, b=$2, test-meta-col=$0}}"); Assertions.assertThat(result).hasToString("[" + String.join(", ", expected) + "]"); } @@ -886,12 +919,12 @@ void testParsingExpressionWithUnicodeLiterals() { new SupportedMetadataColumn[] {})) .map(ProjectionColumn::getScriptExpression) .containsExactly( - "$0", - "$0", - "valueEquals($0, \"" + unicodeString + "\")", - "!valueEquals($0, \"" + unicodeString + "\")", - "valueEquals($0, castToInteger(\"" + unicodeString + "\"))", - "!valueEquals($0, castToInteger(\"" + unicodeString + "\"))"); + "return $0;", + "return $0;", + "return valueEquals($0, \"" + unicodeString + "\");", + "return !valueEquals($0, \"" + unicodeString + "\");", + "return valueEquals($0, castToInteger(\"" + unicodeString + "\"));", + "return !valueEquals($0, castToInteger(\"" + unicodeString + "\"));"); testFilterExpression( "a = '" + unicodeString + "'", "valueEquals(a, \"" + unicodeString + "\")"); @@ -903,6 +936,13 @@ void testParsingExpressionWithUnicodeLiterals() { private static final List DUMMY_COLUMNS = List.of(Column.physicalColumn("id", DataTypes.INT())); + private static String lazyLogicalFunction( + String functionName, String leftOperand, String rightOperand) { + return String.format( + "%s(%s, new java.util.function.Supplier() { public Boolean get() { return %s; } })", + functionName, leftOperand, rightOperand); + } + private void testFilterExpression(String expression, String expressionExpect) { String janinoExpression = TransformParser.translateFilterExpressionToJaninoExpression( @@ -926,13 +966,6 @@ private void testFilterExpressionWithColumns( Assertions.assertThat(janinoExpression).isEqualTo(expressionExpect); } - private static String lazyLogicalFunction( - String functionName, String leftOperand, String rightOperand) { - return String.format( - "%s(%s, new java.util.function.Supplier() { public Boolean get() { return %s; } })", - functionName, leftOperand, rightOperand); - } - private void testFilterExpressionWithUdf(String expression, String expressionExpect) { testFilterExpressionWithUdf( expression, expressionExpect, DUMMY_COLUMNS, Collections.emptyMap());