Skip to content

[FLINK-41068][table] Don't store build rows whose join key can never match in LATERAL SNAPSHOT joins #29565

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rionmonster wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[FLINK-41068][table] Don't store build rows whose join key can never match in LATERAL SNAPSHOT joins #29565

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e09288c
[FLINK-41068][table] Don't store build rows whose join key can never …
rionmonster Oct 22, 2026
da6ebc5
[FLINK-41068][table] Cover update rows in the NULL-key build tests an…
rionmonster Oct 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/content.zh/docs/sql/reference/queries/joins.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -454,7 +454,7 @@ The operator exposes metrics that help you understand which phase the join is in
| `maxJoinFanOut` | Gauge | The largest number of joined rows emitted for a single probe-side row. |
| `avgJoinFanOut` | Gauge | The average number of joined rows emitted per probe-side row. |
| `numUnmatchedProbeRecords` | Counter | The number of probe-side rows that found no build-side match. For an `INNER JOIN` these produce no output; for a `LEFT JOIN` they are emitted null-padded. |
| `numUnmatchedBuildRetractions` | Counter | The number of build-side retractions (`-U`/`-D`) for a row that was not present in state. |
| `numUnmatchedBuildRetractions` | Counter | The number of build-side retractions (`-U`/`-D`) for a row that was not present in state. Retractions of rows with a `NULL` in a join key compared with `=` are not counted, since such rows are never stored. |

To diagnose a join that emits no output, check `currentPhase` first. If it is `0` while `numProbeSideRecordsBuffered` keeps growing, the join is stuck in the load phase.
Compare `currentBuildSideWatermark` against your `load_completed_time`: if the build-side watermark has stopped advancing below it, the build side has become static or idle and the event-time gate will never be crossed.
Expand Down
2 changes: 1 addition & 1 deletion docs/content/docs/sql/reference/queries/joins.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -464,7 +464,7 @@ They are reported under the operator's [metric group]({{< ref "docs/ops/metrics"
| `maxJoinFanOut` | Gauge | The largest number of joined rows emitted for a single probe-side row. |
| `avgJoinFanOut` | Gauge | The average number of joined rows emitted per probe-side row. |
| `numUnmatchedProbeRecords` | Counter | The number of probe-side rows that found no build-side match. For an `INNER JOIN` these produce no output; for a `LEFT JOIN` they are emitted null-padded. |
| `numUnmatchedBuildRetractions` | Counter | The number of build-side retractions (`-U`/`-D`) for a row that was not present in state. |
| `numUnmatchedBuildRetractions` | Counter | The number of build-side retractions (`-U`/`-D`) for a row that was not present in state. Retractions of rows with a `NULL` in a join key compared with `=` are not counted, since such rows are never stored. |

