diff --git a/docs/content.zh/docs/sql/reference/queries/joins.md b/docs/content.zh/docs/sql/reference/queries/joins.md index 2a0cbaa52e10d..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. | +| `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 761fa0cb16ede..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. | +| `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/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..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 @@ -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,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); @@ -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 @@ -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 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.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