This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new e9001f6a85c [FLINK-38233][table] Adds support to
StreamNonDeterministicUpdatePlanVisitor for PTFs
e9001f6a85c is described below
commit e9001f6a85ca1c4f4a7cc8a275d8dcbaf7f98e77
Author: Alan Sheinberg <[email protected]>
AuthorDate: Wed Jul 1 01:20:26 2026 -0700
[FLINK-38233][table] Adds support to
StreamNonDeterministicUpdatePlanVisitor for PTFs
This closes #28061.
---
.../StreamNonDeterministicUpdatePlanVisitor.java | 88 ++++++++
.../table/api/QueryOperationTestPrograms.java | 16 --
.../stream/ProcessTableFunctionSemanticTests.java | 12 -
.../exec/stream/ProcessTableFunctionTestUtils.java | 21 ++
.../plan/stream/sql/NonDeterministicDagTest.scala | 244 +++++++++++++++++++++
5 files changed, 353 insertions(+), 28 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java
index 99016be46fa..344352b958f 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java
@@ -26,6 +26,7 @@ import
org.apache.flink.table.connector.source.abilities.SupportsReadingMetadata
import org.apache.flink.table.legacy.api.TableSchema;
import org.apache.flink.table.legacy.api.constraints.UniqueConstraint;
import org.apache.flink.table.planner.calcite.FlinkTypeFactory;
+import org.apache.flink.table.planner.calcite.RexTableArgCall;
import org.apache.flink.table.planner.connectors.DynamicSourceUtils;
import org.apache.flink.table.planner.plan.metadata.FlinkRelMetadataQuery;
import org.apache.flink.table.planner.plan.nodes.exec.spec.OverSpec;
@@ -48,6 +49,7 @@ import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalM
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalMiniBatchAssigner;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalMultiJoin;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalOverAggregateBase;
+import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalProcessTableFunction;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalRank;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalRel;
import
org.apache.flink.table.planner.plan.nodes.physical.stream.StreamPhysicalSink;
@@ -75,12 +77,16 @@ import org.apache.flink.table.planner.utils.ShortcutUtils;
import org.apache.flink.table.runtime.operators.join.FlinkJoinType;
import
org.apache.flink.table.runtime.operators.join.stream.utils.JoinInputSideSpec;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.inference.StaticArgument;
+import org.apache.flink.table.types.inference.StaticArgumentTrait;
import org.apache.flink.types.RowKind;
+import org.apache.calcite.linq4j.Ord;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.AggregateCall;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexProgram;
import org.apache.calcite.sql.SqlExplainLevel;
@@ -218,6 +224,8 @@ public class StreamNonDeterministicUpdatePlanVisitor {
return transmitDeterminismRequirement(rel, requireDeterminism);
} else if (rel instanceof StreamPhysicalMatch) {
return visitMatch((StreamPhysicalMatch) rel, requireDeterminism);
+ } else if (rel instanceof StreamPhysicalProcessTableFunction) {
+ return visitPtf((StreamPhysicalProcessTableFunction) rel,
requireDeterminism);
} else {
throw new UnsupportedOperationException(
String.format(
@@ -958,6 +966,86 @@ public class StreamNonDeterministicUpdatePlanVisitor {
return transmitDeterminismRequirement(rel, inputRequireDeterminism);
}
+ private StreamPhysicalRel visitPtf(
+ final StreamPhysicalProcessTableFunction ptf,
+ final ImmutableBitSet requireDeterminism) {
+ final RexCall call = ptf.getCall();
+
+ // Concern 1: PTF function itself is non-deterministic and downstream
nodes
+ // require determinism.
+ if (!requireDeterminism.isEmpty()) {
+ final Optional<String> ndCall =
FlinkRexUtil.getNonDeterministicCallName(call);
+ if (ndCall.isPresent()) {
+ throwNonDeterministicColumnsError(
+ requireDeterminism.toList(), ptf.getRowType(), ptf,
null, ndCall);
+ }
+ }
+
+ if (inputInsertOnly(ptf)) {
+ // No retracts arrive at the PTF input, so input-column
determinism does
+ // not affect retract correctness
+ return transmitDeterminismRequirement(ptf,
NO_REQUIRED_DETERMINISM);
+ }
+
+ // Concern 2: non-deterministic input columns.
+ // The PTF is a black box: there is no way to project downstream
column requirements
+ // (requireDeterminism) back through the PTF's internal computation to
determine which
+ // input columns need to be deterministic. requireDeterminism is
therefore ignored here.
+ //
+ // A stricter alternative would be to require all input columns to be
deterministic
+ // whenever any output column downstream requires it. That avoids the
gap described
+ // below but rejects legitimate queries where the PTF does not
actually consume the
+ // non-deterministic column, so we keep the lenient behavior.
+ //
+ // Known gap: cases like the one below are not covered today and may
produce wrong
+ // results on failover, since the requirement on {@code nb} at the
retract sink never
+ // reaches {@code ndFunc(b)} upstream of the PTF.
+ //
+ // <pre>{@code
+ // CREATE VIEW v AS SELECT a, ndFunc(b) AS nb FROM upsert_src;
+ // INSERT INTO retract_sink SELECT * FROM rowPtf(TABLE v);
-- row-semantic
+ // INSERT INTO retract_sink SELECT * FROM setPtf(TABLE v PARTITION BY
a); -- set-semantic
+ // }</pre>
+ //
+ // Note: the physical changelog the PTF receives is not solely
determined by the
+ // REQUIRE_UPDATE_BEFORE trait. Per the SUPPORT_UPDATES contract, the
function receives
+ // {+I,+U,-D} only when the input is upserting on the same key as the
partition key;
+ // otherwise it receives full retractions {+I,-U,+U,-D}, including UB
messages, even
+ // without REQUIRE_UPDATE_BEFORE. UBs may therefore arrive regardless
of the trait.
+ //
+ // What the trait controls is which messages the PTF is contractually
allowed to
+ // consume for state management, and that is what drives the
determinism requirement:
+ // - No REQUIRE_UPDATE_BEFORE: the PTF does not consume UBs; retract
handling is
+ // keyed by partition key, so only partition key columns must be
deterministic.
+ // Any non-key columns on an incidentally-delivered UB are not
used by the PTF.
+ // - Has REQUIRE_UPDATE_BEFORE: the PTF explicitly opts in to
consuming UB to
+ // reconstruct the previously processed row; that row must match
exactly, so all
+ // input columns must be deterministic.
+ final List<Ord<StaticArgument>> providedInputArgs =
+ StreamPhysicalProcessTableFunction.getProvidedInputArgs(call);
+ final List<RexNode> operands = call.getOperands();
+
+ final List<RelNode> newInputs = new ArrayList<>();
+ for (int i = 0; i < ptf.getInputs().size(); i++) {
+ final StreamPhysicalRel input = (StreamPhysicalRel)
ptf.getInput(i);
+ final StaticArgument staticArg = providedInputArgs.get(i).e;
+ final RexTableArgCall tableArgCall =
+ (RexTableArgCall) operands.get(providedInputArgs.get(i).i);
+ final ImmutableBitSet inputReq;
+ if (staticArg.is(StaticArgumentTrait.REQUIRE_UPDATE_BEFORE)) {
+ // The PTF consumes UB to reconstruct the previously processed
row, which must
+ // match exactly — all input columns must be deterministic.
+ inputReq =
ImmutableBitSet.range(input.getRowType().getFieldCount());
+ } else {
+ // The PTF does not consume UB; retract handling is keyed by
partition key, so
+ // only partition key columns must be deterministic.
+ inputReq = ImmutableBitSet.of(tableArgCall.getPartitionKeys());
+ }
+ newInputs.add(visit(input,
requireDeterminismExcludeUpsertKey(input, inputReq)));
+ }
+ return (StreamPhysicalRel) ptf.copy(ptf.getTraitSet(), newInputs);
+ }
+
private void checkNonDeterministicRexProgram(
final ImmutableBitSet requireDeterminism,
final RexProgram program,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
index 03b8a73ed98..e8671f0da73 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
@@ -21,7 +21,6 @@ package org.apache.flink.table.api;
import org.apache.flink.annotation.Internal;
import org.apache.flink.table.api.config.ExecutionConfigOptions;
import
org.apache.flink.table.api.config.ExecutionConfigOptions.AsyncOutputMode;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.table.planner.factories.TestValuesModelFactory;
@@ -1059,11 +1058,6 @@ public class QueryOperationTestPrograms {
public static final TableTestProgram PTF_ROW_SEMANTIC_TABLE =
TableTestProgram.of("ptf-row-semantic-table", "table with row
semantics")
- // TODO [FLINK-38233]: Remove this config when PTF support
in
- // StreamNonDeterministicUpdatePlanVisitor is added.
- .setupConfig(
-
OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY,
-
OptimizerConfigOptions.NonDeterministicUpdateStrategy.IGNORE)
.setupTemporarySystemFunction("f",
RowSemanticTableFunction.class)
.setupSql(BASIC_VALUES)
.setupTableSink(
@@ -1089,11 +1083,6 @@ public class QueryOperationTestPrograms {
static final TableTestProgram PTF_SET_SEMANTIC_TABLE =
TableTestProgram.of("ptf-set-semantic-table", "verifies SQL
serialization")
- // TODO [FLINK-38233]: Remove this config when PTF support
in
- // StreamNonDeterministicUpdatePlanVisitor is added.
- .setupConfig(
-
OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY,
-
OptimizerConfigOptions.NonDeterministicUpdateStrategy.IGNORE)
.setupTemporarySystemFunction("f1",
ChainedSendingFunction.class)
.setupTemporarySystemFunction("f2",
ChainedReceivingFunction.class)
.setupTableSource(TIMED_SOURCE)
@@ -1321,11 +1310,6 @@ public class QueryOperationTestPrograms {
static final TableTestProgram PTF_ORDER_BY =
TableTestProgram.of("ptf-order-by", "verifies SQL serialization
with ORDER BY clause")
- // TODO [FLINK-38233]: Remove this config when PTF support
in
- // StreamNonDeterministicUpdatePlanVisitor is added.
- .setupConfig(
-
OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY,
-
OptimizerConfigOptions.NonDeterministicUpdateStrategy.IGNORE)
.setupTemporarySystemFunction("f",
SetSemanticTableFunction.class)
.setupTableSource(TIMED_SOURCE)
.setupTableSink(
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 0cdc6d9c243..39131ff4ed3 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
@@ -18,8 +18,6 @@
package org.apache.flink.table.planner.plan.nodes.exec.stream;
-import org.apache.flink.table.api.TableConfig;
-import org.apache.flink.table.api.config.OptimizerConfigOptions;
import
org.apache.flink.table.planner.plan.nodes.exec.testutils.SemanticTestBase;
import org.apache.flink.table.test.program.TableTestProgram;
@@ -28,16 +26,6 @@ import java.util.List;
/** Semantic tests for {@link StreamExecProcessTableFunction}. */
public class ProcessTableFunctionSemanticTests extends SemanticTestBase {
- // TODO [FLINK-38233]: Remove this override when PTF support in
- // StreamNonDeterministicUpdatePlanVisitor is added.
- @Override
- protected void applyDefaultEnvironmentOptions(TableConfig config) {
- super.applyDefaultEnvironmentOptions(config);
- config.set(
-
OptimizerConfigOptions.TABLE_OPTIMIZER_NONDETERMINISTIC_UPDATE_STRATEGY,
- OptimizerConfigOptions.NonDeterministicUpdateStrategy.IGNORE);
- }
-
@Override
public List<TableTestProgram> programs() {
return List.of(
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 aaf1e71185a..89fc127a413 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
@@ -952,6 +952,18 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /**
+ * Testing function that is itself non-deterministic (isDeterministic() =
false). Used to verify
+ * that Concern 1 (PTF own non-determinism) is caught by the NDU visitor
when downstream
+ * requires deterministic output columns.
+ */
+ public static class NonDeterministicUpdatingRetractFunction extends
UpdatingRetractFunction {
+ @Override
+ public boolean isDeterministic() {
+ return false;
+ }
+ }
+
/** Testing function. */
public static class UpdatingUpsertFullDeletesFunction
extends ChangelogProcessTableFunctionBase {
@@ -994,6 +1006,15 @@ public class ProcessTableFunctionTestUtils {
}
}
+ /** Row-semantic counterpart of {@link
NonDeterministicUpdatingRetractFunction}. */
+ public static class NonDeterministicUpdatingRetractRowSemanticFunction
+ extends UpdatingRetractRowSemanticFunction {
+ @Override
+ public boolean isDeterministic() {
+ return false;
+ }
+ }
+
/** Testing function. */
public static class InvalidRowKindFunction extends
AppendProcessTableFunctionBase {
public void eval(@ArgumentHint(ROW_SEMANTIC_TABLE) Row r) {
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala
index aa53283add5..24efc06a7f5 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/stream/sql/NonDeterministicDagTest.scala
@@ -28,6 +28,7 @@ import org.apache.flink.table.data.RowData
import org.apache.flink.table.legacy.api.TableSchema
import org.apache.flink.table.planner.JBoolean
import
org.apache.flink.table.planner.expressions.utils.{TestNonDeterministicUdaf,
TestNonDeterministicUdf, TestNonDeterministicUdtf}
+import
org.apache.flink.table.planner.plan.nodes.exec.stream.ProcessTableFunctionTestUtils.{NonDeterministicUpdatingRetractFunction,
NonDeterministicUpdatingRetractRowSemanticFunction, UpdatingJoinFunction,
UpdatingRetractFunction, UpdatingRetractRowSemanticFunction,
UpdatingUpsertFunction}
import
org.apache.flink.table.planner.runtime.utils.JavaUserDefinedTableFunctions.StringSplit
import org.apache.flink.table.planner.utils.{StreamTableTestUtil,
TableTestBase}
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo
@@ -231,6 +232,101 @@ class
NonDeterministicDagTest(nonDeterministicUpdateStrategy: NonDeterministicUp
util.tableEnv.createTemporaryFunction("ndAggFunc", new
TestNonDeterministicUdaf)
// deterministic table function
util.tableEnv.createTemporaryFunction("str_split", new StringSplit())
+
+ // PTF functions for NDU changelog tests
+ util.tableEnv.createTemporaryFunction("updatingUpsertFunc", new
UpdatingUpsertFunction)
+ util.tableEnv.createTemporaryFunction("updatingRetractFunc", new
UpdatingRetractFunction)
+ util.tableEnv.createTemporaryFunction(
+ "nonDeterministicRetractFunc",
+ new NonDeterministicUpdatingRetractFunction)
+ // PTF taking two table arguments (multi-table case)
+ util.tableEnv.createTemporaryFunction("updatingJoinFunc", new
UpdatingJoinFunction)
+ // Row-semantic PTFs
+ util.tableEnv.createTemporaryFunction(
+ "updatingRetractRowSemanticFunc",
+ new UpdatingRetractRowSemanticFunction)
+ util.tableEnv.createTemporaryFunction(
+ "nonDeterministicRetractRowSemanticFunc",
+ new NonDeterministicUpdatingRetractRowSemanticFunction)
+
+ // View with ndFunc applied to the partition key column (INT -> INT)
+ util.tableEnv.executeSql("""CREATE VIEW nd_partition_key_src AS
+ |SELECT ndFunc(a) AS nd_a, b, c, d FROM
upsert_src""".stripMargin)
+
+ // View with ndFunc applied to a non-key column (BIGINT -> BIGINT)
+ util.tableEnv.executeSql("""CREATE VIEW nd_col_src AS
+ |SELECT a, ndFunc(b) AS nd_b, c, d FROM
cdc""".stripMargin)
+
+ // Upsert sink: 4 columns (INT, INT, BIGINT, STRING) matching PTF upsert
output over INT sources
+ util.tableEnv.executeSql("""CREATE TABLE ptf_upsert_sink (
+ | f0 INT, f1 STRING, f2 BIGINT, f3 STRING,
+ | PRIMARY KEY (f0) NOT ENFORCED
+ |) WITH (
+ | 'connector' = 'values',
+ | 'sink-insert-only' = 'false',
+ | 'sink-changelog-mode-enforced' = 'I,UA,D'
+ |)""".stripMargin)
+
+ // Retract sink: same column types, no primary key
+ util.tableEnv.executeSql("""CREATE TABLE ptf_retract_sink (
+ | f0 INT, f1 STRING, f2 BIGINT, f3 STRING
+ |) WITH (
+ | 'connector' = 'values',
+ | 'sink-insert-only' = 'false'
+ |)""".stripMargin)
+
+ // Two upsert sources for the multi-table (join) PTF, both keyed by id.
+ util.tableEnv.executeSql("""CREATE TABLE ptf_score_src (
+ | id INT,
+ | score INT,
+ | PRIMARY KEY (id) NOT ENFORCED
+ |) WITH (
+ | 'connector' = 'values',
+ | 'changelog-mode' = 'I,UA,D',
+ | 'source.produces-delete-by-key' = 'true'
+ |)""".stripMargin)
+ util.tableEnv.executeSql("""CREATE TABLE ptf_city_src (
+ | id INT,
+ | city STRING,
+ | PRIMARY KEY (id) NOT ENFORCED
+ |) WITH (
+ | 'connector' = 'values',
+ | 'changelog-mode' = 'I,UA,D',
+ | 'source.produces-delete-by-key' = 'true'
+ |)""".stripMargin)
+
+ // Views applying ndFunc to each source's key column, so its upsert key no
longer matches the
+ // partition key used by the PTF.
+ util.tableEnv.executeSql("""CREATE VIEW nd_score_src AS
+ |SELECT ndFunc(id) AS id, score FROM
ptf_score_src""".stripMargin)
+ util.tableEnv.executeSql("""CREATE VIEW nd_city_src AS
+ |SELECT ndFunc(id) AS id, city FROM
ptf_city_src""".stripMargin)
+
+ // Upsert sink for the join PTF output: the two table arguments are
co-partitioned on the same
+ // key, so a single partition key (id) is prepended to the PTF output.
+ util.tableEnv.executeSql("""CREATE TABLE ptf_join_sink (
+ | id INT,
+ | `out` STRING,
+ | PRIMARY KEY (id) NOT ENFORCED
+ |) WITH (
+ | 'connector' = 'values',
+ | 'sink-insert-only' = 'false',
+ | 'sink-changelog-mode-enforced' = 'I,UA,D',
+ | 'sink.supports-delete-by-key' = 'true'
+ |)""".stripMargin)
+
+ // Retract sink matching the row-semantic changelog PTF output ROW<name,
count, mode>, no PK.
+ util.tableEnv.executeSql("""CREATE TABLE ptf_row_retract_sink (
+ | f0 STRING, f1 BIGINT, f2 STRING
+ |) WITH (
+ | 'connector' = 'values',
+ | 'sink-insert-only' = 'false'
+ |)""".stripMargin)
+
+ // Insert-only source with a non-deterministic column. Feeding this to a
row-semantic PTF
+ // exercises the inputInsertOnly short-circuit in visitPtf.
+ util.tableEnv.executeSql("""CREATE VIEW nd_insert_only_src AS
+ |SELECT a, ndFunc(b) AS nd_b, c, d FROM
src""".stripMargin)
}
@TestTemplate
@@ -1962,6 +2058,154 @@ class
NonDeterministicDagTest(nonDeterministicUpdateStrategy: NonDeterministicUp
.isInstanceOf[TableException]
}
+ //
---------------------------------------------------------------------------
+ // PTF changelog NDU tests
+ //
---------------------------------------------------------------------------
+
+ /** Partition key equals the source upsert key — requirement excluded, no
NDU error. */
+ @TestTemplate
+ def testPtfUpsertOutputPartitionKeyMatchesUpsertKey(): Unit = {
+ assertThatCode(
+ () =>
+ util.tableEnv.explainSql(
+ "INSERT INTO ptf_upsert_sink" +
+ " SELECT * FROM updatingUpsertFunc(TABLE upsert_src PARTITION BY
a)"))
+ .doesNotThrowAnyException()
+ }
+
+ /**
+ * Non-deterministic function on the partition key column — the partition
key of the PTF input
+ * must be deterministic for upsert output correctness; expect error only
under TRY_RESOLVE.
+ */
+ @TestTemplate
+ def testPtfUpsertOutputNonDeterministicPartitionKey(): Unit = {
+ val callable: ThrowingCallable = () =>
+ util.tableEnv.explainSql(
+ "INSERT INTO ptf_upsert_sink" +
+ " SELECT * FROM updatingUpsertFunc(TABLE nd_partition_key_src
PARTITION BY nd_a)")
+ if (tryResolve) {
+ assertThatThrownBy(callable)
+ .hasMessageContaining("non-deterministic function: ndFunc")
+ .isInstanceOf[TableException]
+ } else {
+ assertThatCode(callable).doesNotThrowAnyException()
+ }
+ }
+
+ /** All CDC input columns are deterministic — full-retract output requires
no NDU error. */
+ @TestTemplate
+ def testPtfRetractOutputDeterministicInput(): Unit = {
+ assertThatCode(
+ () =>
+ util.tableEnv.explainSql(
+ "INSERT INTO ptf_retract_sink" +
+ " SELECT * FROM updatingRetractFunc(TABLE cdc PARTITION BY a)"))
+ .doesNotThrowAnyException()
+ }
+
+ /**
+ * Non-deterministic function on a non-key input column — full-retract
output requires all columns
+ * to be deterministic; expect error only under TRY_RESOLVE.
+ */
+ @TestTemplate
+ def testPtfRetractOutputNonDeterministicInputColumn(): Unit = {
+ val callable: ThrowingCallable = () =>
+ util.tableEnv.explainSql(
+ "INSERT INTO ptf_retract_sink" +
+ " SELECT * FROM updatingRetractFunc(TABLE nd_col_src PARTITION BY
a)")
+ if (tryResolve) {
+ assertThatThrownBy(callable)
+ .hasMessageContaining("non-deterministic function: ndFunc")
+ .isInstanceOf[TableException]
+ } else {
+ assertThatCode(callable).doesNotThrowAnyException()
+ }
+ }
+
+ /**
+ * A non-deterministic PTF writing to a retract sink (no PK): the retract
sink requires all PTF
+ * output columns to be deterministic, and the retract PTF has no upsert key
so nothing is zeroed
+ * by requireDeterminismExcludeUpsertKey — requireDeterminism is fully
non-empty when it reaches
+ * visitPtf. The PTF claims isDeterministic() == false (Concern 1), so the
NDU visitor must flag
+ * it.
+ *
+ * <p>Inputs are deterministic (cdc with no ndFunc), so Concern 2
(non-deterministic input
+ * columns) does not trigger.
+ */
+ @TestTemplate
+ def testPtfNonDeterministicFunctionWithRequiredOutputDeterminism(): Unit = {
+ val callable: ThrowingCallable = () =>
+ util.tableEnv.explainSql(
+ "INSERT INTO ptf_retract_sink" +
+ " SELECT * FROM nonDeterministicRetractFunc(TABLE cdc PARTITION BY
a)")
+ if (tryResolve) {
+ assertThatThrownBy(callable)
+ .hasMessageContaining("non-deterministic function")
+ .isInstanceOf[TableException]
+ } else {
+ assertThatCode(callable).doesNotThrowAnyException()
+ }
+ }
+
+ /** Multi-table PTF: both table arguments are partitioned by their
respective upsert keys. */
+ @TestTemplate
+ def testPtfMultiInputUpsertOutputPartitionKeyMatchesUpsertKey(): Unit = {
+ assertThatCode(
+ () =>
+ util.tableEnv.explainSql(
+ "INSERT INTO ptf_join_sink SELECT id, `out` FROM updatingJoinFunc(" +
+ "scoreTable => TABLE ptf_score_src PARTITION BY id, " +
+ "cityTable => TABLE ptf_city_src PARTITION BY id)"))
+ .doesNotThrowAnyException()
+ }
+
+ /**
+ * Multi-table (join) PTF where one table argument is partitioned by a
non-deterministic column
+ * (ndFunc(id)) that no longer matches the source's upsert key.
+ */
+ @TestTemplate
+ def testPtfMultiInputUpsertOutputNonDeterministicPartitionKey(): Unit = {
+ val callable: ThrowingCallable = () =>
+ util.tableEnv.explainSql(
+ "INSERT INTO ptf_join_sink SELECT id, `out` FROM updatingJoinFunc(" +
+ "scoreTable => TABLE nd_score_src PARTITION BY id, " +
+ "cityTable => TABLE ptf_city_src PARTITION BY id)")
+ if (tryResolve) {
+ assertThatThrownBy(callable)
+ .hasMessageContaining("non-deterministic function: ndFunc")
+ .isInstanceOf[TableException]
+ } else {
+ assertThatCode(callable).doesNotThrowAnyException()
+ }
+ }
+
+ /** Row-semantic PTF that is itself non-deterministic, writing to a retract
sink. */
+ @TestTemplate
+ def testPtfRowSemanticNonDeterministicFunctionWithRetractSink(): Unit = {
+ val callable: ThrowingCallable = () =>
+ util.tableEnv.explainSql(
+ "INSERT INTO ptf_row_retract_sink" +
+ " SELECT * FROM nonDeterministicRetractRowSemanticFunc(TABLE cdc)")
+ if (tryResolve) {
+ assertThatThrownBy(callable)
+ .hasMessageContaining("non-deterministic function")
+ .isInstanceOf[TableException]
+ } else {
+ assertThatCode(callable).doesNotThrowAnyException()
+ }
+ }
+
+ /** Row-semantic PTF reading an insert-only input that carries a
non-deterministic column. */
+ @TestTemplate
+ def testPtfRowSemanticInsertOnlyInputWithRetractSink(): Unit = {
+ assertThatCode(
+ () =>
+ util.tableEnv.explainSql(
+ "INSERT INTO ptf_row_retract_sink" +
+ " SELECT * FROM updatingRetractRowSemanticFunc(TABLE
nd_insert_only_src)"))
+ .doesNotThrowAnyException()
+ }
+
/**
* This upsert test sink does support getting primary key from table schema.
We defined a similar
* test sink here not using existing {@link TestingUpsertTableSink} in
{@link StreamTestSink}
