gustavodemorais commented on code in PR #28061:
URL: https://github.com/apache/flink/pull/28061#discussion_r3162163077
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java:
##########
@@ -958,6 +966,71 @@ private StreamPhysicalRel visitJoinChild(
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.
+ // requireDeterminism is the set of PTF output columns downstream
requires to be
+ // deterministic (for correct retract matching). All PTF output
columns are assumed to
+ // come from PTF's computation — there are pass-through input columns
potentially, but
+ // they are not considered specially for now. If the PTF is
non-deterministic and any of
+ // those columns are required, we cannot satisfy the requirement.
Review Comment:
Suggestion: the code is self-explanatory, we can simplify the comment
```suggestion
// Concern 1: PTF function itself is non-deterministic and downstream
nodes
// require determinism. PTFs can have pass-through input columns, but they
// are not considered specially for now.
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java:
##########
@@ -958,6 +966,71 @@ private StreamPhysicalRel visitJoinChild(
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.
+ // requireDeterminism is the set of PTF output columns downstream
requires to be
+ // deterministic (for correct retract matching). All PTF output
columns are assumed to
+ // come from PTF's computation — there are pass-through input columns
potentially, but
+ // they are not considered specially for now. If the PTF is
non-deterministic and any of
+ // those columns are required, we cannot satisfy the requirement.
+ if (!requireDeterminism.isEmpty()) {
+ final Optional<String> ndCall =
FlinkRexUtil.getNonDeterministicCallName(call);
+ if (ndCall.isPresent()) {
+ throwNonDeterministicColumnsError(
+ requireDeterminism.toList(), ptf.getRowType(), ptf,
null, ndCall);
+ }
+ }
+
+ if (inputInsertOnly(ptf)) {
+ // Insert-only input: the PTF manages retract correctness
internally via its own
+ // state, so no requirement is pushed to inputs.
Review Comment:
The PTF doesn't even see retracts so it doesn't have any retract correctness
to manage internally
```suggestion
// No retracts arrive at the PTF input, so input-column
determinism does
// not affect retract correctness
```
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java:
##########
@@ -958,6 +966,71 @@ private StreamPhysicalRel visitJoinChild(
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.
+ // requireDeterminism is the set of PTF output columns downstream
requires to be
+ // deterministic (for correct retract matching). All PTF output
columns are assumed to
+ // come from PTF's computation — there are pass-through input columns
potentially, but
+ // they are not considered specially for now. If the PTF is
non-deterministic and any of
+ // those columns are required, we cannot satisfy the requirement.
+ if (!requireDeterminism.isEmpty()) {
+ final Optional<String> ndCall =
FlinkRexUtil.getNonDeterministicCallName(call);
Review Comment:
Checking if the function output is deterministic fits the current code well
since everything is there. From a correctness and completeness perspective, we
would have optimally an option so users tell exactly which columns are
deterministic and which aren't so we check that against the ImmutableBitSet
that requireDeterminism.
I don't see user's taking the time to configure that properly but I see
builtin PTFs requiring this to make sure the planner can properly check
determinism per column
##########
flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/optimize/StreamNonDeterministicUpdatePlanVisitor.java:
##########
@@ -958,6 +966,71 @@ private StreamPhysicalRel visitJoinChild(
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.
+ // requireDeterminism is the set of PTF output columns downstream
requires to be
+ // deterministic (for correct retract matching). All PTF output
columns are assumed to
+ // come from PTF's computation — there are pass-through input columns
potentially, but
+ // they are not considered specially for now. If the PTF is
non-deterministic and any of
+ // those columns are required, we cannot satisfy the requirement.
+ if (!requireDeterminism.isEmpty()) {
+ final Optional<String> ndCall =
FlinkRexUtil.getNonDeterministicCallName(call);
+ if (ndCall.isPresent()) {
+ throwNonDeterministicColumnsError(
+ requireDeterminism.toList(), ptf.getRowType(), ptf,
null, ndCall);
+ }
+ }
+
+ if (inputInsertOnly(ptf)) {
+ // Insert-only input: the PTF manages retract correctness
internally via its own
+ // state, so no requirement is pushed to inputs.
+ 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.
+ //
+ // Instead, the check is derived per-argument from the argument's
input traits:
+ // - No REQUIRE_UPDATE_BEFORE: Retracts are routed by partition key,
so only partition
+ // key columns must be deterministic.
+ // - Has REQUIRE_UPDATE_BEFORE: the PTF explicitly requests the old
row as UPDATE_BEFORE
+ // for correct state management; that row must exactly match the
row previously
+ // processed, 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)) {
+ // UPDATE_BEFORE must exactly match the previously processed
row — all columns
+ // must be deterministic.
+ inputReq =
ImmutableBitSet.range(input.getRowType().getFieldCount());
+ } else {
+ // Retracts are routed by partition key only — only partition
key columns must
+ // be deterministic.
Review Comment:
It thing we are worried about in this case aren't the retractions - but
rather a I, U and D events should land in the same node
```suggestion
// Inserts, update and deletions are routed by partition key
only — thus the partition key columns must be deterministic.
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
