From e09288c3230052e7dd702ba0c4e3994c2a7dfbc1 Mon Sep 17 00:00:00 2001 From: Rion Williams Date: Thu, 22 Oct 2026 08:16:36 -0500 Subject: [PATCH 1/2] [FLINK-41068][table] Don't store build rows whose join key can never match in LATERAL SNAPSHOT joins A build row whose join key has a NULL in a field that is not null-safe can never match, but LateralSnapshotJoinOperator buffered it and added it to the build table, where it stayed until the build side retracted it or state TTL removed it. All such rows share one key, so they ended up in the state of a single subtask, and every probe with a NULL key scanned them. The operator no longer stores these rows. The build side is never preserved, so the output doesn't change. Retractions are still applied, so a row that an earlier version stored is removed, but a retraction of such a row that finds nothing is not counted in numUnmatchedBuildRetractions. Generated-by: Claude Code (Claude Opus 5.5) --- .../docs/sql/reference/queries/joins.md | 2 +- .../docs/sql/reference/queries/joins.md | 2 +- .../snapshot/LateralSnapshotJoinOperator.java | 17 ++++++-- .../LateralSnapshotJoinOperatorTest.java | 39 +++++++++++++++++++ 4 files changed, 55 insertions(+), 5 deletions(-) diff --git a/docs/content.zh/docs/sql/reference/queries/joins.md b/docs/content.zh/docs/sql/reference/queries/joins.md index 2a0cbaa52e10d..5013d968e45fd 100644 --- a/docs/content.zh/docs/sql/reference/queries/joins.md +++ b/docs/content.zh/docs/sql/reference/queries/joins.md @@ -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 whose join key can never match, because a key field that is not compared null-safely is `NULL`, are not counted, since such rows are not 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. diff --git a/docs/content/docs/sql/reference/queries/joins.md b/docs/content/docs/sql/reference/queries/joins.md index 761fa0cb16ede..75bb9070a5515 100644 --- a/docs/content/docs/sql/reference/queries/joins.md +++ b/docs/content/docs/sql/reference/queries/joins.md @@ -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 whose join key can never match, because a key field that is not compared null-safely is `NULL`, are not counted, since such rows are not 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. diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperator.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperator.java index d9c3acf43bbe7..72564e067d30d 100644 --- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperator.java +++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperator.java @@ -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; @@ -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. *
  • 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) @@ -577,10 +579,15 @@ private void bufferProbe(RowData probe) throws Exception { @Override public void processElement2(StreamRecord 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 atRowtime = buildChangeBuffer.get(rowtime); if (atRowtime == null) { @@ -947,7 +954,11 @@ private void applyBuildChanges(List 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) { diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperatorTest.java index 4f22e3f85727c..f9548d4efbb8a 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperatorTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperatorTest.java @@ -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; @@ -1250,6 +1251,8 @@ void joinRespectsNullKeysFilter(boolean leftOuter, boolean filterNullKey) throws newHarness(op)) { h.open(); addBuildChange(h, insertRecord(null, "v_null", 1L)); + // A build row that can never match is not stored + assertThat(bufferedChangesForKey(h, op, null)).hasSize(filterNullKey ? 0 : 1); addProbeRecord(h, 1L, null, "p_null"); addProbeWm(h, 100L); addBuildWm(h, 100L); @@ -1280,6 +1283,42 @@ void joinRespectsNullKeysFilter(boolean leftOuter, boolean filterNullKey) throws } } + @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 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 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.DELETE, 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 From da6ebc5408edd3694c218d88862238a27324051d Mon Sep 17 00:00:00 2001 From: Rion Williams Date: Thu, 22 Oct 2026 13:05:05 -0500 Subject: [PATCH 2/2] [FLINK-41068][table] Cover update rows in the NULL-key build tests and simplify the metric docs Generated-by: Claude Code (Claude Opus 5.5) --- .../docs/sql/reference/queries/joins.md | 2 +- docs/content/docs/sql/reference/queries/joins.md | 2 +- .../snapshot/LateralSnapshotJoinOperatorTest.java | 15 +++++++++++---- 3 files changed, 13 insertions(+), 6 deletions(-) diff --git a/docs/content.zh/docs/sql/reference/queries/joins.md b/docs/content.zh/docs/sql/reference/queries/joins.md index 5013d968e45fd..4221f5b8cc4dc 100644 --- a/docs/content.zh/docs/sql/reference/queries/joins.md +++ b/docs/content.zh/docs/sql/reference/queries/joins.md @@ -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. Retractions of rows whose join key can never match, because a key field that is not compared null-safely is `NULL`, are not counted, since such rows are not stored. | +| `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. diff --git a/docs/content/docs/sql/reference/queries/joins.md b/docs/content/docs/sql/reference/queries/joins.md index 75bb9070a5515..923c7e4a5b8c0 100644 --- a/docs/content/docs/sql/reference/queries/joins.md +++ b/docs/content/docs/sql/reference/queries/joins.md @@ -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. Retractions of rows whose join key can never match, because a key field that is not compared null-safely is `NULL`, are not counted, since such rows are not stored. | +| `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. diff --git a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperatorTest.java b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperatorTest.java index f9548d4efbb8a..14aef4beba93d 100644 --- a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperatorTest.java +++ b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/snapshot/LateralSnapshotJoinOperatorTest.java @@ -1251,8 +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 : 1); + assertThat(bufferedChangesForKey(h, op, null)).hasSize(filterNullKey ? 0 : 2); addProbeRecord(h, 1L, null, "p_null"); addProbeWm(h, 100L); addBuildWm(h, 100L); @@ -1266,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 @@ -1278,7 +1282,10 @@ 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)); } } } @@ -1309,7 +1316,7 @@ void retractsNullKeyBuildRowStoredByEarlierVersion(StateBackend backend) throws h.initializeState(state); h.open(); // Binary rows, as deserialized input is, so that they equal the restored rows on heap - addBuildChange(h, binaryRecord(RowKind.DELETE, null, "v_null", 1L)); + 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");