To diagnose a join that emits no output, check `currentPhase` first. If it is `0` while `numProbeSideRecordsBuffered` keeps growing, the join is stuck in the load phase.
Compare `currentBuildSideWatermark` against your `load_completed_time`: if the build-side watermark has stopped advancing below it, the build side has become static or idle and the event-time gate will never be crossed.
Expand Down
17 changes: 14 additions & 3 deletions ...a/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.util.RowDataUtil;
import org.apache.flink.table.data.utils.JoinedRowData;
import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
import org.apache.flink.table.runtime.generated.JoinCondition;
Expand Down Expand Up @@ -86,7 +87,8 @@
* once the build watermark reaches it; changes sharing a row-time are applied in arrival
* order, which mirrors the upstream emit order (a retraction directly precedes its
* accumulation). Gating application on the row-time preserves atomic update visibility across
* {@code -U}/{@code +U} pairs in JOIN phase.
* {@code -U}/{@code +U} pairs in JOIN phase. Accumulations whose join key can never match are
* not stored.
* <li>Probe-side (input1 / left) records are handled differently per phase. During LOAD each is
* buffered in {@code probeBuffer} under a synthetic, increasing timestamp with one event-time
* timer registered per record. At the flip these timers fire (one record each, interruptibly)
Expand Down Expand Up @@ -577,10 +579,15 @@ private void bufferProbe(RowData probe) throws Exception {
@Override
public void processElement2(StreamRecord<RowData> element) throws Exception {
armIdleFlipTimerIfNotArmed();
RowData build = element.getValue();
if (RowDataUtil.isAccumulateMsg(build) && joinCondition.currentKeyCannotMatch()) {
// A build row that can never match is not stored. Its retraction is still applied, to
// remove the row if an earlier version stored it.
return;
}
// Apply any buffered build-side changes if the build-side watermark has advanced.
Long bufferedAt = applyBufferedChangesUpToCurrentWm();
// Buffer the change under its build row-time (grouping changes that share a row-time).
RowData build = element.getValue();
long rowtime = build.getLong(buildRowtimeIndex);
List<RowData> atRowtime = buildChangeBuffer.get(rowtime);
if (atRowtime == null) {
Expand Down Expand Up @@ -947,7 +954,11 @@ private void applyBuildChanges(List<RowData> changes) throws Exception {
buildTableState.put(change, currentCnt == null ? 1L : currentCnt + 1L);
} else {
if (currentCnt == null || currentCnt <= 0L) {
numUnmatchedBuildRetractions.inc();
// Rows that can never match are not stored, so their retractions are expected
// to find nothing.
if (!joinCondition.currentKeyCannotMatch()) {
numUnmatchedBuildRetractions.inc();
}
continue;
}
if (currentCnt == 1L) {
Expand Down
50 changes: 48 additions & 2 deletions ...g/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperatorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,7 @@
import static org.apache.flink.table.runtime.operators.join.snapshot.LateralSnapshotJoinOperator.BUILD_CHANGE_BUFFER_STATE_NAME;
import static org.apache.flink.table.runtime.operators.join.snapshot.LateralSnapshotJoinOperator.BUILD_TABLE_STATE_NAME;
import static org.apache.flink.table.runtime.operators.join.snapshot.LateralSnapshotJoinOperator.PROBE_BUFFER_STATE_NAME;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.binaryRecord;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.deleteRecord;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.insertRecord;
import static org.apache.flink.table.runtime.util.StreamRecordUtils.updateAfterRecord;
Expand Down Expand Up @@ -1250,6 +1251,9 @@ void joinRespectsNullKeysFilter(boolean leftOuter, boolean filterNullKey) throws
newHarness(op)) {
h.open();
addBuildChange(h, insertRecord(null, "v_null", 1L));
addBuildChange(h, updateAfterRecord(null, "v_null_u", 1L));
// A build row that can never match is not stored
assertThat(bufferedChangesForKey(h, op, null)).hasSize(filterNullKey ? 0 : 2);
addProbeRecord(h, 1L, null, "p_null");
addProbeWm(h, 100L);
addBuildWm(h, 100L);
Expand All @@ -1263,7 +1267,10 @@ void joinRespectsNullKeysFilter(boolean leftOuter, boolean filterNullKey) throws
assertThat(h.getOutput()).isEmpty();
}
} else {
JOINED_ASSERTOR.shouldEmitAll(h, row(1L, null, "p_null", null, "v_null", 1L));
JOINED_ASSERTOR.shouldEmitAll(
h,
row(1L, null, "p_null", null, "v_null", 1L),
row(1L, null, "p_null", null, "v_null_u", 1L));
}

// test joining in JOIN phase
Expand All @@ -1275,11 +1282,50 @@ void joinRespectsNullKeysFilter(boolean leftOuter, boolean filterNullKey) throws
assertThat(h.getOutput()).isEmpty();
}
} else {
JOINED_ASSERTOR.shouldEmitAll(h, row(2L, null, "p_null", null, "v_null", 1L));
JOINED_ASSERTOR.shouldEmitAll(
h,
row(2L, null, "p_null", null, "v_null", 1L),
row(2L, null, "p_null", null, "v_null_u", 1L));
}
}
}

@ParameterizedTest(name = "backend={0}")
@MethodSource("stateBackends")
void retractsNullKeyBuildRowStoredByEarlierVersion(StateBackend backend) throws Exception {
// Store a build row with a null key, as an earlier version did, by comparing it null-safely
LateralSnapshotJoinOperator op1 =
newOperator(false, newTrueCondition(), new boolean[] {false}, 100L, null, null);
OperatorSubtaskState state;
try (KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData, RowData> h =
newHarness(op1)) {
h.setStateBackend(backend);
h.open();
addBuildChange(h, insertRecord(null, "v_null", 1L));
addBuildWm(h, 100L);
addProbeRecord(h, 1L, null, "p_null");
assertThat(buildTableForKey(h, op1, null))
.containsExactlyInAnyOrderEntriesOf(Map.of("v_null", 1L));
state = h.snapshot(0L, 0L);
}

LateralSnapshotJoinOperator op2 = newOperator(false, 100L, null, null);
try (KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData, RowData> h =
newHarness(op2)) {
h.setStateBackend(backend);
h.initializeState(state);
h.open();
// Binary rows, as deserialized input is, so that they equal the restored rows on heap
addBuildChange(h, binaryRecord(RowKind.UPDATE_BEFORE, null, "v_null", 1L));
addBuildChange(h, binaryRecord(RowKind.DELETE, null, "never_stored", 1L));
addProbeRecord(h, 2L, null, "p_null");

assertThat(buildTableForKey(h, op2, null)).isEmpty();
// Retractions of rows that can never match are not counted as unmatched
assertThat(op2.getNumUnmatchedBuildRetractions().getCount()).isZero();
}
}

// ----------------------------------------------------------------- Watermark status

@Test
Expand Down