Copilot commented on code in PR #28952:
URL: https://github.com/apache/flink/pull/28952#discussion_r3754692634
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/EmitAwareCollector.java:
##########
@@ -19,17 +19,29 @@
package org.apache.flink.table.runtime.operators.join.interval;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;
/**
* Collector to wrap a [[org.apache.flink.table.dataformat.RowData]] and to
track whether a row has
* been emitted by the inner collector.
Review Comment:
The JavaDoc uses Scala-style `[[...]]` and references
`org.apache.flink.table.dataformat.RowData`, which doesn't exist. This is not
valid JavaDoc and makes the link misleading; use a standard `{@link RowData}`
reference instead.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java:
##########
@@ -178,10 +225,28 @@ public void processElement1(RowData leftRow, Context ctx,
Collector<RowData> out
if (rightTime >= rightQualifiedLowerBound
&& rightTime <= rightQualifiedUpperBound) {
List<Tuple2<RowData, Boolean>> rightRows =
rightEntry.getValue();
+ List<Boolean> rightFired =
+ earlyFireEnabled
+ ? firedBits(rightFiredState, rightTime,
rightRows)
+ : null;
Review Comment:
`rightFired` is computed for every matching right-side bucket even when the
join type is not right-outer (e.g., LEFT outer join). This adds an extra
MapState read/allocation on the hot path but is only needed when retractions
for right-side early-fire pads are possible.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/EmitAwareCollector.java:
##########
@@ -19,17 +19,29 @@
package org.apache.flink.table.runtime.operators.join.interval;
import org.apache.flink.table.data.RowData;
+import org.apache.flink.types.RowKind;
import org.apache.flink.util.Collector;
/**
* Collector to wrap a [[org.apache.flink.table.dataformat.RowData]] and to
track whether a row has
* been emitted by the inner collector.
+ *
+ * <p>The collector can be armed with a correction before a single matched row
is collected. When
+ * armed, the next collected row is treated as the corrected result of a
previously emitted
+ * speculative outer-join pad: the pending pad is emitted first stamped {@link
+ * RowKind#UPDATE_BEFORE}, then the matched row is stamped {@link
RowKind#UPDATE_AFTER}. This turns
+ * the join function's single {@code INSERT} emit into the {@code -U}/{@code
+U} pair without the
+ * join function knowing about changelogs. When not armed, collected rows are
forwarded with their
+ * existing {@link RowKind}.
Review Comment:
The JavaDoc says that when the collector is not armed it forwards the record
with its existing RowKind, but the implementation forcibly sets the kind to
INSERT to avoid leaking UPDATE_AFTER. The JavaDoc should reflect the actual
behavior.
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/interval/TimeIntervalJoin.java:
##########
@@ -261,10 +341,26 @@ public void processElement2(RowData rightRow, Context
ctx, Collector<RowData> ou
Long leftTime = leftEntry.getKey();
if (leftTime >= leftQualifiedLowerBound && leftTime <=
leftQualifiedUpperBound) {
List<Tuple2<RowData, Boolean>> leftRows =
leftEntry.getValue();
+ List<Boolean> leftFired =
+ earlyFireEnabled ? firedBits(leftFiredState,
leftTime, leftRows) : null;
Review Comment:
`leftFired` is loaded for every matching left-side bucket even when the join
type is not left-outer (e.g., RIGHT outer join). This state access is only
needed for retracting left-side early-fire pads, so it should be gated by
`joinType.isLeftOuter()` to reduce overhead.
--
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]
