[FLINK-40681][table] Reproducer: streaming outer join drops the null-padded row after an inner-side update followed by delete - #29204
Draft
MartijnVisser wants to merge 4 commits into
Conversation
…d with bare UPDATE_AFTER
When the planner does not require UPDATE_BEFORE on a join input (its upsert key contains the join
key and the sink's primary key is satisfied by the query's upsert key), the inner side of an outer
join receives a bare UPDATE_AFTER for a changed row. StreamingJoinOperator stores the replacement in
the JoinKeyContainsUniqueKey ValueState but still increments numOfAssociations of every matching
outer row, so a later DELETE of that inner row finds the counter at 2 instead of 1 and never
re-emits the null-padded outer row. The same arithmetic lives in JoinHelper (async state) and
StreamingSemiAntiJoinOperator; MiniBatchStreamingJoinOperator additionally folds an UPDATE_AFTER
followed by a DELETE within one bundle into nothing while the previous row version stays in state.
The new tests assert the null-padded row (or the anti-join row) after the DELETE and fail on master:
testLeftOuterJoinInnerSideUpdateAfterThenDeleteRestoresNullPadding (sync and async state)
actual [-D(Ord#1,LineOrd#1,...,LineOrd#1,SHIP)]
expected [-D(Ord#1,LineOrd#1,...,LineOrd#1,SHIP), +I(Ord#1,LineOrd#1,...,null,null)]
testRightOuterJoin..., testFullOuterJoinRightSide..., testFullOuterJoinLeftSide...: same shape
testLeftJoinJoinKeyContainsUniqueKeyUpdateAfterThenDeleteWithinBundle (mini-batch size 2)
actual []
expected [-D(...,Ord#X,LineOrd#1,AIR), +I(...,null,null,null)]
testLeftAntiJoinUpdateAfterThenDeleteRestoresLeftRow
actual []
expected [+I(Ord#1,LineOrd#1,...)]
Three existing tests assert the drifted output and are left unchanged for now:
StreamingJoinOperatorTest#testLeftOuterJoinWithStateRetentionDisabled,
StreamingSemiAntiJoinOperatorTest#testLeftSemiJoinWithDifferentStateRetentionTime and
#testLeftSemiJoinWithStateRetentionDisabled.
Generated-by: Claude Code (Claude Fable 5.1)
…ert tables into an upsert sink
When the sink's primary key is satisfied by the query's upsert key, the planner lets an outer join
run without UPDATE_BEFORE on both inputs (Join ... changelogMode=[I,UA,D] directly over the upsert
sources, no ChangelogNormalize, no SinkUpsertMaterializer). StreamingJoinOperator then counts a
replacing UPDATE_AFTER of the inner side as an additional association of the outer row and never
re-emits the null-padded row when that inner row is deleted, so the outer row disappears from the
sink.
OuterJoinUpsertInputITCase drives the replacing UPDATE_AFTER through a gated source so that it
always arrives after the joined row reached the sink, and fails on master with HEAP and ROCKSDB:
testLeftJoinUpsertInputsIntoUpsertSinkWithSamePk
left_t: +I(1,a); right_t: +I(1,x), +U(1,y), -D(1,y)
actual []
expected ["+I[1, a, null]"]
testNestedLeftJoinsUpsertInputsIntoUpsertSinkOnLeftPk (the control query of FLINK-23740 with
B LEFT OUTER JOIN C, sink keyed on A's primary key)
actual []
expected ["+I[1, 1, 1, 1, 1, a, b, null, d]"]
testLeftJoinUpsertInputsIntoUpsertSinkWithLargerPk is the control: with sink primary key (k, v) the
planner keeps UPDATE_BEFORE and the same input yields ["+I[1, a, null]"].
The two JoinTest plans pin the changelog modes the ITCases rely on.
Generated-by: Claude Code (Claude Fable 5.1)
…elog in the reproducer
A repeated INSERT for the same key reaches the join directly when no ChangelogNormalize is
required and inflates numOfAssociations exactly like a bare UPDATE_AFTER, so the defect is any
accumulate that replaces a row on a unique-key side. The new harness case fails on master with
testLeftOuterJoinInnerSideDuplicateInsertThenDeleteRestoresNullPadding (sync and async state)
actual [-D(Ord#1,LineOrd#1,...,LineOrd#1,AIR)]
expected [-D(Ord#1,LineOrd#1,...,LineOrd#1,AIR), +I(Ord#1,LineOrd#1,...,null,null)]
OuterJoinUpsertInputITCase now gates both leading rows of the inner side on the sink contents,
which makes the raw changelog deterministic, and asserts that changelog before the materialized
view so that the failure localizes the missing row to the join:
testLeftJoinUpsertInputsIntoUpsertSinkWithSamePk (HEAP and ROCKSDB)
actual [+I[1, a, null], -D[1, a, null], +I[1, a, x], +I[1, a, y], -D[1, a, y]]
expected the same followed by +I[1, a, null]
The control with sink primary key (k, v) asserts the nine-row retract changelog including
-U[1, a, x] and stays green.
Generated-by: Claude Code (Claude Fable 5.1)
…uter join query of FLINK-23740 The query of FLINK-23740 (A LEFT JOIN (B FULL OUTER JOIN C) LEFT JOIN D over four upsert tables) is run over seeded random upsert changelogs into an upsert sink keyed on A's primary key and compared with the same query in batch mode over the final snapshot of every table, 20 seeds per state backend, parallelism 1. The reported FULL OUTER JOIN variant matches the batch result on every seed: its upsert key is the concatenation of all keys and differs from the sink key, so the planner keeps UPDATE_BEFORE. The LEFT OUTER JOIN control variant runs without UPDATE_BEFORE and diverges on 20 of 20 seeds on HEAP and ROCKSDB, either as a missing row or as a DELETE reaching the sink for a row it never held. Generated-by: Claude Code (Claude Fable 5.1)
Collaborator
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What is the purpose of the change
Failing tests only, no fix: this draft PR shows the defect described in FLINK-40681. When the planner does not require UPDATE_BEFORE on the inputs of an outer join, an accumulate message that replaces a row on a unique-key inner side is counted as an additional association of the outer row. The later DELETE of that inner row then never re-emits the null-padded outer row, and the row disappears from an upsert sink. The same arithmetic is in
JoinHelper(async state) andStreamingSemiAntiJoinOperator; the mini-batch join additionally folds an UPDATE_AFTER followed by a DELETE within one bundle into nothing.The nested query of FLINK-23740 is included as a differential test against a batch oracle: the reported FULL OUTER JOIN variant is correct on master, its LEFT OUTER JOIN control variant is not.
Brief change log
StreamingJoinOperatorTest: LEFT, RIGHT and FULL outer join cases where the inner side sends a replacing UPDATE_AFTER (or a duplicate INSERT) and then a DELETE; run with sync and async stateStreamingMiniBatchJoinOperatorTest: the same across bundles and within one bundleStreamingSemiAntiJoinOperatorTest: the anti join variantOuterJoinUpsertInputITCase:A LEFT JOIN Dand the nested LEFT JOIN query of FLINK-23740 into an upsert sink, asserting the raw changelog; a control with a larger sink key that keeps UPDATE_BEFORE stays greenJoinTest: plans pinning that the joins run in[I,UA,D]for these queriesNestedOuterJoinChangelogStressITCase: seeded differential test of the FLINK-23740 query in both variants against a batch oracleVerifying this change
This change added tests. All of them except the control and the FULL OUTER JOIN variant fail on master by design; CI is expected to be red until a fix lands.
Does this pull request potentially affect one of the following parts:
@Public(Evolving): noDocumentation
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Claude Fable 5.1)