This is an automated email from the ASF dual-hosted git repository.
fhueske 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 087e8772291 [FLINK-40902][table] Support LATERAL SNAPSHOT join without
an equi-key (#29421)
087e8772291 is described below
commit 087e8772291b65b7c22f60a3846a5e181cc9bea7
Author: Fabian Hueske <[email protected]>
AuthorDate: Tue Oct 13 10:29:21 2026 +0200
[FLINK-40902][table] Support LATERAL SNAPSHOT join without an equi-key
(#29421)
* [FLINK-40902][table] Support LATERAL SNAPSHOT join without an equi-key
LATERAL SNAPSHOT join required at least one equality predicate and rejected
any
join without one. This also rejected valid equi-joins whose equi-key was
folded
to a constant by a local filter on the join key (e.g. ON o.k = c.k WHERE
o.k =
'x'): Calcite pins both sides to the constant and simplifies the join
condition,
leaving no equi-key for the rule to find.
Lift the equi-key requirement so LATERAL SNAPSHOT behaves like other Flink
joins
that tolerate a missing equi-key. A join with an equi-key is unchanged. A
join
without one is executed single-threaded in streaming (SINGLETON
distribution; the
operator already supports an empty key) and as a broadcast nested-loop join
in
batch, which keeps the probe side parallel.
This resolves FLINK-40902: the filtered query above now plans and executes
correctly instead of being rejected.
Generated-by: Claude Code (Claude Opus 4.8)
---
.../content.zh/docs/sql/reference/queries/joins.md | 4 +-
docs/content/docs/sql/reference/queries/joins.md | 4 +-
.../LogicalJoinToLateralSnapshotJoinRule.java | 8 -
.../BatchPhysicalLateralSnapshotJoinRule.java | 54 +++++--
.../StreamPhysicalLateralSnapshotJoinRule.java | 3 +-
.../batch/sql/join/LateralSnapshotJoinTest.java | 28 +++-
.../LateralSnapshotJoinBatchSemanticTests.java | 3 +-
.../batch/LateralSnapshotJoinTestPrograms.java | 32 +++-
.../LateralSnapshotJoinSemanticTestPrograms.java | 82 +++++++++-
.../stream/LateralSnapshotJoinSemanticTests.java | 3 +
.../stream/sql/join/LateralSnapshotJoinTest.java | 54 +++++--
.../batch/sql/join/LateralSnapshotJoinTest.xml | 46 ++++++
.../stream/sql/join/LateralSnapshotJoinTest.xml | 88 ++++++++++
.../join/snapshot/LateralSnapshotJoinOperator.java | 9 ++
.../snapshot/LateralSnapshotJoinOperatorTest.java | 177 +++++++++++++++++++++
15 files changed, 543 insertions(+), 52 deletions(-)
diff --git a/docs/content.zh/docs/sql/reference/queries/joins.md
b/docs/content.zh/docs/sql/reference/queries/joins.md
index 1d83fd7e270..012d8b14327 100644
--- a/docs/content.zh/docs/sql/reference/queries/joins.md
+++ b/docs/content.zh/docs/sql/reference/queries/joins.md
@@ -380,7 +380,7 @@ If a `LATERAL SNAPSHOT` join produces no output, first
check whether it is stuck
**Syntax**
The build side is wrapped in the `SNAPSHOT` table function, which is called
with `LATERAL`. The outer (probe-side) table must be an append-only table.
-Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. The join requires at
least one conjunctive equality predicate; additional non-equi predicates are
allowed in the `ON` clause.
+Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. An equality predicate
is recommended but not required: the `ON` clause may contain equality and
non-equi predicates in any combination. In streaming mode, a join with at least
one equality predicate is partitioned on the join key and runs in parallel,
whereas a join without one runs single-threaded; prefer an equality predicate
for better performance. In batch mode, a join without an equality predicate
uses a broadcast nested-loop [...]
```sql
SELECT [column_list]
@@ -412,7 +412,7 @@ The result is append-only and preserves the probe-side time
attributes. A build-
Because rows are joined against the build-side state that is current at
processing time, the result is **not deterministic**. A given probe-side row
may be joined with different build-side versions across different runs,
depending on the relative timing of the two inputs. Probe and build-side inputs
can be configured with watermark alignment to keep the two inputs roughly
aligned on event time, so that a probe-side row tends to be joined with
build-side changes of a similar event time. T [...]
-The build-side state grows with the number of distinct build-side keys, and
during the load phase the buffered probe-side rows add to the state footprint
until the operator transitions to the join phase. Use `state_ttl` to bound the
build-side state for keys that are no longer updated or joined. You can reduce
the amount of data that is buffered and processed during the load phase by
configuring scan start offsets on the build and probe-side inputs, for example
with a `scan.startup.*` [d [...]
+The build-side state grows with the number of distinct build-side keys, and
during the load phase the buffered probe-side rows add to the state footprint
until the operator transitions to the join phase. Use `state_ttl` to bound the
build-side state for keys that are no longer updated or joined. Without an
equality predicate the join has no join key, so all build-side rows share a
single state entry; `state_ttl` can then only evict the whole build side at
once, once it has been idle (no [...]
**Monitoring**
diff --git a/docs/content/docs/sql/reference/queries/joins.md
b/docs/content/docs/sql/reference/queries/joins.md
index ba31ded4566..2a5a4b07182 100644
--- a/docs/content/docs/sql/reference/queries/joins.md
+++ b/docs/content/docs/sql/reference/queries/joins.md
@@ -389,7 +389,7 @@ If a `LATERAL SNAPSHOT` join produces no output, first
check whether it is stuck
**Syntax**
The build side is wrapped in the `SNAPSHOT` table function, which is called
with `LATERAL`. The outer (probe-side) table must be an append-only table.
-Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. The join requires at
least one conjunctive equality predicate; additional non-equi predicates are
allowed in the `ON` clause.
+Both `INNER JOIN` and `LEFT [OUTER] JOIN` are supported. An equality predicate
is recommended but not required: the `ON` clause may contain equality and
non-equi predicates in any combination. In streaming mode, a join with at least
one equality predicate is partitioned on the join key and runs in parallel,
whereas a join without one runs single-threaded; prefer an equality predicate
for better performance. In batch mode, a join without an equality predicate
uses a broadcast nested-loop [...]
```sql
SELECT [column_list]
@@ -421,7 +421,7 @@ The result is append-only and preserves the probe-side time
attributes. A build-
Because rows are joined against the build-side state that is current at
processing time, the result is **not deterministic**. A given probe-side row
may be joined with different build-side versions across different runs,
depending on the relative timing of the two inputs. Probe and build-side inputs
can be configured with watermark alignment to keep the two inputs roughly
aligned on event time, so that a probe-side row tends to be joined with
build-side changes of a similar event time. T [...]
-The build-side state grows with the number of distinct build-side keys, and
during the load phase the buffered probe-side rows add to the state footprint
until the operator transitions to the join phase. Use `state_ttl` to bound the
build-side state for keys that are no longer updated or joined. You can reduce
the amount of data that is buffered and processed during the load phase by
configuring scan start offsets on the build and probe-side inputs, for example
with a `scan.startup.*` [d [...]
+The build-side state grows with the number of distinct build-side keys, and
during the load phase the buffered probe-side rows add to the state footprint
until the operator transitions to the join phase. Use `state_ttl` to bound the
build-side state for keys that are no longer updated or joined. Without an
equality predicate the join has no join key, so all build-side rows share a
single state entry; `state_ttl` can then only evict the whole build side at
once, once it has been idle (no [...]
**Monitoring**
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java
index 6edb403f934..325e7128c27 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/logical/LogicalJoinToLateralSnapshotJoinRule.java
@@ -35,7 +35,6 @@ import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.plan.RelRule;
import org.apache.calcite.plan.hep.HepRelVertex;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.core.JoinInfo;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rel.type.RelDataTypeField;
@@ -107,13 +106,6 @@ public class LogicalJoinToLateralSnapshotJoinRule
joinType));
}
- // Require at least one equality predicate so the operator can
hash-partition both inputs.
- final JoinInfo joinInfo = join.analyzeCondition();
- if (joinInfo.leftKeys.isEmpty()) {
- throw new ValidationException(
- "LATERAL SNAPSHOT join requires at least one equality
predicate.");
- }
-
final RexCall snapshotCall = (RexCall) scan.getCall();
// Replace the SNAPSHOT TableFunctionScan with its TABLE input,
preserving any
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLateralSnapshotJoinRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLateralSnapshotJoinRule.java
index 9338ffb8feb..dc4ecf8fa6a 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLateralSnapshotJoinRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalLateralSnapshotJoinRule.java
@@ -21,6 +21,7 @@ package
org.apache.flink.table.planner.plan.rules.physical.batch;
import org.apache.flink.table.planner.plan.nodes.FlinkConventions;
import
org.apache.flink.table.planner.plan.nodes.logical.FlinkLogicalLateralSnapshotJoin;
import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalHashJoin;
+import
org.apache.flink.table.planner.plan.nodes.physical.batch.BatchPhysicalNestedLoopJoin;
import
org.apache.flink.table.planner.plan.rules.logical.LogicalJoinToLateralSnapshotJoinRule;
import
org.apache.flink.table.planner.plan.rules.physical.stream.StreamPhysicalLateralSnapshotJoinRule;
import org.apache.flink.table.planner.plan.trait.FlinkRelDistribution;
@@ -34,8 +35,9 @@ import org.apache.calcite.util.ImmutableIntList;
/**
* Converts a {@link FlinkLogicalLateralSnapshotJoin} (created by {@link
- * LogicalJoinToLateralSnapshotJoinRule}) into a regular batch {@link
BatchPhysicalHashJoin} for
- * batch execution.
+ * LogicalJoinToLateralSnapshotJoinRule}) into a regular batch join for batch
execution: a {@link
+ * BatchPhysicalHashJoin} when the join has an equi-key, or a {@link
BatchPhysicalNestedLoopJoin}
+ * when it does not.
*
* <p>In batch all input is bounded and append-only (batch rejects or in the
future materializes
* non-insert-only sources up front), so the processing-time {@code LATERAL
SNAPSHOT} join
@@ -45,8 +47,10 @@ import org.apache.calcite.util.ImmutableIntList;
* stream operator.
*
* <p>The SNAPSHOT input is the build (right) side of the LATERAL join and is
the dimension-like
- * side, expected to be smaller than the probe (left) side. The join is
therefore emitted as a
- * shuffle hash join that builds on the right input.
+ * side, expected to be smaller than the probe (left) side. With an equi-key
the join is emitted as
+ * a shuffle hash join that builds on the right input; without an equi-key
(e.g. a non-equi
+ * condition, or an equi-key that a local filter folded to a constant) it is
emitted as a broadcast
+ * nested-loop join that broadcasts the right input, keeping the probe side
parallel.
*/
public class BatchPhysicalLateralSnapshotJoinRule extends ConverterRule {
@@ -68,8 +72,40 @@ public class BatchPhysicalLateralSnapshotJoinRule extends
ConverterRule {
final RelTraitSet providedTraitSet =
rel.getTraitSet().replace(FlinkConventions.BATCH_PHYSICAL());
- // Both inputs are hash-partitioned on their join keys (shuffle hash
join).
final JoinInfo joinInfo = join.analyzeCondition();
+ if (joinInfo.leftKeys.isEmpty()) {
+ // Without an equi-key a hash join is not applicable (the join is
non-equi, or its
+ // equi-key was constant-folded away by a local filter on the join
key). Mirror how
+ // regular batch joins handle this and emit a broadcast
nested-loop join that broadcasts
+ // the (smaller, dimension-like) SNAPSHOT build side to the probe
side, which keeps the
+ // probe side parallel.
+ final RelNode newLeft =
+ RelOptRule.convert(
+ join.getLeft(),
+ join.getLeft()
+ .getTraitSet()
+
.replace(FlinkConventions.BATCH_PHYSICAL()));
+ final RelNode newRight =
+ RelOptRule.convert(
+ join.getRight(),
+ join.getRight()
+ .getTraitSet()
+ .replace(FlinkConventions.BATCH_PHYSICAL())
+
.replace(FlinkRelDistribution.BROADCAST_DISTRIBUTED()));
+ return new BatchPhysicalNestedLoopJoin(
+ join.getCluster(),
+ providedTraitSet,
+ newLeft,
+ newRight,
+ join.getCondition(),
+ join.getJoinType(),
+ // leftIsBuild = false: build the right (SNAPSHOT) side.
+ false,
+ // singleRowJoin = false: neither side is known to produce
a single row.
+ false);
+ }
+
+ // Both inputs are hash-partitioned on their join keys (shuffle hash
join).
final RelNode newLeft = convertInput(join.getLeft(),
joinInfo.leftKeys);
final RelNode newRight = convertInput(join.getRight(),
joinInfo.rightKeys);
@@ -92,17 +128,13 @@ public class BatchPhysicalLateralSnapshotJoinRule extends
ConverterRule {
/**
* Converts a join input to the batch-physical convention and requires it
to be hash-partitioned
- * on the given join {@code keys} (or a singleton distribution if there
are none).
+ * on the given join {@code keys}.
*/
private static RelNode convertInput(RelNode input, ImmutableIntList keys) {
- final FlinkRelDistribution distribution =
- keys.isEmpty()
- ? FlinkRelDistribution.SINGLETON()
- : FlinkRelDistribution.hash(keys.toIntArray(), true);
final RelTraitSet requiredTraitSet =
input.getTraitSet()
.replace(FlinkConventions.BATCH_PHYSICAL())
- .replace(distribution);
+ .replace(FlinkRelDistribution.hash(keys.toIntArray(),
true));
return RelOptRule.convert(input, requiredTraitSet);
}
}
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLateralSnapshotJoinRule.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLateralSnapshotJoinRule.java
index ec8010deed1..2d1408cbddb 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLateralSnapshotJoinRule.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/StreamPhysicalLateralSnapshotJoinRule.java
@@ -56,7 +56,8 @@ public class StreamPhysicalLateralSnapshotJoinRule extends
ConverterRule {
final RelTraitSet providedTraitSet =
rel.getTraitSet().replace(FlinkConventions.STREAM_PHYSICAL());
- // Both inputs are hash-partitioned on their join keys.
+ // Both inputs are hash-partitioned on their join keys, or co-located
with a singleton
+ // distribution when the join has no equi-key
final JoinInfo joinInfo = join.analyzeCondition();
final RelNode newLeft = convertInput(join.getLeft(),
joinInfo.leftKeys);
final RelNode newRight = convertInput(join.getRight(),
joinInfo.rightKeys);
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
index 11aefe68908..1849f672d34 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.java
@@ -152,23 +152,35 @@ public class LateralSnapshotJoinTest extends
TableTestBase {
}
//
------------------------------------------------------------------------------------------
- // Validation: rejection paths
+ // Joins without an equi-key (degrade to a nested-loop join)
//
------------------------------------------------------------------------------------------
@Test
- void testRejectMissingEqualityPredicate() {
- final String sql =
+ void testNonEquiJoinUsesNestedLoopJoin() {
+ // Keyless inner join: execute as broadcast NL join.
+ util.verifyRelPlan(
"SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ "input => TABLE b, "
+ "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))"
+ ") AS s "
- + "ON probe.pv > s.bv";
- assertThatThrownBy(() -> util.verifyExecPlan(sql))
- .isInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "LATERAL SNAPSHOT join requires at least one equality
predicate.");
+ + "ON probe.pv > s.bv");
+ }
+
+ @Test
+ void testLeftJoinWithoutEquiKeyUsesNestedLoopJoin() {
+ // Keyless left outer join: execute as broadcast NL join.
+ util.verifyRelPlan(
+ "SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT("
+ + "input => TABLE b, "
+ + "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))"
+ + ") AS s "
+ + "ON probe.pv > s.bv");
}
+ //
------------------------------------------------------------------------------------------
+ // Validation: rejection paths
+ //
------------------------------------------------------------------------------------------
+
@Test
void testRejectSnapshotOutsideLateral() {
// SNAPSHOT used outside a LATERAL context is not rewritten into a
join;
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinBatchSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinBatchSemanticTests.java
index 87725a60a26..e575e54d128 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinBatchSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinBatchSemanticTests.java
@@ -36,7 +36,8 @@ public class LateralSnapshotJoinBatchSemanticTests extends
BatchSemanticTestBase
return List.of(
LateralSnapshotJoinTestPrograms.INNER_JOIN,
LateralSnapshotJoinTestPrograms.LEFT_JOIN,
-
LateralSnapshotJoinTestPrograms.INNER_JOIN_WITH_NON_EQUI_CONDITION,
+ LateralSnapshotJoinTestPrograms.COMPLEX_PRED,
+ LateralSnapshotJoinTestPrograms.NON_EQUI,
LateralSnapshotJoinTestPrograms.SNAPSHOT_ARGUMENTS_IGNORED);
}
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
index f33e94e5c5c..589fc440f77 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/batch/LateralSnapshotJoinTestPrograms.java
@@ -85,10 +85,10 @@ public class LateralSnapshotJoinTestPrograms {
+ "SNAPSHOT(input => TABLE b) AS s ON
probe.pk = s.bk")
.build();
- public static final TableTestProgram INNER_JOIN_WITH_NON_EQUI_CONDITION =
+ public static final TableTestProgram COMPLEX_PRED =
TableTestProgram.of(
- "lateral-snapshot-join-non-equi",
- "batch LATERAL SNAPSHOT join with an additional
non-equi predicate")
+ "lateral-snapshot-join-complex-pred",
+ "batch LATERAL SNAPSHOT join on an equi-key with
an additional non-equi predicate")
.setupTableSource(probe())
.setupTableSource(build())
.setupTableSink(
@@ -103,6 +103,32 @@ public class LateralSnapshotJoinTestPrograms {
+ "ON probe.pk = s.bk AND s.bv > 10")
.build();
+ public static final TableTestProgram NON_EQUI =
+ TableTestProgram.of(
+ "lateral-snapshot-join-non-equi",
+ "batch LATERAL SNAPSHOT join with only a non-equi
condition (no equi-key) "
+ + "uses a broadcast nested-loop join")
+ .setupTableSource(
+ SourceTestStep.newBuilder("probe")
+ .addSchema("pk STRING", "pv INT")
+ .producedValues(
+ Row.of("a", 5), Row.of("b", 15),
Row.of("c", 25))
+ .build())
+ .setupTableSource(
+ SourceTestStep.newBuilder("b")
+ .addSchema("bk STRING", "bv INT")
+ .producedValues(Row.of("a", 10),
Row.of("b", 20))
+ .build())
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("pv INT", "bv INT")
+ .consumedValues(Row.of(5, 10), Row.of(5,
20), Row.of(15, 20))
+ .build())
+ .runSql(
+ "INSERT INTO sink SELECT pv, bv FROM probe JOIN
LATERAL "
+ + "SNAPSHOT(input => TABLE b) AS s ON
probe.pv < s.bv")
+ .build();
+
public static final TableTestProgram SNAPSHOT_ARGUMENTS_IGNORED =
TableTestProgram.of(
"lateral-snapshot-join-arguments-ignored",
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java
index 0d951924c87..6dc6ddf719f 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTestPrograms.java
@@ -174,8 +174,10 @@ public class LateralSnapshotJoinSemanticTestPrograms {
"probe.pk = s.bk AND probe.pv = s.bv"))
.build();
- public static final TableTestProgram NON_EQUI =
- TableTestProgram.of("lateral-snapshot-non-equi", "join with a
non-equi condition")
+ public static final TableTestProgram COMPLEX_PRED =
+ TableTestProgram.of(
+ "lateral-snapshot-complex-pred",
+ "join on an equi-key with an additional non-equi
predicate")
.setupTableSource(
probe(
Arrays.asList(
@@ -200,6 +202,82 @@ public class LateralSnapshotJoinSemanticTestPrograms {
"probe.pk = s.bk AND probe.pv > s.bv"))
.build();
+ public static final TableTestProgram NON_EQUI =
+ TableTestProgram.of(
+ "lateral-snapshot-non-equi",
+ "join with only a non-equi condition (no equi-key)
runs single-threaded")
+ .setupTableSource(
+ probe(
+ Arrays.asList(
+ Row.of("a", 5, ts("00:01:00")),
+ Row.of("a", 15, ts("00:01:01")))))
+ .setupTableSource(
+ appendBuild(
+ withFlipTrigger(
+ Arrays.asList(
+ Row.of("a", 10,
ts("00:00:01")),
+ Row.of("a", 20,
ts("00:00:02"))))))
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("pv INT", "bv INT")
+ .testMaterializedData()
+ .consumedValues("+I[5, 10]", "+I[5, 20]",
"+I[15, 20]")
+ .build())
+ // probe.pv < s.bv keeps the bv=0 flip-trigger row from
matching any probe.
+ .runSql(innerJoin("probe.pv, s.bv", MID_FLIP, "probe.pv <
s.bv"))
+ .build();
+
+ public static final TableTestProgram NON_EQUI_LEFT =
+ TableTestProgram.of(
+ "lateral-snapshot-non-equi-left",
+ "left join with only a non-equi condition
null-pads unmatched probe rows")
+ .setupTableSource(
+ probe(
+ Arrays.asList(
+ Row.of("a", 5, ts("00:01:00")),
+ Row.of("a", 100, ts("00:01:01")))))
+ .setupTableSource(
+ appendBuild(
+ withFlipTrigger(
+ Arrays.asList(
+ Row.of("a", 10,
ts("00:00:01")),
+ Row.of("a", 20,
ts("00:00:02"))))))
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("pv INT", "bv INT")
+ .testMaterializedData()
+ .consumedValues("+I[5, 10]", "+I[5, 20]",
"+I[100, null]")
+ .build())
+ // probe.pv < s.bv keeps the bv=0 flip-trigger row from
matching any probe.
+ .runSql(leftJoin("probe.pv, s.bv", MID_FLIP, "probe.pv <
s.bv"))
+ .build();
+
+ public static final TableTestProgram CROSS_JOIN =
+ TableTestProgram.of(
+ "lateral-snapshot-cross-join",
+ "inner join with a TRUE condition produces the
probe-build cross product")
+ .setupTableSource(
+ probe(
+ Arrays.asList(
+ Row.of("a", 5, ts("00:01:00")),
+ Row.of("a", 15, ts("00:01:01")))))
+ // No flip-trigger row: with a TRUE condition it would
match every probe. The
+ // far-future flip condition flips the operator only at
end-of-input.
+ .setupTableSource(
+ appendBuild(
+ Arrays.asList(
+ Row.of("a", 10, ts("00:00:01")),
+ Row.of("a", 20, ts("00:00:02")))))
+ .setupTableSink(
+ SinkTestStep.newBuilder("sink")
+ .addSchema("pv INT", "bv INT")
+ .testMaterializedData()
+ .consumedValues(
+ "+I[5, 10]", "+I[5, 20]", "+I[15,
10]", "+I[15, 20]")
+ .build())
+ .runSql(innerJoin("probe.pv, s.bv", END_FLIP, "TRUE"))
+ .build();
+
public static final TableTestProgram EMPTY_BUILD_INNER =
TableTestProgram.of(
"lateral-snapshot-empty-build-inner",
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java
index 5ab4917eaea..4805f282875 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/nodes/exec/stream/LateralSnapshotJoinSemanticTests.java
@@ -32,7 +32,10 @@ public class LateralSnapshotJoinSemanticTests extends
SemanticTestBase {
LateralSnapshotJoinSemanticTestPrograms.LEFT_JOIN,
LateralSnapshotJoinSemanticTestPrograms.SELECT_STAR,
LateralSnapshotJoinSemanticTestPrograms.COMPOSITE_KEYS,
+ LateralSnapshotJoinSemanticTestPrograms.COMPLEX_PRED,
LateralSnapshotJoinSemanticTestPrograms.NON_EQUI,
+ LateralSnapshotJoinSemanticTestPrograms.NON_EQUI_LEFT,
+ LateralSnapshotJoinSemanticTestPrograms.CROSS_JOIN,
LateralSnapshotJoinSemanticTestPrograms.EMPTY_BUILD_INNER,
LateralSnapshotJoinSemanticTestPrograms.EMPTY_BUILD_LEFT,
LateralSnapshotJoinSemanticTestPrograms.FLIP_AT_END,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java
index aa5ee760490..1873de41b9f 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.java
@@ -207,6 +207,46 @@ public class LateralSnapshotJoinTest extends TableTestBase
{
+ "ON probe.pk = s.bk");
}
+ @Test
+ void testNonEquiInnerLateralSnapshotJoin() {
+ // An INNER LATERAL SNAPSHOT join without an equi-key runs with a
SINGLETON distribution
+ // (single-threaded), like other Flink joins without an equi-key.
+ util.verifyRelPlan(
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ + "input => TABLE b, on_time => DESCRIPTOR(bts), "
+ + "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))"
+ + ") AS s "
+ + "ON probe.pv > s.bv");
+ }
+
+ @Test
+ void testNonEquiLeftLateralSnapshotJoin() {
+ // A LEFT OUTER LATERAL SNAPSHOT join without an equi-key runs with a
SINGLETON distribution
+ // (single-threaded), like other Flink joins without an equi-key.
+ util.verifyRelPlan(
+ "SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT("
+ + "input => TABLE b, on_time => DESCRIPTOR(bts), "
+ + "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))"
+ + ") AS s "
+ + "ON probe.pv > s.bv");
+ }
+
+ @Test
+ void testLocalPredicateOnJoinKeyColumn() {
+ // Reproduces FLINK-40902: a local filter on the same column that
carries the equi-join
+ // predicate (probe.pk = s.bk together with WHERE probe.pk = 'key1')
must still plan as a
+ // LATERAL SNAPSHOT join. Calcite folds the equi-key to a constant on
each side (pk = 'key1'
+ // AND bk = 'key1') and simplifies the join condition away; the join
then runs single-
+ // threaded (SINGLETON).
+ util.verifyRelPlan(
+ "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
+ + "input => TABLE b, on_time => DESCRIPTOR(bts), "
+ + "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))"
+ + ") AS s "
+ + "ON probe.pk = s.bk "
+ + "WHERE probe.pk = 'key1'");
+ }
+
//
------------------------------------------------------------------------------------------
// Behavior and compilation smoke tests
//
------------------------------------------------------------------------------------------
@@ -510,20 +550,6 @@ public class LateralSnapshotJoinTest extends TableTestBase
{
+ "consuming update and delete changes");
}
- @Test
- void testRejectMissingEqualityPredicate() {
- final String sql =
- "SELECT * FROM probe JOIN LATERAL SNAPSHOT("
- + "input => TABLE b, on_time => DESCRIPTOR(bts), "
- + "load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))"
- + ") AS s "
- + "ON probe.pv > s.bv";
- assertThatThrownBy(() -> util.verifyRelPlan(sql))
- .isInstanceOf(ValidationException.class)
- .hasMessageContaining(
- "LATERAL SNAPSHOT join requires at least one equality
predicate.");
- }
-
@Test
void testRejectNonConstantLoadCompletedTime() {
final String sql =
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.xml
index 3f2e196ab54..17cfaa08bde 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/join/LateralSnapshotJoinTest.xml
@@ -185,6 +185,52 @@ HashJoin(joinType=[LeftOuterJoin], where=[=(pk, bk)],
select=[pk, pv, pts, bk, b
: +- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
+- Exchange(distribution=[hash[bk]])
+- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testLeftJoinWithoutEquiKeyUsesNestedLoopJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT(input => TABLE
b, load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS
TIMESTAMP_LTZ(3))) AS s ON probe.pv > s.bv]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalJoin(condition=[>($1, $4)], joinType=[left])
+ :- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(),
CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL,
DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv,
TIMESTAMP(3) bts)])
+ +- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, b]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+NestedLoopJoin(joinType=[LeftOuterJoin], where=[>(pv, bv)], select=[pk, pv,
pts, bk, bv, bts], build=[right])
+:- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
++- Exchange(distribution=[broadcast])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testNonEquiJoinUsesNestedLoopJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b,
load_completed_time => CAST(TIMESTAMP '2026-07-01 00:00:00' AS
TIMESTAMP_LTZ(3))) AS s ON probe.pv > s.bv]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalJoin(condition=[>($1, $4)], joinType=[inner])
+ :- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0), DEFAULT(),
CAST(2026-07-01 00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL,
DEFAULT(), DEFAULT())], rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv,
TIMESTAMP(3) bts)])
+ +- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, b]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+NestedLoopJoin(joinType=[InnerJoin], where=[>(pv, bv)], select=[pk, pv, pts,
bk, bv, bts], build=[right])
+:- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
++- Exchange(distribution=[broadcast])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.xml
index ecfbe8d985e..5a4bdb252ca 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/join/LateralSnapshotJoinTest.xml
@@ -304,6 +304,94 @@ Calc(select=[pk, pv, bv])
+- Exchange(distribution=[hash[bk]])
+- WatermarkAssigner(rowtime=[bts], watermark=[bts])
+- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testLocalPredicateOnJoinKeyColumn">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b,
on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pk = s.bk WHERE probe.pk =
'key1']]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalFilter(condition=[=($0, _UTF-16LE'key1')])
+ +- LogicalJoin(condition=[=($0, $3)], joinType=[inner])
+ :- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
+ : +- LogicalTableScan(table=[[default_catalog, default_database,
probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0),
DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01
00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())],
rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+ +- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database,
b]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+Calc(select=[CAST('key1' AS VARCHAR(2147483647)) AS pk, pv, pts, CAST('key1'
AS VARCHAR(2147483647)) AS bk, bv, bts])
++- LateralSnapshotJoin(joinType=[InnerJoin], where=[true], select=[pv, pts,
bv, bts], loadCompletedCondition=[user_time], loadCompletedTime=[1782864000000])
+ :- Exchange(distribution=[single])
+ : +- Calc(select=[pv, pts], where=[=(pk, 'key1')])
+ : +- WatermarkAssigner(rowtime=[pts], watermark=[pts])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
probe]], fields=[pk, pv, pts])
+ +- Exchange(distribution=[single])
+ +- Calc(select=[bv, bts], where=[=(bk, 'key1')])
+ +- WatermarkAssigner(rowtime=[bts], watermark=[bts])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testNonEquiInnerLateralSnapshotJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM probe JOIN LATERAL SNAPSHOT(input => TABLE b,
on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP '2026-07-01
00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pv > s.bv]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalJoin(condition=[>($1, $4)], joinType=[inner])
+ :- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
+ : +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0),
DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01
00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())],
rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+ +- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, b]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LateralSnapshotJoin(joinType=[InnerJoin], where=[>(pv, bv)], select=[pk, pv,
pts, bk, bv, bts], loadCompletedCondition=[user_time],
loadCompletedTime=[1782864000000])
+:- Exchange(distribution=[single])
+: +- WatermarkAssigner(rowtime=[pts], watermark=[pts])
+: +- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
++- Exchange(distribution=[single])
+ +- WatermarkAssigner(rowtime=[bts], watermark=[bts])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testNonEquiLeftLateralSnapshotJoin">
+ <Resource name="sql">
+ <![CDATA[SELECT * FROM probe LEFT JOIN LATERAL SNAPSHOT(input => TABLE
b, on_time => DESCRIPTOR(bts), load_completed_time => CAST(TIMESTAMP
'2026-07-01 00:00:00' AS TIMESTAMP_LTZ(3))) AS s ON probe.pv > s.bv]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(pk=[$0], pv=[$1], pts=[$2], bk=[$3], bv=[$4], bts=[$5])
++- LogicalJoin(condition=[>($1, $4)], joinType=[left])
+ :- LogicalWatermarkAssigner(rowtime=[pts], watermark=[$2])
+ : +- LogicalTableScan(table=[[default_catalog, default_database, probe]])
+ +- LogicalTableFunctionScan(invocation=[SNAPSHOT(TABLE(#0),
DESCRIPTOR(_UTF-16LE'bts'), CAST(2026-07-01
00:00:00):TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL, DEFAULT(), DEFAULT())],
rowType=[RecordType(VARCHAR(2147483647) bk, INTEGER bv, TIMESTAMP(3) bts)])
+ +- LogicalProject(bk=[$0], bv=[$1], bts=[$2])
+ +- LogicalWatermarkAssigner(rowtime=[bts], watermark=[$2])
+ +- LogicalTableScan(table=[[default_catalog, default_database, b]])
+]]>
+ </Resource>
+ <Resource name="optimized rel plan">
+ <![CDATA[
+LateralSnapshotJoin(joinType=[LeftOuterJoin], where=[>(pv, bv)], select=[pk,
pv, pts, bk, bv, bts], loadCompletedCondition=[user_time],
loadCompletedTime=[1782864000000])
+:- Exchange(distribution=[single])
+: +- WatermarkAssigner(rowtime=[pts], watermark=[pts])
+: +- TableSourceScan(table=[[default_catalog, default_database, probe]],
fields=[pk, pv, pts])
++- Exchange(distribution=[single])
+ +- WatermarkAssigner(rowtime=[bts], watermark=[bts])
+ +- TableSourceScan(table=[[default_catalog, default_database, b]],
fields=[bk, bv, bts])
]]>
</Resource>
</TestCase>
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 628eebe12aa..d9c3acf43bb 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
@@ -512,6 +512,15 @@ public class LateralSnapshotJoinOperator extends
AbstractStreamOperator<RowData>
minStateTtlMs,
maxStateTtlMs,
buildRowtimeIndex);
+
+ if (filterNullKeys.length == 0 && minStateTtlMs > 0) {
+ // No equi-key: all build-side rows share a single (empty) key, so
state_ttl can only
+ // evict the whole build side at once after it has been idle
longer than the TTL.
+ LOG.warn(
+ "LATERAL SNAPSHOT join has no equi-key; state_ttl evicts
the entire build side "
+ + "at once after it has been idle for longer than
the TTL, not per join "
+ + "key.");
+ }
}
@Override
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 1fb82fad800..4f22e3f8572 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
@@ -36,6 +36,7 @@ import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.binary.BinaryRowData;
import org.apache.flink.table.data.writer.BinaryRowWriter;
import org.apache.flink.table.runtime.generated.GeneratedJoinCondition;
+import org.apache.flink.table.runtime.keyselector.EmptyRowDataKeySelector;
import
org.apache.flink.table.runtime.operators.join.snapshot.LateralSnapshotJoinOperator.Phase;
import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
import org.apache.flink.table.runtime.util.RowDataHarnessAssertor;
@@ -146,6 +147,22 @@ class LateralSnapshotJoinOperatorTest {
+ " }\n"
+ "}\n";
+ /**
+ * Non-equi join condition that matches when the probe id (field 0) is
less than the build
+ * row-time (field 2). Used to exercise the generated condition in
empty-key (keyless) mode.
+ */
+ private static final String PROBE_ID_LT_BUILD_RT_JOIN_FUNC_CODE =
+ "public class LateralSnapshotJoinConditionIdLtRt extends "
+ +
"org.apache.flink.api.common.functions.AbstractRichFunction "
+ + "implements
org.apache.flink.table.runtime.generated.JoinCondition {\n"
+ + " public LateralSnapshotJoinConditionIdLtRt(Object[]
reference) {}\n"
+ + " @Override public boolean apply("
+ + " org.apache.flink.table.data.RowData in1,"
+ + " org.apache.flink.table.data.RowData in2) {\n"
+ + " return in1.getLong(0) < in2.getLong(2);\n"
+ + " }\n"
+ + "}\n";
+
private static GeneratedJoinCondition newTrueCondition() {
return new GeneratedJoinCondition(
"LateralSnapshotJoinConditionStub",
ALWAYS_TRUE_JOIN_FUNC_CODE, new Object[0]);
@@ -156,6 +173,13 @@ class LateralSnapshotJoinOperatorTest {
"LateralSnapshotJoinConditionMatchVal",
MATCH_VAL_JOIN_FUNC_CODE, new Object[0]);
}
+ private static GeneratedJoinCondition newProbeIdLtBuildRtCondition() {
+ return new GeneratedJoinCondition(
+ "LateralSnapshotJoinConditionIdLtRt",
+ PROBE_ID_LT_BUILD_RT_JOIN_FUNC_CODE,
+ new Object[0]);
+ }
+
// -----------------------------------------------------------------
Operator / harness
// factories
@@ -200,6 +224,20 @@ class LateralSnapshotJoinOperatorTest {
op, PROBE_KEY_SELECTOR, BUILD_KEY_SELECTOR, KEY_TYPE);
}
+ /**
+ * Harness that runs the operator in non-key (SINGLETON) mode: both inputs
use {@link
+ * EmptyRowDataKeySelector}, so every row maps to one empty key. This is
the shape a LATERAL
+ * SNAPSHOT join takes when it doesn't have an equality predicate.
+ */
+ private static KeyedTwoInputStreamOperatorTestHarness<RowData, RowData,
RowData, RowData>
+ newEmptyKeyHarness(LateralSnapshotJoinOperator op) throws
Exception {
+ return new KeyedTwoInputStreamOperatorTestHarness<>(
+ op,
+ EmptyRowDataKeySelector.INSTANCE,
+ EmptyRowDataKeySelector.INSTANCE,
+ EmptyRowDataKeySelector.INSTANCE.getProducedType());
+ }
+
private static KeySelector<RowData, RowData>
nullSafeStringKeySelector(final int keyIdx) {
return value -> {
BinaryRowData ret = new BinaryRowData(1);
@@ -509,6 +547,112 @@ class LateralSnapshotJoinOperatorTest {
}
}
+ @Test
+ void emptyKeyModeInnerNonEquiJoin() throws Exception {
+ // INNER JOIN with an empty keyset and a non-equi condition (probe.id
< build.rt). This
+ // happens if the join has no equality predicate (or it was folded
away). The empty key puts
+ // all build rows in one group; the generated condition then selects
the matching ones.
+ LateralSnapshotJoinOperator op =
+ newOperator(
+ false, newProbeIdLtBuildRtCondition(), new boolean[0],
100L, null, null);
+ try (KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData> h =
+ newEmptyKeyHarness(op)) {
+ h.open();
+ addBuildChange(h, insertRecord("k1", "v1", 10L));
+ addBuildChange(h, insertRecord("k2", "v2", 20L));
+
+ // LOAD phase: buffered probe id=15 matches only k2 at the flip
(15 < 20, not < 10).
+ addProbeRecord(h, 15L, "px", "p1");
+ addProbeWm(h, 80L); // lets the flip drain the buffered probe
+ addBuildWm(h, 100L); // flip to JOIN
+ assertPhase(op, Phase.JOIN);
+ stripWatermarksAndStatusesFromOutput(h);
+ JOINED_ASSERTOR.shouldEmitAll(h, row(15L, "px", "p1", "k2", "v2",
20L));
+
+ // JOIN phase: id=5 matches k1 and k2; id=18 matches k2; id=25
matches none (no output).
+ addProbeRecord(h, 5L, "py", "p2");
+ addProbeRecord(h, 18L, "pz", "p3");
+ addProbeRecord(h, 25L, "pz", "p4");
+ stripWatermarksAndStatusesFromOutput(h);
+ JOINED_ASSERTOR.shouldEmitAll(
+ h,
+ row(5L, "py", "p2", "k1", "v1", 10L),
+ row(5L, "py", "p2", "k2", "v2", 20L),
+ row(18L, "pz", "p3", "k2", "v2", 20L));
+ }
+ }
+
+ @Test
+ void emptyKeyModeLeftOuterNonEquiJoin() throws Exception {
+ // LEFT OUTER JOIN with an empty keyset and a non-equi condition
(probe.id < build.rt). A
+ // probe that matches no build row in the group is emitted null-padded.
+ LateralSnapshotJoinOperator op =
+ newOperator(true, newProbeIdLtBuildRtCondition(), new
boolean[0], 100L, null, null);
+ try (KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData> h =
+ newEmptyKeyHarness(op)) {
+ h.open();
+ addBuildChange(h, insertRecord("k1", "v1", 10L));
+ addBuildChange(h, insertRecord("k2", "v2", 20L));
+
+ // LOAD phase: buffered probe id=100 matches no build row ->
null-padded at the flip.
+ addProbeRecord(h, 100L, "px", "p1");
+ addProbeWm(h, 80L); // lets the flip drain the buffered probe
+ addBuildWm(h, 100L); // flip to JOIN
+ assertPhase(op, Phase.JOIN);
+ stripWatermarksAndStatusesFromOutput(h);
+ JOINED_ASSERTOR.shouldEmitAll(h, row(100L, "px", "p1", null, null,
null));
+
+ // JOIN phase: id=5 matches k1 and k2; id=18 matches k2; id=25
matches none
+ // (null-padded).
+ addProbeRecord(h, 5L, "py", "p2");
+ addProbeRecord(h, 18L, "pz", "p3");
+ addProbeRecord(h, 25L, "pzz", "p4");
+ stripWatermarksAndStatusesFromOutput(h);
+ JOINED_ASSERTOR.shouldEmitAll(
+ h,
+ row(5L, "py", "p2", "k1", "v1", 10L),
+ row(5L, "py", "p2", "k2", "v2", 20L),
+ row(18L, "pz", "p3", "k2", "v2", 20L),
+ row(25L, "pzz", "p4", null, null, null));
+ }
+ }
+
+ @Test
+ void emptyKeyModeLeftOuterReflectsBuildInsertAndDelete() throws Exception {
+ // LEFT OUTER with an empty keyset: a matching build row is added and
later removed, so the
+ // same probe sees null (no match), then non-null (match), then null
(match removed).
+ LateralSnapshotJoinOperator op =
+ newOperator(true, newProbeIdLtBuildRtCondition(), new
boolean[0], 100L, null, null);
+ try (KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData> h =
+ newEmptyKeyHarness(op)) {
+ h.open();
+ addBuildWm(h, 100L); // flip to JOIN with an empty build table
+ assertPhase(op, Phase.JOIN);
+
+ // (1) No matching build row yet -> null-padded.
+ addProbeRecord(h, 5L, "p", "p1");
+ stripWatermarksAndStatusesFromOutput(h);
+ JOINED_ASSERTOR.shouldEmitAll(h, row(5L, "p", "p1", null, null,
null));
+
+ // Add a matching build row (probe id 5 < build rt 110) and drain
it.
+ addBuildChange(h, insertRecord("k1", "v1", 110L));
+ addBuildWm(h, 111L);
+ // (2) Probe now matches -> non-null.
+ addProbeRecord(h, 5L, "p", "p2");
+ stripWatermarksAndStatusesFromOutput(h);
+ JOINED_ASSERTOR.shouldEmitAll(h, row(5L, "p", "p2", "k1", "v1",
110L));
+
+ // Remove the build row again (retraction matches the inserted
row, incl. row-time) and
+ // drain it.
+ addBuildChange(h, deleteRecord("k1", "v1", 110L));
+ addBuildWm(h, 121L);
+ // (3) Probe matches nothing again -> null-padded.
+ addProbeRecord(h, 5L, "p", "p3");
+ stripWatermarksAndStatusesFromOutput(h);
+ JOINED_ASSERTOR.shouldEmitAll(h, row(5L, "p", "p3", null, null,
null));
+ }
+ }
+
@Test
void joinPhaseProbeIsBufferedWhileKeyStillDraining() throws Exception {
// Flip without a probe watermark leaves the LOAD-buffered probe in
place. A new JOIN-phase
@@ -1415,6 +1559,39 @@ class LateralSnapshotJoinOperatorTest {
}
}
+ @Test
+ void emptyKeyModeStateTtlEvictsEntireBuildTable() throws Exception {
+ // With no equi-key all build rows share one (empty) key, so a single
state_ttl timer covers
+ // the whole build side. After an idle period longer than the TTL, the
entire build table is
+ // evicted at once; a later probe then matches nothing (here: LEFT ->
null-padded).
+ LateralSnapshotJoinOperator op =
+ newOperator(true, newProbeIdLtBuildRtCondition(), new
boolean[0], 100L, null, 100L);
+ try (KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData,
RowData> h =
+ newEmptyKeyHarness(op)) {
+ h.setProcessingTime(0);
+ h.open();
+ addBuildChange(h, insertRecord("k1", "v1", 10L));
+ addBuildChange(h, insertRecord("k2", "v2", 20L));
+ addBuildWm(h, 100L); // flip to JOIN
+ assertPhase(op, Phase.JOIN);
+
+ // Probe id=5 matches both build rows (5 < 10, 5 < 20).
+ addProbeRecord(h, 5L, "p", "p1");
+ stripWatermarksAndStatusesFromOutput(h);
+ JOINED_ASSERTOR.shouldEmitAll(
+ h, row(5L, "p", "p1", "k1", "v1", 10L), row(5L, "p", "p1",
"k2", "v2", 20L));
+
+ // Idle well past the TTL: the single empty-key timer evicts the
whole build table.
+ h.setProcessingTime(1000);
+
+ // The same probe now matches nothing -> null-padded, confirming
the whole table was
+ // evicted together.
+ addProbeRecord(h, 5L, "p", "p2");
+ stripWatermarksAndStatusesFromOutput(h);
+ JOINED_ASSERTOR.shouldEmitAll(h, row(5L, "p", "p2", null, null,
null));
+ }
+ }
+
// -----------------------------------------------------------------
Snapshot / restore
@ParameterizedTest(name = "backend={0}")
