weiqingy commented on code in PR #28952:
URL: https://github.com/apache/flink/pull/28952#discussion_r3755153897
##########
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:
Fixed in ad2fdc1ea6d. Now `{@link RowData}`. The old link also pointed at
`org.apache.flink.table.dataformat.RowData`, which does not exist.
##########
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:
You are right, the doc was wrong and the code is deliberate. Reworded in
ad2fdc1ea6d to say the row is stamped `INSERT`, and why: the join function
reuses one row instance whose kind may have been left at `UPDATE_AFTER` by an
earlier correction. `testRowTimeEarlyFireRowKindIsolation` covers it.
##########
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:
Gated in 7912a13e609. It is worse than an extra read: on a LEFT join nothing
ever writes `rightFiredState`, so the list was always all-false.
Used `isRightOuter()` rather than an equality test, since that returns true
for FULL as well. Adding the gate showed the suite could not tell the two
apart, so I extended `testRowTimeFullOuterEarlyFireOneMatches` to drive the
right-side retraction too. With the gate written as `== FlinkJoinType.RIGHT`
that test now fails; before the change the whole suite stayed green.
##########
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:
Same fix in 7912a13e609, gated on `isLeftOuter()`.
--
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]
