This is an automated email from the ASF dual-hosted git repository.
twalthr pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new 9d49efb4a34 [FLINK-39679][table-planner] Fix implicit casts for PTF
calls
9d49efb4a34 is described below
commit 9d49efb4a34927315b2c6c21fb346020c80bb199
Author: Timo Walther <[email protected]>
AuthorDate: Tue Jun 16 16:26:48 2026 +0200
[FLINK-39679][table-planner] Fix implicit casts for PTF calls
This closes #28460.
---
.../functions/casting/RowToRowCastRule.java | 3 ++
.../codegen/ProcessTableRunnerGenerator.scala | 56 ++++++++++++++++++----
.../stream/ProcessTableFunctionSemanticTests.java | 3 +-
.../stream/ProcessTableFunctionTestPrograms.java | 21 ++++++++
.../exec/stream/ProcessTableFunctionTestUtils.java | 25 ++++++++++
5 files changed, 98 insertions(+), 10 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/RowToRowCastRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/RowToRowCastRule.java
index 119a3b4a6b9..2b0d05be177 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/RowToRowCastRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/casting/RowToRowCastRule.java
@@ -40,6 +40,9 @@ import static
org.apache.flink.table.planner.functions.casting.CastRuleUtils.met
/**
* {@link LogicalTypeRoot#ROW} and {@link LogicalTypeRoot#STRUCTURED_TYPE} to
{@link
* LogicalTypeRoot#ROW} and {@link LogicalTypeRoot#STRUCTURED_TYPE} cast rule.
+ *
+ * <p>This rule is used for arbitrary nested ROW/STRUCTURED casts where
carrying RowKind would be
+ * semantically wrong. This should be rather done on the caller side.
*/
class RowToRowCastRule extends AbstractNullAwareCodeGeneratorCastRule<RowData,
RowData> {
diff --git
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
index 933ea65a5c4..c6b6534be45 100644
---
a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
+++
b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/ProcessTableRunnerGenerator.scala
@@ -30,7 +30,7 @@ import
org.apache.flink.table.planner.calcite.{FlinkTypeFactory, RexTableArgCall
import org.apache.flink.table.planner.codegen.CodeGenUtils._
import org.apache.flink.table.planner.codegen.GeneratedExpression.{NEVER_NULL,
NO_CODE}
import org.apache.flink.table.planner.codegen.Indenter.toISC
-import org.apache.flink.table.planner.codegen.calls.BridgingFunctionGenUtil
+import org.apache.flink.table.planner.codegen.calls.{BridgingFunctionGenUtil,
ScalarOperatorGens}
import
org.apache.flink.table.planner.codegen.calls.BridgingFunctionGenUtil.{verifyFunctionAwareOutputType,
DefaultExpressionEvaluatorFactory}
import org.apache.flink.table.planner.delegation.PlannerBase
import org.apache.flink.table.planner.functions.bridging.BridgingSqlFunction
@@ -45,7 +45,7 @@ import org.apache.flink.table.types.extraction.ExtractionUtils
import org.apache.flink.table.types.inference.TypeInferenceUtil
import org.apache.flink.table.types.inference.TypeInferenceUtil.StateInfo
import org.apache.flink.table.types.logical.LogicalType
-import org.apache.flink.table.types.logical.utils.LogicalTypeChecks
+import org.apache.flink.table.types.logical.utils.{LogicalTypeCasts,
LogicalTypeChecks}
import org.apache.flink.types.Row
import org.apache.calcite.rex.{RexCall, RexNode}
@@ -321,24 +321,61 @@ object ProcessTableRunnerGenerator {
private def generateEvalOperands(
ctx: CodeGeneratorContext,
call: RexCall,
+ enrichedArgumentDataTypes: Seq[DataType],
inputIndexTerm: String,
inputRowTerm: String): Seq[GeneratedExpression] = {
val generator = new ExprCodeGenerator(ctx, false)
+
+ // Generate (potentially cast) operands
call.getOperands.asScala
+ .zip(enrichedArgumentDataTypes)
.map {
- case tableArgCall: RexTableArgCall =>
+ case (tableArgCall: RexTableArgCall, targetDataType: DataType) =>
val inputIndex = tableArgCall.getInputIndex
- val tableType =
FlinkTypeFactory.toLogicalType(call.getType).copy(true)
- GeneratedExpression(
+ val tableType =
FlinkTypeFactory.toLogicalType(tableArgCall.getType).copy(true)
+ val tableArgAccess = GeneratedExpression(
s"$inputRowTerm",
- s"$inputIndexTerm != $inputIndex",
+ s"($inputIndexTerm != $inputIndex)",
NO_CODE,
tableType)
- case rexNode: RexNode =>
- generator.generateExpression(rexNode)
+ val (castTableArg, hasCast) =
+ castOperand(ctx, tableArgAccess, targetDataType.getLogicalType)
+ val tableArgWithKind = if (hasCast) {
+ // Casting a table argument to the target type (potentially)
produces a fresh row that
+ // defaults to RowKind.INSERT. We forward the table's underlying
changelog
+ // information to the eval() here.
+ val rowKindCode =
+ s"""
+ |if (!${castTableArg.nullTerm}) {
+ |
${castTableArg.resultTerm}.setRowKind($inputRowTerm.getRowKind());
+ |}
+ |""".stripMargin
+ castTableArg.copy(code = s"${castTableArg.code}\n$rowKindCode")
+ } else {
+ tableArgAccess
+ }
+ tableArgWithKind
+
+ case (rexNode: RexNode, targetDataType: DataType) =>
+ val scalarArg = generator.generateExpression(rexNode)
+ val (castScalarArg, _) = castOperand(ctx, scalarArg,
targetDataType.getLogicalType)
+ castScalarArg
}
}
+ /** Applies a cast if necessary. Also returns whether a cast was necessary.
*/
+ private def castOperand(
+ ctx: CodeGeneratorContext,
+ operand: GeneratedExpression,
+ targetType: LogicalType): (GeneratedExpression, Boolean) = {
+ if (LogicalTypeCasts.supportsAvoidingCast(operand.resultType, targetType))
{
+ return (operand, false)
+ }
+ val cast =
+ ScalarOperatorGens.generateCast(ctx, operand, targetType, nullOnFailure
= false)
+ (cast, true)
+ }
+
private def verifyMethodImplementation(
methodName: String,
contextClass: Class[_],
@@ -388,7 +425,8 @@ object ProcessTableRunnerGenerator {
val inputIndexTerm = "inputIndex"
val inputRowTerm = "inputRow"
val collectorTerm = "evalCollector"
- val callOperands = generateEvalOperands(ctx, call, inputIndexTerm,
inputRowTerm)
+ val callOperands =
+ generateEvalOperands(ctx, call, enrichedArgumentDataTypes,
inputIndexTerm, inputRowTerm)
val externalCallOperands =
BridgingFunctionGenUtil.prepareExternalOperands(ctx, callOperands,
enrichedArgumentDataTypes)
val allExternalOperands = externalStateOperands ++ externalCallOperands
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
index aa547037a13..e2b5e76167a 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionSemanticTests.java
@@ -104,6 +104,7 @@ public class ProcessTableFunctionSemanticTests extends
SemanticTestBase {
ProcessTableFunctionTestPrograms.PROCESS_UPDATING_MULTI_INPUT,
ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY,
ProcessTableFunctionTestPrograms.PROCESS_MULTI_INPUT_ORDER_BY,
- ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY_TABLE_API);
+ ProcessTableFunctionTestPrograms.PROCESS_ORDER_BY_TABLE_API,
+ ProcessTableFunctionTestPrograms.PROCESS_IMPLICIT_CASTS);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
index 062de5c2416..887eab68f45 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestPrograms.java
@@ -28,6 +28,7 @@ import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctio
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ContextFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.DescriptorFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.EmptyArgFunction;
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.ImplicitCastingFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.IntervalDayArgFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.IntervalYearArgFunction;
import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.InvalidPassThroughTimersFunction;
@@ -1920,4 +1921,24 @@ public class ProcessTableFunctionTestPrograms {
+ "in2 => TABLE t_large PARTITION BY name
ORDER BY (ts ASC, score DESC), "
+ "on_time => DESCRIPTOR(ts))")
.build();
+
+ public static final TableTestProgram PROCESS_IMPLICIT_CASTS =
+ TableTestProgram.of(
+ "process-implicit-casts",
+ "implicits casts for both scalar and table args")
+ .setupTemporarySystemFunction("f",
ImplicitCastingFunction.class)
+ .setupSql(
+ "CREATE VIEW v AS SELECT 'Bob', 12, (42, TIMESTAMP
'2020-01-01 12:12:12')")
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema(BASE_SINK_SCHEMA)
+ .consumedValues(
+ "+I[{CastingPojo(s='Bob', b=12,
r=+I[42, 2020-01-01T12:12:12]), 42}]")
+ .build())
+ // The function expects wider types (BIGINT and
TIMESTAMP(6)),
+ // implicit casts are added to fix the mismatch between
INT vs. BIGINT and
+ // TIMESTAMP(0) vs. TIMESTAMP(6).
+ // Also in constructed types: ROW (table input) vs.
STRUCTURED (expected).
+ .runSql("INSERT INTO sink SELECT * FROM f(p => TABLE v, b
=> 42)")
+ .build();
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
index 495293c9244..8a75cf35416 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/ProcessTableFunctionTestUtils.java
@@ -1148,6 +1148,13 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /** Forces implicit casts on the input. */
+ public static class ImplicitCastingFunction extends
AppendProcessTableFunctionBase {
+ public void eval(@ArgumentHint(ROW_SEMANTIC_TABLE) CastingPojo p, long
b) {
+ collectObjects(p, b);
+ }
+ }
+
//
--------------------------------------------------------------------------------------------
// Helpers
//
--------------------------------------------------------------------------------------------
@@ -1207,4 +1214,22 @@ public class ProcessTableFunctionTestUtils {
private static String toModeSummary(ChangelogMode mode) {
return MODE_SUMMARY.get(mode.toString());
}
+
+ /** POJO expecting BIGINT. */
+ public static class CastingPojo {
+ public String s;
+ public long b;
+ public @DataTypeHint("ROW<b BIGINT, t TIMESTAMP(6)>") Row r;
+
+ public CastingPojo(String s, long b, Row r) {
+ this.s = s;
+ this.b = b;
+ this.r = r;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("CastingPojo(s='%s', b=%s, r=%s)", s, b, r);
+ }
+ }
}
