This is an automated email from the ASF dual-hosted git repository.
gustavodemorais pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new ccdab3c44f7 [FLINK-41068][table] Don't store build rows whose join key
can never match in LATERAL SNAPSHOT joins
ccdab3c44f7 is described below
commit ccdab3c44f7a9e8d32d59f4c7a594856cc1c80f3
Author: Rion Williams <[email protected]>
AuthorDate: Mon Oct 26 08:16:44 2026 -0500
[FLINK-41068][table] Don't store build rows whose join key can never match
in LATERAL SNAPSHOT joins
Fixed in #29565
---
.../content.zh/docs/sql/reference/queries/joins.md | 2 +-
docs/content/docs/sql/reference/queries/joins.md | 2 +-
.../join/snapshot/LateralSnapshotJoinOperator.java | 17 ++++++--
.../snapshot/LateralSnapshotJoinOperatorTest.java | 50 +++++++++++++++++++++-
4 files changed, 64 insertions(+), 7 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/queries/joins.md
b/docs/content.zh/docs/sql/reference/queries/joins.md
index 2a0cbaa52e1..4221f5b8cc4 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 761fa0cb16e..923c7e4a5b8 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 d9c3acf43bb..72564e067d3 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.streamrecord.StreamRecord;
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 @@ import java.util.concurrent.ScheduledFuture;
* 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)
@@ -577,10 +579,15 @@ public class LateralSnapshotJoinOperator extends
AbstractStreamOperator<RowData>
@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) {
@@ -947,7 +954,11 @@ public class LateralSnapshotJoinOperator extends
AbstractStreamOperator<RowData>
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 4f22e3f8572..14aef4beba9 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 java.util.stream.Stream;
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 @@ class LateralSnapshotJoinOperatorTest {
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 @@ class LateralSnapshotJoinOperatorTest {
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 @@ class LateralSnapshotJoinOperatorTest {
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
