You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "haruki-830 (via GitHub)" <gi...@apache.org> on 2026/07/22 03:48:20 UTC
[PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
haruki-830 opened a new pull request, #4482:
URL: https://github.com/apache/flink-cdc/pull/4482
#### Summary
This commit introduces statement-level code generation for Flink CDC YAML Transform expressions. Transform expressions are now represented by a `GeneratedExpression` intermediate structure and compiled through Janino `ScriptEvaluator`, allowing nullable `AND` / `OR` short-circuit logic to be generated as explicit `if` / `else` statements instead of per-record `Supplier` wrappers.
#### Key Changes
1. Introduce Statement-Level Transform Codegen
- Added `GeneratedExpression` to carry generated statements, result term, and result Java type.
- Added `TransformExpressionEvaluator` as a small wrapper around Janino `ScriptEvaluator`.
- Updated projection and filter processors to compile and evaluate generated scripts.
2. Support Statement-Level Nullable AND / OR
- Added `translateSqlNodeToGeneratedExpression(...)` in `JaninoCompiler`.
- Generates explicit short-circuit `if` / `else` blocks for nullable `AND` and `OR`.
- Ensures the right operand is evaluated only when required.
3. Refine Nullability Handling
- Added Calcite / RelNode nullability lookup for transform sub-expressions.
- Uses RelNode nullability only to refine the existing conservative fallback, avoiding broader semantic changes in this PR.
4. Test Coverage
- Added coverage for nullable `AND` / `OR` SQL three-valued logic.
- Added short-circuit tests for:
- `false AND (1 / 0 > 0)`
- `true OR (1 / 0 > 0)`
- Added projection and filter coverage.
- Added assertions that generated code no longer contains `new Supplier`.
#### JIRA Reference
[https://issues.apache.org/jira/browse/FLINK-40204](url)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
Posted by "haruki-830 (via GitHub)" <gi...@apache.org>.
haruki-830 commented on code in PR #4482:
URL: https://github.com/apache/flink-cdc/pull/4482#discussion_r3637176143
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java:
##########
@@ -804,6 +904,265 @@ 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<SqlNode> 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<SqlNode> operands = sqlBasicCall.getOperandList();
+ if (operands.size() != 3) {
+ throw new ParseException("Unrecognized expression: " + sqlBasicCall);
+ }
+
+ GeneratedExpression condition = translate(operands.get(0), Boolean.class);
+ GeneratedExpression thenExpression = translate(operands.get(1), resultClass);
+ GeneratedExpression elseExpression = translate(operands.get(2), resultClass);
+
+ 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());
+ code.append(resultTerm)
+ .append(" = ")
+ .append(thenExpression.getResultTerm())
+ .append(";\n");
+ code.append("} else {\n");
+ appendCode(code, elseExpression.getCode());
+ code.append(resultTerm)
+ .append(" = ")
+ .append(elseExpression.getResultTerm())
+ .append(";\n");
+ 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.getWhenOperands(),
+ sqlCase.getThenOperands(),
+ sqlCase.getElseOperand(),
+ 0,
+ resultClass);
+ return GeneratedExpression.of(code.toString(), resultTerm, resultClass);
+ }
+
+ private void appendCaseBranch(
+ StringBuilder code,
+ String resultTerm,
+ SqlNodeList whenOperands,
+ SqlNodeList thenOperands,
+ SqlNode elseOperand,
+ int index,
+ Class<?> resultClass) {
+ if (index >= whenOperands.size()) {
+ GeneratedExpression elseExpression =
+ elseOperand == null
+ ? GeneratedExpression.fromExpression("null", resultClass)
+ : translate(elseOperand, resultClass);
+ appendCode(code, elseExpression.getCode());
+ code.append(resultTerm)
+ .append(" = ")
+ .append(elseExpression.getResultTerm())
+ .append(";\n");
+ return;
+ }
+
+ GeneratedExpression whenExpression = translate(whenOperands.get(index), Boolean.class);
+ GeneratedExpression thenExpression = translate(thenOperands.get(index), resultClass);
+ 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());
+ code.append(resultTerm)
+ .append(" = ")
+ .append(thenExpression.getResultTerm())
+ .append(";\n");
+ code.append("} else {\n");
+ appendCaseBranch(
+ code,
+ resultTerm,
+ whenOperands,
+ thenOperands,
+ elseOperand,
+ index + 1,
+ resultClass);
+ code.append("}\n");
+ }
+
+ private GeneratedExpression translateGenericBasicCall(
+ SqlBasicCall sqlBasicCall, Class<?> resultClass) {
+ List<GeneratedExpression> atoms = new ArrayList<>();
+ for (SqlNode sqlNode : sqlBasicCall.getOperandList()) {
+ translateSqlNodeToGeneratedAtoms(sqlNode, atoms);
+ }
+ if (TIMEZONE_FREE_TEMPORAL_FUNCTIONS.contains(
+ sqlBasicCall.getOperator().getName().toUpperCase())) {
+ atoms.add(GeneratedExpression.fromExpression(DEFAULT_EPOCH_TIME, Long.class));
+ } else if (TIMEZONE_REQUIRED_TEMPORAL_FUNCTIONS.contains(
+ sqlBasicCall.getOperator().getName().toUpperCase())) {
+ atoms.add(GeneratedExpression.fromExpression(DEFAULT_EPOCH_TIME, Long.class));
+ atoms.add(GeneratedExpression.fromExpression(DEFAULT_TIME_ZONE, String.class));
+ } else if (TIMEZONE_REQUIRED_TEMPORAL_CONVERSION_FUNCTIONS.contains(
+ sqlBasicCall.getOperator().getName().toUpperCase())) {
+ atoms.add(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++) {
+ GeneratedExpression atom = atoms.get(i);
+ appendCode(code, atom.getCode());
+ rvalues[i] = toRvalue(atom);
+ }
+ Java.Rvalue rvalue = sqlBasicCallToJaninoRvalue(context, sqlBasicCall, rvalues);
+ return GeneratedExpression.of(code.toString(), rvalue.toString(), resultClass);
+ }
+
+ private void translateSqlNodeToGeneratedAtoms(
+ SqlNode sqlNode, List<GeneratedExpression> atoms) {
+ if (sqlNode instanceof SqlNodeList) {
+ for (SqlNode node : (SqlNodeList) sqlNode) {
+ translateSqlNodeToGeneratedAtoms(node, atoms);
+ }
+ } else if (sqlNode instanceof SqlIdentifier
+ || sqlNode instanceof SqlLiteral
+ || sqlNode instanceof SqlBasicCall
+ || sqlNode instanceof SqlCase) {
+ atoms.add(translate(sqlNode, deduceGeneratedExpressionClass(context, sqlNode)));
+ }
+ }
Review Comment:
Addressed in the latest commit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4482:
URL: https://github.com/apache/flink-cdc/pull/4482#discussion_r3636121566
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java:
##########
@@ -804,6 +904,265 @@ 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<SqlNode> 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<SqlNode> operands = sqlBasicCall.getOperandList();
+ if (operands.size() != 3) {
+ throw new ParseException("Unrecognized expression: " + sqlBasicCall);
+ }
+
+ GeneratedExpression condition = translate(operands.get(0), Boolean.class);
+ GeneratedExpression thenExpression = translate(operands.get(1), resultClass);
+ GeneratedExpression elseExpression = translate(operands.get(2), resultClass);
+
+ 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());
+ code.append(resultTerm)
+ .append(" = ")
+ .append(thenExpression.getResultTerm())
+ .append(";\n");
+ code.append("} else {\n");
+ appendCode(code, elseExpression.getCode());
+ code.append(resultTerm)
+ .append(" = ")
+ .append(elseExpression.getResultTerm())
+ .append(";\n");
+ 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.getWhenOperands(),
+ sqlCase.getThenOperands(),
+ sqlCase.getElseOperand(),
+ 0,
+ resultClass);
+ return GeneratedExpression.of(code.toString(), resultTerm, resultClass);
+ }
+
+ private void appendCaseBranch(
+ StringBuilder code,
+ String resultTerm,
+ SqlNodeList whenOperands,
+ SqlNodeList thenOperands,
+ SqlNode elseOperand,
+ int index,
+ Class<?> resultClass) {
+ if (index >= whenOperands.size()) {
+ GeneratedExpression elseExpression =
+ elseOperand == null
+ ? GeneratedExpression.fromExpression("null", resultClass)
+ : translate(elseOperand, resultClass);
+ appendCode(code, elseExpression.getCode());
+ code.append(resultTerm)
+ .append(" = ")
+ .append(elseExpression.getResultTerm())
+ .append(";\n");
+ return;
+ }
+
+ GeneratedExpression whenExpression = translate(whenOperands.get(index), Boolean.class);
+ GeneratedExpression thenExpression = translate(thenOperands.get(index), resultClass);
+ 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());
+ code.append(resultTerm)
+ .append(" = ")
+ .append(thenExpression.getResultTerm())
+ .append(";\n");
+ code.append("} else {\n");
+ appendCaseBranch(
+ code,
+ resultTerm,
+ whenOperands,
+ thenOperands,
+ elseOperand,
+ index + 1,
+ resultClass);
+ code.append("}\n");
+ }
+
+ private GeneratedExpression translateGenericBasicCall(
+ SqlBasicCall sqlBasicCall, Class<?> resultClass) {
+ List<GeneratedExpression> atoms = new ArrayList<>();
+ for (SqlNode sqlNode : sqlBasicCall.getOperandList()) {
+ translateSqlNodeToGeneratedAtoms(sqlNode, atoms);
+ }
+ if (TIMEZONE_FREE_TEMPORAL_FUNCTIONS.contains(
+ sqlBasicCall.getOperator().getName().toUpperCase())) {
+ atoms.add(GeneratedExpression.fromExpression(DEFAULT_EPOCH_TIME, Long.class));
+ } else if (TIMEZONE_REQUIRED_TEMPORAL_FUNCTIONS.contains(
+ sqlBasicCall.getOperator().getName().toUpperCase())) {
+ atoms.add(GeneratedExpression.fromExpression(DEFAULT_EPOCH_TIME, Long.class));
+ atoms.add(GeneratedExpression.fromExpression(DEFAULT_TIME_ZONE, String.class));
+ } else if (TIMEZONE_REQUIRED_TEMPORAL_CONVERSION_FUNCTIONS.contains(
+ sqlBasicCall.getOperator().getName().toUpperCase())) {
+ atoms.add(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++) {
+ GeneratedExpression atom = atoms.get(i);
+ appendCode(code, atom.getCode());
+ rvalues[i] = toRvalue(atom);
+ }
+ Java.Rvalue rvalue = sqlBasicCallToJaninoRvalue(context, sqlBasicCall, rvalues);
+ return GeneratedExpression.of(code.toString(), rvalue.toString(), resultClass);
+ }
+
+ private void translateSqlNodeToGeneratedAtoms(
+ SqlNode sqlNode, List<GeneratedExpression> atoms) {
+ if (sqlNode instanceof SqlNodeList) {
+ for (SqlNode node : (SqlNodeList) sqlNode) {
+ translateSqlNodeToGeneratedAtoms(node, atoms);
+ }
+ } else if (sqlNode instanceof SqlIdentifier
+ || sqlNode instanceof SqlLiteral
+ || sqlNode instanceof SqlBasicCall
+ || sqlNode instanceof SqlCase) {
+ atoms.add(translate(sqlNode, deduceGeneratedExpressionClass(context, sqlNode)));
+ }
+ }
Review Comment:
translateSqlNodeToGeneratedAtoms(...) silently ignores unsupported SqlNode operand types (falls through without adding an atom or throwing). This can mask invalid expressions and may lead to wrong argument counts being passed to sqlBasicCallToJaninoRvalue(), producing confusing errors or incorrect generated code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #4482:
URL: https://github.com/apache/flink-cdc/pull/4482#issuecomment-5090750411
> Thanks @lvyanquan and @yuxiqian for the feedback.
>
> I ran a JMH benchmark comparing the Supplier path against an equivalent if/else script and found no statistically significant difference — all configurations fall in the 330k-372k ops/ms range with overlapping confidence intervals, that means the JIT escape analysis reliably eliminates the anonymous `Supplier` allocation, so the per-record allocation concern is not a real bottleneck in practice.
>
> Given the lack of measurable benefit and the correctness risks @yuxiqian identified, I agree this change is not warranted. I'll go ahead and close this PR later.
Thanks for the detailed benchmark report.
> rvalue.toString() is for debugging purposes only and may not generate Java code with correct parentheses.
>
> For example, SQL statement (a - b) / (c + 1) will be converted to a - b / c + 1.
Seems this is an existing bug, not introduced by changes in this PR. Could you please investigate this later?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #4482:
URL: https://github.com/apache/flink-cdc/pull/4482#discussion_r3654317371
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java:
##########
@@ -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);
Review Comment:
`rvalue.toString()` is for debugging purposes only and may not generate Java code with correct parentheses.
For example, SQL statement `(a - b) / (c + 1)` will be converted to `a - b / c + 1`.
##########
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');
+ }
+ }
Review Comment:
Ditto, avoid string ops
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java:
##########
@@ -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<Class<?>> 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) {
Review Comment:
One may write a UDF returning `BOOLEAN`?
##########
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();
+ }
Review Comment:
Better avoid string manipulations on dumped Java codes. Is `Java.ReturnStatement` useful here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4482:
URL: https://github.com/apache/flink-cdc/pull/4482#issuecomment-5087076479
This change offers very limited benefits but has a relatively large impact surface. I think we need to evaluate it carefully.
CC @yuxiqian
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #4482:
URL: https://github.com/apache/flink-cdc/pull/4482#discussion_r3629085632
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java:
##########
@@ -804,6 +904,263 @@ 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);
+ return GeneratedExpression.fromExpression(
+ rvalue == null ? "" : rvalue.toString(), resultClass);
Review Comment:
Creating a `GeneratedExpression` with an empty `resultTerm` when `translateSqlNodeToJaninoRvalue(...)` returns null can produce syntactically invalid scripts later (e.g., `return ;`), and the failure will surface far away from the root cause. Prefer failing fast here (e.g., throw `ParseException` with the offending `SqlNode`) instead of emitting an empty result term.
##########
flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/TransformParser.java:
##########
@@ -589,6 +598,27 @@ public static String translateFilterExpressionToJaninoExpression(
where);
}
+ public static GeneratedExpression translateFilterExpressionToGeneratedExpression(
+ String filterExpression,
+ List<Column> columns,
+ List<UserDefinedFunctionDescriptor> udfDescriptors,
+ SupportedMetadataColumn[] supportedMetadataColumns,
+ Map<String, String> columnNameMap) {
+ if (isNullOrWhitespaceOnly(filterExpression)) {
+ return GeneratedExpression.fromExpression("", Boolean.class);
+ }
+ SqlSelect sqlSelect = TransformParser.parseFilterExpression(filterExpression);
+ if (!sqlSelect.hasWhere()) {
+ return GeneratedExpression.fromExpression("", Boolean.class);
+ }
Review Comment:
Returning an empty-result `GeneratedExpression` for empty filters / missing WHERE can conflict with `TransformExpressionKey.getFullScript()` validation (which rejects empty result terms) and may break compilation paths that still attempt to compile the filter. If the intended semantics are 'no filter means pass-through', consider returning a non-empty constant script term (e.g., `Boolean.TRUE`) or handling the no-op case by skipping key creation/compilation entirely so the rest of the pipeline never sees an empty result term.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
Posted by "haruki-830 (via GitHub)" <gi...@apache.org>.
haruki-830 commented on PR #4482:
URL: https://github.com/apache/flink-cdc/pull/4482#issuecomment-5041653096
@lvyanquan PTAL when you have time, Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
Posted by "haruki-830 (via GitHub)" <gi...@apache.org>.
haruki-830 commented on PR #4482:
URL: https://github.com/apache/flink-cdc/pull/4482#issuecomment-5091010527
>
> > rvalue.toString() is for debugging purposes only and may not generate Java code with correct parentheses.
> > For example, SQL statement (a - b) / (c + 1) will be converted to a - b / c + 1.
>
> Seems this is an existing bug, not introduced by changes in this PR. Could you please investigate this later?
Thanks for pointing out this bug! I confirmed that `BinaryOperation.toString()` does lose parentheses when a sub-expression has lower precedence than its parent operator. This affects the existing `translateSqlNodeToJaninoExpression` path as well.
I'll open a separate PR to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-40204][runtime] Introduce statement-level codegen for YAML Transform expressions [flink-cdc]
Posted by "haruki-830 (via GitHub)" <gi...@apache.org>.
haruki-830 commented on PR #4482:
URL: https://github.com/apache/flink-cdc/pull/4482#issuecomment-5088084664
Thanks @lvyanquan and @yuxiqian for the feedback.
I ran a JMH benchmark comparing the Supplier path against an equivalent if/else script and found no statistically significant difference — all configurations fall in the 330k-372k ops/ms range with overlapping confidence intervals, that means the JIT escape analysis reliably eliminates the anonymous `Supplier` allocation, so the per-record allocation concern is not a real bottleneck in practice.
Given the lack of measurable benefit and the correctness risks @yuxiqian identified, I agree this change is not warranted. I'll go ahead and close this PR later.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org