xishuaidelin commented on code in PR #28671:
URL: https://github.com/apache/flink/pull/28671#discussion_r3568880742
##########
flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/stream/bundle/InputSideHasUniqueKeyBundle.java:
##########
@@ -33,8 +34,23 @@
/** For the case that input has uniqueKey which is not contained by joinKey. */
public class InputSideHasUniqueKeyBundle extends BufferBundle<Map<RowData,
List<RowData>>> {
+ /**
+ * Tracks (joinKey → set of uniqueKeys) whose first accumulate record (+I)
was folded away with
+ * a retract in the same batch. A subsequent +U for such a unique key must
be treated as +I.
+ */
+ private final Map<RowData, Set<RowData>> insertedAndCleared = new
HashMap<>();
Review Comment:
Cross-batch boundary: the correction is lost when an update pair is split
across batches
insertedAndCleared is cleared at the end of every batch (clear()), so the
correction only works when the folded +I and the following +U land in the same
batch. Since
CountCoBundleTrigger flushes purely on element count, it can split an
update sequence across a batch boundary:
Input: +I(k), -U(k), +U(k)
Batch: [ +I(k), -U(k) ] | [ +U(k) ]
- Batch N — +I(k) and -U(k) fold to empty and mark k in
insertedAndCleared, but the mark is wiped by clear() at batch end. Nothing is
emitted, and no +I for k ever
reaches downstream.
- Batch N+1 — +U(k) finds no mark, so it is emitted unchanged as +U for a
key downstream has never seen.
--
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]
