[
https://issues.apache.org/jira/browse/FLINK-41368?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Fabian Hueske updated FLINK-41368:
----------------------------------
Description:
[FLIP-579|https://cwiki.apache.org/confluence/display/FLINK/FLIP-579%3A+LATERAL+SNAPSHOT+Join]
adds {{LATERAL SNAPSHOT}} join, a stream enrichment join. It joins an
append-only probe table with the *current* state of an updating build table. A
load phase first fills the build-side state and buffers probe rows. Afterwards,
every probe row is joined immediately, without waiting for build-side
watermarks. Neither side needs a primary key. The result is non-deterministic
in streaming mode and deterministic in batch mode.
Docs: [LATERAL SNAPSHOT
Join|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/sql/reference/queries/joins/#lateral-snapshot-join]
and [Lateral Snapshot Join (Table
API)|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/dev/table/tableapi/#lateral-snapshot-join].
Planner, operator and restore tests are covered by unit and IT tests. This
ticket is about the end-to-end path on a real cluster: phases, metrics,
savepoints, batch mode and the Table API.
h3. 0. Setup
Build {{release-2.4}}, or use the 2.4.0 RC.
{code:bash}
./mvnw clean install -DskipTests -Dfast -Pskip-webui-build -T1C
cd build-target
mkdir -p /tmp/flip579/build /tmp/flip579/savepoints /tmp/flip579/checkpoints
{code}
In {{conf/config.yaml}}, change {{numberOfTaskSlots: 1}} to {{4}}, and append:
{code:yaml}
execution.checkpointing.dir: file:///tmp/flip579/checkpoints
execution.checkpointing.savepoint-dir: file:///tmp/flip579/savepoints
execution.checkpointing.interval: 10s
{code}
Start the cluster with {{./bin/start-cluster.sh}}. Save the following as
{{/tmp/flip579/ddl.sql}} and start the SQL Client with {{./bin/sql-client.sh
\-i /tmp/flip579/ddl.sql}}. Run every query in the sections below in that
session. Queries that start with {{INSERT INTO}} run detached on the cluster,
and you can see their result in
{{log/flink-<user>-taskexecutor-<n>-<host>.out}} (the {{print}} sink). Use
{{tail \-f}} on it, and do not delete or truncate the file while the
TaskManager is running (output is lost until the cluster restarts). Note the
line count instead.
The SQL Client only loads {{ddl.sql}} via {{\-i}} on start, and the tables and
views created in section 4 live only in the session. If you restart the client,
pass {{\-i}} again and recreate them.
The build side is a directory of {{debezium-json}} files that the filesystem
source monitors. A file you drop into the directory is read as build-side
changes, including updates and deletes. The probe side is a datagen source with
one order per second and the currency cycling A, B, C.
{code:sql}
SET 'table.local-time-zone' = 'UTC';
SET 'sql-client.execution.result-mode' = 'tableau';
CREATE TABLE rates (
currency STRING, rate DOUBLE, update_time TIMESTAMP(3),
WATERMARK FOR update_time AS update_time
) WITH (
'connector' = 'filesystem', 'path' = '/tmp/flip579/build',
'format' = 'debezium-json', 'source.monitor-interval' = '2s');
CREATE TABLE orders (
order_id BIGINT,
currency AS CASE order_id % 3 WHEN 0 THEN 'A' WHEN 1 THEN 'B' ELSE 'C' END,
ts AS LOCALTIMESTAMP,
WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
) WITH ('connector' = 'datagen', 'rows-per-second' = '1',
'fields.order_id.kind' = 'sequence', 'fields.order_id.start' = '1',
'fields.order_id.end' = '100000');
CREATE TABLE out_print (order_id BIGINT, currency STRING, rate DOUBLE) WITH
('connector' = 'print');
{code}
{{SET}} statements stay active for the rest of the SQL Client session. The
sections below tell you when to {{RESET}} a setting. Alternatively, restart the
SQL Client between sections.
The operator metrics are visible in the Web UI ({{http://localhost:8081}}):
open the job, click the {{LateralSnapshotJoin}} vertex and go to the Metrics
tab (the metric names are prefixed with {{LateralSnapshotJoin[n].}}). The same
values are available via REST, e.g. {{GET
/jobs/<job-id>/vertices/<vertex-id>/subtasks/metrics?get=LateralSnapshotJoin[9].currentPhase}}.
The operator number may differ. Metric values refresh only about every 10
seconds, so a check such as "immediately" or "after a few seconds" can look
stale for that long.
*Build-side changes are applied when the build-side watermark passes their row
time* (they are buffered until then, see the FLIP). A build row is therefore
invisible to the join until a later build-side watermark arrives. All build
files below end with a row for a dummy key {{Z}} with a later {{update_time}}
for this reason. It pushes the watermark, and no probe row matches it. Without
it, the last rows of a file stay invisible. In real pipelines the next
build-side record does this.
h3. 1. Load phase and flip via idle timeout
Start with an empty {{/tmp/flip579/build}} and submit:
{code:sql}
INSERT INTO out_print
SELECT o.order_id, o.currency, r.rate
FROM orders AS o
JOIN LATERAL SNAPSHOT(
input => TABLE rates,
on_time => DESCRIPTOR(update_time),
load_completed_idle_timeout => INTERVAL '10' SECOND) AS r
ON o.currency = r.currency;
{code}
The probe side runs, but nothing is emitted yet. Then write the initial build
state:
{code:bash}
cat > /tmp/flip579/build/001.json <<'EOF'
{"before":null,"after":{"currency":"A","rate":1.0,"update_time":"2026-01-01
10:00:00"},"op":"c"}
{"before":null,"after":{"currency":"B","rate":2.0,"update_time":"2026-01-01
10:00:00"},"op":"c"}
{"before":null,"after":{"currency":"Z","rate":0.0,"update_time":"2026-01-01
11:00:00"},"op":"c"}
EOF
{code}
Expected:
* Before the file is written, {{currentPhase}} = 0 and
{{numProbeSideRecordsBuffered}} grows by about 1 per second. Nothing is printed.
* About 10 seconds after the last build-side watermark advance, the join flips:
{{currentPhase}} = 1 and {{numProbeSideRecordsBuffered}} drops to 0.
* All buffered orders are printed at once: A orders with rate 1.0, B orders
with rate 2.0. C orders are not printed (inner join), and
{{numUnmatchedProbeRecords}} counts them.
* New orders are printed immediately afterwards.
* {{currentBuildSideWatermark}} = {{1767265200000}} ({{2026-01-01 11:00:00}}
UTC). {{currentProbeSideWatermark}} is close to the current time in epoch
milliseconds. {{maxJoinFanOut}} = 1.
h3. 2. Stuck in the load phase
Cancel the job and empty the build directory ({{rm /tmp/flip579/build/\*}}).
Submit the same query *without* {{load_completed_idle_timeout}}, and drop
{{001.json}} again. The default {{load_completed_time}} is the wall-clock time
at query compilation, so the build-side watermark (January 2026) never reaches
it.
Expected: {{currentPhase}} stays 0, {{numProbeSideRecordsBuffered}} keeps
growing, there is no output, and {{currentBuildSideWatermark}} stays at
{{1767265200000}}.
Then fix it with an explicit load completion time. Cancel the job, empty the
build directory, and submit:
{code:sql}
INSERT INTO out_print
SELECT o.order_id, o.currency, r.rate
FROM orders AS o
JOIN LATERAL SNAPSHOT(input => TABLE rates, on_time => DESCRIPTOR(update_time),
load_completed_time => CAST(TIMESTAMP '2026-01-02 00:00:00' AS
TIMESTAMP_LTZ(3))) AS r
ON o.currency = r.currency;
{code}
Drop {{001.json}} (its dummy {{Z}} row moves the watermark to 2026-01-01 11:00,
which is below the load completion time). The join stays in the load phase.
Then drop a file that moves the watermark past the load completion time:
{code:bash}
cat > /tmp/flip579/build/002.json <<'EOF'
{"before":null,"after":{"currency":"Z","rate":0.0,"update_time":"2026-01-02
00:00:00"},"op":"c"}
EOF
{code}
Expected: the join flips as soon as the watermark reaches 2026-01-02 00:00:00
({{currentBuildSideWatermark}} = {{1767312000000}}), {{currentPhase}} = 1,
buffered orders are printed.
h3. 3. Build-side updates, deletes and LEFT JOIN
Cancel the job, empty the build directory and submit the section 1 query as
{{LEFT JOIN LATERAL SNAPSHOT(...)}} (still with the idle timeout). Drop
{{001.json}} from section 1 and wait for the flip. A orders have rate 1.0, B
orders 2.0, and C orders are printed with a {{NULL}} rate. Then drop:
{code:bash}
cat > /tmp/flip579/build/003.json <<'EOF'
{"before":{"currency":"A","rate":1.0,"update_time":"2026-01-01
10:00:00"},"after":{"currency":"A","rate":1.5,"update_time":"2026-02-01
10:00:00"},"op":"u"}
{"before":{"currency":"B","rate":2.0,"update_time":"2026-01-01
10:00:00"},"after":null,"op":"d"}
{"before":null,"after":{"currency":"C","rate":3.0,"update_time":"2026-02-01
10:00:00"},"op":"c"}
{"before":null,"after":{"currency":"Z","rate":0.0,"update_time":"2026-02-02
10:00:00"},"op":"c"}
EOF
{code}
Expected after a few seconds:
* New A orders have rate 1.5. Rate 1.0 is never mixed with 1.5 in an
inconsistent way: the update becomes visible atomically.
* New B orders have a {{NULL}} rate (the row was deleted).
* New C orders have rate 3.0.
* Already printed rows are never changed. The result is append-only.
* {{numUnmatchedBuildRetractions}} stays 0, because all retractions hit
existing rows.
* Note: {{numUnmatchedProbeRecords}} stays 0 for a LEFT JOIN, although
unmatched rows are emitted null-padded. It only counts unmatched rows of INNER
joins (the docs describe it more generally).
h3. 4. Planning-time validation
Create the extra tables and views (add to the SQL Client session):
{code:sql}
CREATE TABLE rates_nowm (currency STRING, rate DOUBLE, update_time TIMESTAMP(3))
WITH
('connector'='filesystem','path'='/tmp/flip579/build','format'='debezium-json','source.monitor-interval'='2s');
CREATE VIEW rates_rand AS SELECT currency, rate, RAND() AS r, update_time FROM
rates;
CREATE VIEW probe_agg AS SELECT currency, COUNT(*) AS c FROM orders GROUP BY
currency;
{code}
Each of the following statements must be rejected with the quoted error. The
full statement is {{EXPLAIN SELECT o.order_id FROM orders AS o <fragment> ON
o.currency = r.currency;}}, with the {{<fragment>}} from the first column (in
the {{probe_agg}} row the fragment replaces {{orders AS o}} as well, and the
standalone {{SNAPSHOT}} row is a complete statement without {{ON}}). For
{{RIGHT}}/{{FULL}} use the arguments {{(input => TABLE rates, on_time =>
DESCRIPTOR(update_time))}}. A failing statement aborts a {{\-f}} script, so
paste the statements one at a time into the interactive client.
|| Statement || Expected error ||
| {{JOIN LATERAL SNAPSHOT(input => TABLE rates_nowm, on_time =>
DESCRIPTOR(update_time)) AS r}} | {{Argument 'on_time' of SNAPSHOT must
reference a rowtime attribute (a column with a watermark), but column
'update_time' is not one.}} |
| {{JOIN LATERAL SNAPSHOT(input => TABLE rates) AS r}} (no {{on_time}}) |
{{LATERAL SNAPSHOT requires the 'on_time' argument to identify the build-side
rowtime attribute.}} |
| {{on_time => DESCRIPTOR(rate)}} | {{Argument 'on_time' of SNAPSHOT must
reference a TIMESTAMP or TIMESTAMP_LTZ column (up to precision 3), but column
'rate' has type 'DOUBLE'.}} |
| {{FROM probe_agg AS o JOIN LATERAL SNAPSHOT(input => TABLE rates, on_time =>
DESCRIPTOR(update_time)) AS r}} | {{The probe (left) input of LATERAL SNAPSHOT
join doesn't support consuming update changes ...}} |
| {{RIGHT JOIN LATERAL SNAPSHOT(...)}} / {{FULL JOIN LATERAL SNAPSHOT(...)}} |
{{LATERAL SNAPSHOT join only supports INNER JOIN and LEFT OUTER JOIN, but was
RIGHT JOIN.}} (and {{FULL JOIN}}) |
| {{SELECT * FROM TABLE(SNAPSHOT(input => TABLE rates, on_time =>
DESCRIPTOR(update_time)))}} | {{The SNAPSHOT function can only be used as the
build side (right-hand side) of a LATERAL join. ...}} |
Non-deterministic build columns (FLINK-40890). The check is part of the
non-deterministic update handling, so enable it first:
{code:sql}
SET 'table.optimizer.non-deterministic-update.strategy' = 'TRY_RESOLVE';
EXPLAIN SELECT o.order_id, r.rate, r.r FROM orders AS o
JOIN LATERAL SNAPSHOT(input => TABLE rates_rand, on_time =>
DESCRIPTOR(update_time)) AS r ON o.currency = r.currency;
-- TableException: The column(s): r(generated by non-deterministic function:
RAND ) can not satisfy the determinism requirement ...
RESET 'table.optimizer.non-deterministic-update.strategy';
{code}
The same statement is accepted with the default strategy ({{IGNORE}}), and also
with {{TRY_RESOLVE}} if the non-deterministic column is not selected.
The {{TABLE}} keyword in {{LATERAL}} calls is optional. Verify that {{JOIN
LATERAL SNAPSHOT(...)}}, {{JOIN LATERAL TABLE(SNAPSHOT(...))}} and {{CROSS JOIN
LATERAL SNAPSHOT(...) ... WHERE o.currency = r.currency}} all plan, and that
{{EXPLAIN}} shows a {{LateralSnapshotJoin}} node.
h3. 5. Join predicates and parallelism
Run the section 1 query as an {{INSERT INTO}} with {{SET 'parallelism.default'
= '2';}} and different {{ON}} clauses, and check the parallelism of the
{{LateralSnapshotJoin}} vertex in the Web UI. At parallelism 2, one build-side
source subtask never reads data, so aggregated values of
{{currentBuildSideWatermark}} can show Long.MIN_VALUE. This is expected, so
look at the metric per subtask. Cancel each job before submitting the next one,
and run {{RESET 'parallelism.default';}} when you are done.
* {{ON o.currency = r.currency AND o.order_id >= 0}} (equi and non-equi mixed):
the join runs with parallelism 2.
* {{ON o.currency <> r.currency}} (no equality predicate): the join runs with
parallelism 1, while the sources run with 2. After flipping, each order is
joined with all non-matching currencies, so {{maxJoinFanOut}} > 1 and
{{avgJoinFanOut}} shows the average (the dummy {{Z}} row also matches).
h3. 6. State TTL
Run the section 1 query with an additional {{state_ttl => INTERVAL '20'
SECOND}}, and drop {{001.json}}. After the flip and the TTL ({{t ~ 40-50 s}}):
* {{numStateTtlEvictions}} = 1. The key {{Z}} was never accessed by a probe row
and is evicted.
* The keys {{A}} and {{B}} are accessed by probe rows all the time, so they are
kept and orders are still matched with their rates.
* {{numStateTtlEvictions}} stays 0 during the load phase.
h3. 7. Savepoint and restore
*Join phase.* Run the job from section 6 (or section 1) until {{currentPhase}}
= 1, then:
{code:bash}
./bin/flink stop <job-id>
# prints: Savepoint completed. Path: file:/tmp/flip579/savepoints/savepoint-...
{code}
In the SQL Client, run {{SET 'execution.state-recovery.path' = '<savepoint
path>';}} and submit the same {{INSERT INTO}} again. Then run {{RESET
'execution.state-recovery.path';}}, otherwise every later submission in this
session restores from that savepoint. Expected:
* The join is in the join phase right away ({{currentPhase}} = 1), without any
new build-side data.
* A and B orders are still matched. The build-side state was restored.
* The metric values are not restored. Counters start at 0, and
{{currentBuildSideWatermark}} is {{\-9.2E18}} (Long.MIN_VALUE) until the next
build-side watermark arrives.
*Load phase.* Cancel the job and empty the build directory. Submit the second
query of section 2 (with {{load_completed_time}}) without build data. Let it
buffer about 30 probe rows, then stop it with {{./bin/flink stop}}. Restore it
with {{execution.state-recovery.path}} (and {{RESET}} it afterwards), and
verify that {{currentPhase}} = 0 after the restore. Then drop {{001.json}} with
the {{Z}} row at {{2026-01-03 00:00:00}}. Expected: the join flips, and the
output contains the orders that were buffered *before* the savepoint
({{order_id}} starting at 1, with rates for A and B). The
{{numProbeSideRecordsBuffered}} gauge only counts rows seen since the restore,
so check the output instead.
h3. 8. Batch mode
{code:bash}
mkdir -p /tmp/flip579/orders_b /tmp/flip579/rates_b
printf '1,A\n2,B\n3,C\n4,A\n' > /tmp/flip579/orders_b/o.csv
printf 'A,1.0\nB,2.0\nA,1.5\n' > /tmp/flip579/rates_b/r.csv
{code}
{code:sql}
SET 'execution.runtime-mode' = 'batch';
CREATE TABLE orders_b (order_id BIGINT, currency STRING) WITH ('connector' =
'filesystem', 'path' = '/tmp/flip579/orders_b', 'format' = 'csv');
CREATE TABLE rates_b (currency STRING, rate DOUBLE) WITH ('connector' =
'filesystem', 'path' = '/tmp/flip579/rates_b', 'format' = 'csv');
SELECT o.order_id, o.currency, r.rate
FROM orders_b AS o
LEFT JOIN LATERAL SNAPSHOT(input => TABLE rates_b, state_ttl => INTERVAL '1'
HOUR) AS r
ON o.currency = r.currency ORDER BY o.order_id, r.rate;
RESET 'execution.runtime-mode';
{code}
Expected: the query finishes without watermarks and without {{on_time}}.
{{state_ttl}} is accepted and ignored. The result is the regular left join
against the complete build side (the SQL Client renders it as a table and
prints {{<NULL>}}; the rows are the same):
{noformat}
1 | A | 1.0
1 | A | 1.5
2 | B | 2.0
3 | C | NULL
4 | A | 1.0
4 | A | 1.5
{noformat}
Repeated runs return the same result. {{EXPLAIN}} of the inner join variant
shows a regular join ({{HashJoin}}/{{AdaptiveJoin}}) and no
{{LateralSnapshotJoin}}.
h3. 9. Table API
Save the following as {{LsjTableApi.java}} in {{build-target}} and run it from
there with {{java -cp "lib/\*" LsjTableApi.java}} from {{build-target}} (Java
17 single-file launch). The columns of the two tables must have different
names, hence the {{r_}} prefix.
{code:java}
import static org.apache.flink.table.api.Expressions.*;
import org.apache.flink.table.api.*;
public class LsjTableApi {
public static void main(String[] args) throws Exception {
TableEnvironment env =
TableEnvironment.create(EnvironmentSettings.inStreamingMode());
env.executeSql(
"CREATE TABLE rates (r_currency STRING, r_rate DOUBLE,
update_time TIMESTAMP(3),"
+ " WATERMARK FOR update_time AS update_time) WITH ("
+
"'connector'='filesystem','path'='/tmp/flip579/build','format'='debezium-json')");
env.executeSql(
"CREATE TABLE orders (order_id BIGINT, currency STRING, ts AS
LOCALTIMESTAMP,"
+ " WATERMARK FOR ts AS ts - INTERVAL '1' SECOND) WITH
('connector'='datagen')");
Table orders = env.from("orders");
Table rates = env.from("rates");
Table inner = orders.joinLateral(
call("SNAPSHOT", rates.asArgument("input"),
descriptor("update_time").asArgument("on_time")),
$("currency").isEqual($("r_currency")));
System.out.println(inner.explain());
Table left = orders.leftOuterJoinLateral(
call("SNAPSHOT", rates.asArgument("input"),
descriptor("update_time").asArgument("on_time")),
$("currency").isEqual($("r_currency")));
System.out.println(left.explain());
}
}
{code}
Expected: both plans contain a {{LateralSnapshotJoin(joinType=[InnerJoin])}}
and {{LateralSnapshotJoin(joinType=[LeftOuterJoin])}} node respectively.
Optionally replace {{explain()}} with {{executeInsert}} into a {{print}} table
and compare the behavior to sections 1 and 3 (the Table API job runs in a local
MiniCluster, so there is no Web UI).
h3. Exploratory
* Real sources: a probe side from Kafka and a build side from an
{{upsert-kafka}} topic, with watermark alignment
({{scan.watermark.alignment.group}}) on both sources and
{{table.exec.source.idle-timeout}}.
* A large build-side backlog and a probe-side start offset ({{/\*+
OPTIONS('scan.startup.mode'=...) \*/}}) to see the amount of buffered probe
rows and the output spike at the flip.
* Failure recovery in the join phase (kill a TaskManager). Results after
recovery may differ from before, but must not contain half-applied update pairs.
* NULL join keys, composite keys, a view or CTE as build side, and
{{state_ttl}} without an equality predicate (the whole build side is evicted at
once when idle).
* RocksDB state backend ({{state.backend.type: rocksdb}}) for sections 1, 3 and
7.
h3. Reporting
Comment with the commit or RC and give pass or fail per section. File new bugs
as linked issues with Fix Version 2.4.0.
was:
[FLIP-579|https://cwiki.apache.org/confluence/display/FLINK/FLIP-579%3A+LATERAL+SNAPSHOT+Join]
adds {{LATERAL SNAPSHOT}} join, a stream enrichment join. It joins an
append-only probe table with the *current* state of an updating build table. A
load phase first fills the build-side state and buffers probe rows. Afterwards,
every probe row is joined immediately, without waiting for build-side
watermarks. Neither side needs a primary key. The result is non-deterministic
in streaming mode and deterministic in batch mode.
Docs: [LATERAL SNAPSHOT
Join|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/sql/reference/queries/joins/#lateral-snapshot-join]
and [Lateral Snapshot Join (Table
API)|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/dev/table/tableapi/#lateral-snapshot-join].
Planner, operator and restore tests are covered by unit and IT tests. This
ticket is about the end-to-end path on a real cluster: phases, metrics,
savepoints, batch mode and the Table API.
h3. 0. Setup
Build {{release-2.4}}, or use the 2.4.0 RC.
{code:bash}
./mvnw clean install -DskipTests -Dfast -Pskip-webui-build -T1C
cd build-target
mkdir -p /tmp/flip579/build /tmp/flip579/savepoints /tmp/flip579/checkpoints
{code}
In {{conf/config.yaml}}, change {{numberOfTaskSlots: 1}} to {{4}}, and append:
{code:yaml}
execution.checkpointing.dir: file:///tmp/flip579/checkpoints
execution.checkpointing.savepoint-dir: file:///tmp/flip579/savepoints
execution.checkpointing.interval: 10s
{code}
Start the cluster with {{./bin/start-cluster.sh}}. Save the following as
{{/tmp/flip579/ddl.sql}} and start the SQL Client with {{./bin/sql-client.sh -i
/tmp/flip579/ddl.sql}}. Run every query in the sections below in that session.
Queries that start with {{INSERT INTO}} run detached on the cluster, and you
can see their result in {{log/flink-<user>-taskexecutor-<n>-<host>.out}} (the
{{print}} sink). Use {{tail -f}} on it, and do not delete or truncate the file
while the TaskManager is running (output is lost until the cluster restarts).
Note the line count instead.
The SQL Client only loads {{ddl.sql}} via {{-i}} on start, and the tables and
views created in section 4 live only in the session. If you restart the client,
pass {{-i}} again and recreate them.
The build side is a directory of {{debezium-json}} files that the filesystem
source monitors. A file you drop into the directory is read as build-side
changes, including updates and deletes. The probe side is a datagen source with
one order per second and the currency cycling A, B, C.
{code:sql}
SET 'table.local-time-zone' = 'UTC';
SET 'sql-client.execution.result-mode' = 'tableau';
CREATE TABLE rates (
currency STRING, rate DOUBLE, update_time TIMESTAMP(3),
WATERMARK FOR update_time AS update_time
) WITH (
'connector' = 'filesystem', 'path' = '/tmp/flip579/build',
'format' = 'debezium-json', 'source.monitor-interval' = '2s');
CREATE TABLE orders (
order_id BIGINT,
currency AS CASE order_id % 3 WHEN 0 THEN 'A' WHEN 1 THEN 'B' ELSE 'C' END,
ts AS LOCALTIMESTAMP,
WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
) WITH ('connector' = 'datagen', 'rows-per-second' = '1',
'fields.order_id.kind' = 'sequence', 'fields.order_id.start' = '1',
'fields.order_id.end' = '100000');
CREATE TABLE out_print (order_id BIGINT, currency STRING, rate DOUBLE) WITH
('connector' = 'print');
{code}
{{SET}} statements stay active for the rest of the SQL Client session. The
sections below tell you when to {{RESET}} a setting. Alternatively, restart the
SQL Client between sections.
The operator metrics are visible in the Web UI ({{http://localhost:8081}}):
open the job, click the {{LateralSnapshotJoin}} vertex and go to the Metrics
tab (the metric names are prefixed with {{LateralSnapshotJoin[n].}}). The same
values are available via REST, e.g. {{GET
/jobs/<job-id>/vertices/<vertex-id>/subtasks/metrics?get=LateralSnapshotJoin[9].currentPhase}}.
The operator number may differ. Metric values refresh only about every 10
seconds, so a check such as "immediately" or "after a few seconds" can look
stale for that long.
*Build-side changes are applied when the build-side watermark passes their row
time* (they are buffered until then, see the FLIP). A build row is therefore
invisible to the join until a later build-side watermark arrives. All build
files below end with a row for a dummy key {{Z}} with a later {{update_time}}
for this reason. It pushes the watermark, and no probe row matches it. Without
it, the last rows of a file stay invisible. In real pipelines the next
build-side record does this.
h3. 1. Load phase and flip via idle timeout
Start with an empty {{/tmp/flip579/build}} and submit:
{code:sql}
INSERT INTO out_print
SELECT o.order_id, o.currency, r.rate
FROM orders AS o
JOIN LATERAL SNAPSHOT(
input => TABLE rates,
on_time => DESCRIPTOR(update_time),
load_completed_idle_timeout => INTERVAL '10' SECOND) AS r
ON o.currency = r.currency;
{code}
The probe side runs, but nothing is emitted yet. Then write the initial build
state:
{code:bash}
cat > /tmp/flip579/build/001.json <<'EOF'
{"before":null,"after":{"currency":"A","rate":1.0,"update_time":"2026-01-01
10:00:00"},"op":"c"}
{"before":null,"after":{"currency":"B","rate":2.0,"update_time":"2026-01-01
10:00:00"},"op":"c"}
{"before":null,"after":{"currency":"Z","rate":0.0,"update_time":"2026-01-01
11:00:00"},"op":"c"}
EOF
{code}
Expected:
* Before the file is written, {{currentPhase}} = 0 and
{{numProbeSideRecordsBuffered}} grows by about 1 per second. Nothing is printed.
* About 10 seconds after the last build-side watermark advance, the join flips:
{{currentPhase}} = 1 and {{numProbeSideRecordsBuffered}} drops to 0.
* All buffered orders are printed at once: A orders with rate 1.0, B orders
with rate 2.0. C orders are not printed (inner join), and
{{numUnmatchedProbeRecords}} counts them.
* New orders are printed immediately afterwards.
* {{currentBuildSideWatermark}} = {{1767265200000}} ({{2026-01-01 11:00:00}}
UTC). {{currentProbeSideWatermark}} is close to the current time in epoch
milliseconds. {{maxJoinFanOut}} = 1.
h3. 2. Stuck in the load phase
Cancel the job and empty the build directory ({{rm /tmp/flip579/build/*}}).
Submit the same query *without* {{load_completed_idle_timeout}}, and drop
{{001.json}} again. The default {{load_completed_time}} is the wall-clock time
at query compilation, so the build-side watermark (January 2026) never reaches
it.
Expected: {{currentPhase}} stays 0, {{numProbeSideRecordsBuffered}} keeps
growing, there is no output, and {{currentBuildSideWatermark}} stays at
{{1767265200000}}.
Then fix it with an explicit load completion time. Cancel the job, empty the
build directory, and submit:
{code:sql}
INSERT INTO out_print
SELECT o.order_id, o.currency, r.rate
FROM orders AS o
JOIN LATERAL SNAPSHOT(input => TABLE rates, on_time => DESCRIPTOR(update_time),
load_completed_time => CAST(TIMESTAMP '2026-01-02 00:00:00' AS
TIMESTAMP_LTZ(3))) AS r
ON o.currency = r.currency;
{code}
Drop {{001.json}} (its dummy {{Z}} row moves the watermark to 2026-01-01 11:00,
which is below the load completion time). The join stays in the load phase.
Then drop a file that moves the watermark past the load completion time:
{code:bash}
cat > /tmp/flip579/build/002.json <<'EOF'
{"before":null,"after":{"currency":"Z","rate":0.0,"update_time":"2026-01-02
00:00:00"},"op":"c"}
EOF
{code}
Expected: the join flips as soon as the watermark reaches 2026-01-02 00:00:00
({{currentBuildSideWatermark}} = {{1767312000000}}), {{currentPhase}} = 1,
buffered orders are printed.
h3. 3. Build-side updates, deletes and LEFT JOIN
Cancel the job, empty the build directory and submit the section 1 query as
{{LEFT JOIN LATERAL SNAPSHOT(...)}} (still with the idle timeout). Drop
{{001.json}} from section 1 and wait for the flip. A orders have rate 1.0, B
orders 2.0, and C orders are printed with a {{NULL}} rate. Then drop:
{code:bash}
cat > /tmp/flip579/build/003.json <<'EOF'
{"before":{"currency":"A","rate":1.0,"update_time":"2026-01-01
10:00:00"},"after":{"currency":"A","rate":1.5,"update_time":"2026-02-01
10:00:00"},"op":"u"}
{"before":{"currency":"B","rate":2.0,"update_time":"2026-01-01
10:00:00"},"after":null,"op":"d"}
{"before":null,"after":{"currency":"C","rate":3.0,"update_time":"2026-02-01
10:00:00"},"op":"c"}
{"before":null,"after":{"currency":"Z","rate":0.0,"update_time":"2026-02-02
10:00:00"},"op":"c"}
EOF
{code}
Expected after a few seconds:
* New A orders have rate 1.5. Rate 1.0 is never mixed with 1.5 in an
inconsistent way: the update becomes visible atomically.
* New B orders have a {{NULL}} rate (the row was deleted).
* New C orders have rate 3.0.
* Already printed rows are never changed. The result is append-only.
* {{numUnmatchedBuildRetractions}} stays 0, because all retractions hit
existing rows.
* Note: {{numUnmatchedProbeRecords}} stays 0 for a LEFT JOIN, although
unmatched rows are emitted null-padded. It only counts unmatched rows of INNER
joins (the docs describe it more generally).
h3. 4. Planning-time validation
Create the extra tables and views (add to the SQL Client session):
{code:sql}
CREATE TABLE rates_nowm (currency STRING, rate DOUBLE, update_time TIMESTAMP(3))
WITH
('connector'='filesystem','path'='/tmp/flip579/build','format'='debezium-json','source.monitor-interval'='2s');
CREATE VIEW rates_rand AS SELECT currency, rate, RAND() AS r, update_time FROM
rates;
CREATE VIEW probe_agg AS SELECT currency, COUNT(*) AS c FROM orders GROUP BY
currency;
{code}
Each of the following statements must be rejected with the quoted error. The
full statement is {{EXPLAIN SELECT o.order_id FROM orders AS o <fragment> ON
o.currency = r.currency;}}, with the {{<fragment>}} from the first column (in
the {{probe_agg}} row the fragment replaces {{orders AS o}} as well, and the
standalone {{SNAPSHOT}} row is a complete statement without {{ON}}). For
{{RIGHT}}/{{FULL}} use the arguments {{(input => TABLE rates, on_time =>
DESCRIPTOR(update_time))}}. A failing statement aborts a {{-f}} script, so
paste the statements one at a time into the interactive client.
|| Statement || Expected error ||
| {{JOIN LATERAL SNAPSHOT(input => TABLE rates_nowm, on_time =>
DESCRIPTOR(update_time)) AS r}} | {{Argument 'on_time' of SNAPSHOT must
reference a rowtime attribute (a column with a watermark), but column
'update_time' is not one.}} |
| {{JOIN LATERAL SNAPSHOT(input => TABLE rates) AS r}} (no {{on_time}}) |
{{LATERAL SNAPSHOT requires the 'on_time' argument to identify the build-side
rowtime attribute.}} |
| {{on_time => DESCRIPTOR(rate)}} | {{Argument 'on_time' of SNAPSHOT must
reference a TIMESTAMP or TIMESTAMP_LTZ column (up to precision 3), but column
'rate' has type 'DOUBLE'.}} |
| {{FROM probe_agg AS o JOIN LATERAL SNAPSHOT(input => TABLE rates, on_time =>
DESCRIPTOR(update_time)) AS r}} | {{The probe (left) input of LATERAL SNAPSHOT
join doesn't support consuming update changes ...}} |
| {{RIGHT JOIN LATERAL SNAPSHOT(...)}} / {{FULL JOIN LATERAL SNAPSHOT(...)}} |
{{LATERAL SNAPSHOT join only supports INNER JOIN and LEFT OUTER JOIN, but was
RIGHT JOIN.}} (and {{FULL JOIN}}) |
| {{SELECT * FROM TABLE(SNAPSHOT(input => TABLE rates, on_time =>
DESCRIPTOR(update_time)))}} | {{The SNAPSHOT function can only be used as the
build side (right-hand side) of a LATERAL join. ...}} |
Non-deterministic build columns (FLINK-40890). The check is part of the
non-deterministic update handling, so enable it first:
{code:sql}
SET 'table.optimizer.non-deterministic-update.strategy' = 'TRY_RESOLVE';
EXPLAIN SELECT o.order_id, r.rate, r.r FROM orders AS o
JOIN LATERAL SNAPSHOT(input => TABLE rates_rand, on_time =>
DESCRIPTOR(update_time)) AS r ON o.currency = r.currency;
-- TableException: The column(s): r(generated by non-deterministic function:
RAND ) can not satisfy the determinism requirement ...
RESET 'table.optimizer.non-deterministic-update.strategy';
{code}
The same statement is accepted with the default strategy ({{IGNORE}}), and also
with {{TRY_RESOLVE}} if the non-deterministic column is not selected.
The {{TABLE}} keyword in {{LATERAL}} calls is optional. Verify that {{JOIN
LATERAL SNAPSHOT(...)}}, {{JOIN LATERAL TABLE(SNAPSHOT(...))}} and {{CROSS JOIN
LATERAL SNAPSHOT(...) ... WHERE o.currency = r.currency}} all plan, and that
{{EXPLAIN}} shows a {{LateralSnapshotJoin}} node.
h3. 5. Join predicates and parallelism
Run the section 1 query as an {{INSERT INTO}} with {{SET 'parallelism.default'
= '2';}} and different {{ON}} clauses, and check the parallelism of the
{{LateralSnapshotJoin}} vertex in the Web UI. At parallelism 2, one build-side
source subtask never reads data, so aggregated values of
{{currentBuildSideWatermark}} can show Long.MIN_VALUE. This is expected, so
look at the metric per subtask. Cancel each job before submitting the next one,
and run {{RESET 'parallelism.default';}} when you are done.
* {{ON o.currency = r.currency AND o.order_id >= 0}} (equi and non-equi mixed):
the join runs with parallelism 2.
* {{ON o.currency <> r.currency}} (no equality predicate): the join runs with
parallelism 1, while the sources run with 2. After flipping, each order is
joined with all non-matching currencies, so {{maxJoinFanOut}} > 1 and
{{avgJoinFanOut}} shows the average (the dummy {{Z}} row also matches).
h3. 6. State TTL
Run the section 1 query with an additional {{state_ttl => INTERVAL '20'
SECOND}}, and drop {{001.json}}. After the flip and the TTL ({{t ~ 40-50 s}}):
* {{numStateTtlEvictions}} = 1. The key {{Z}} was never accessed by a probe row
and is evicted.
* The keys {{A}} and {{B}} are accessed by probe rows all the time, so they are
kept and orders are still matched with their rates.
* {{numStateTtlEvictions}} stays 0 during the load phase.
h3. 7. Savepoint and restore
*Join phase.* Run the job from section 6 (or section 1) until {{currentPhase}}
= 1, then:
{code:bash}
./bin/flink stop <job-id>
# prints: Savepoint completed. Path: file:/tmp/flip579/savepoints/savepoint-...
{code}
In the SQL Client, run {{SET 'execution.state-recovery.path' = '<savepoint
path>';}} and submit the same {{INSERT INTO}} again. Then run {{RESET
'execution.state-recovery.path';}}, otherwise every later submission in this
session restores from that savepoint. Expected:
* The join is in the join phase right away ({{currentPhase}} = 1), without any
new build-side data.
* A and B orders are still matched. The build-side state was restored.
* The metric values are not restored. Counters start at 0, and
{{currentBuildSideWatermark}} is {{-9.2E18}} (Long.MIN_VALUE) until the next
build-side watermark arrives.
*Load phase.* Cancel the job and empty the build directory. Submit the second
query of section 2 (with {{load_completed_time}}) without build data. Let it
buffer about 30 probe rows, then stop it with {{./bin/flink stop}}. Restore it
with {{execution.state-recovery.path}} (and {{RESET}} it afterwards), and
verify that {{currentPhase}} = 0 after the restore. Then drop {{001.json}} with
the {{Z}} row at {{2026-01-03 00:00:00}}. Expected: the join flips, and the
output contains the orders that were buffered *before* the savepoint
({{order_id}} starting at 1, with rates for A and B). The
{{numProbeSideRecordsBuffered}} gauge only counts rows seen since the restore,
so check the output instead.
h3. 8. Batch mode
{code:bash}
mkdir -p /tmp/flip579/orders_b /tmp/flip579/rates_b
printf '1,A\n2,B\n3,C\n4,A\n' > /tmp/flip579/orders_b/o.csv
printf 'A,1.0\nB,2.0\nA,1.5\n' > /tmp/flip579/rates_b/r.csv
{code}
{code:sql}
SET 'execution.runtime-mode' = 'batch';
CREATE TABLE orders_b (order_id BIGINT, currency STRING) WITH ('connector' =
'filesystem', 'path' = '/tmp/flip579/orders_b', 'format' = 'csv');
CREATE TABLE rates_b (currency STRING, rate DOUBLE) WITH ('connector' =
'filesystem', 'path' = '/tmp/flip579/rates_b', 'format' = 'csv');
SELECT o.order_id, o.currency, r.rate
FROM orders_b AS o
LEFT JOIN LATERAL SNAPSHOT(input => TABLE rates_b, state_ttl => INTERVAL '1'
HOUR) AS r
ON o.currency = r.currency ORDER BY o.order_id, r.rate;
RESET 'execution.runtime-mode';
{code}
Expected: the query finishes without watermarks and without {{on_time}}.
{{state_ttl}} is accepted and ignored. The result is the regular left join
against the complete build side (the SQL Client renders it as a table and
prints {{<NULL>}}; the rows are the same):
{noformat}
1 | A | 1.0
1 | A | 1.5
2 | B | 2.0
3 | C | NULL
4 | A | 1.0
4 | A | 1.5
{noformat}
Repeated runs return the same result. {{EXPLAIN}} of the inner join variant
shows a regular join ({{HashJoin}}/{{AdaptiveJoin}}) and no
{{LateralSnapshotJoin}}.
h3. 9. Table API
Save the following as {{LsjTableApi.java}} in {{build-target}} and run it from
there with {{java -cp "lib/*" LsjTableApi.java}} from {{build-target}} (Java 17
single-file launch). The columns of the two tables must have different names,
hence the {{r_}} prefix.
{code:java}
import static org.apache.flink.table.api.Expressions.*;
import org.apache.flink.table.api.*;
public class LsjTableApi {
public static void main(String[] args) throws Exception {
TableEnvironment env =
TableEnvironment.create(EnvironmentSettings.inStreamingMode());
env.executeSql(
"CREATE TABLE rates (r_currency STRING, r_rate DOUBLE,
update_time TIMESTAMP(3),"
+ " WATERMARK FOR update_time AS update_time) WITH ("
+
"'connector'='filesystem','path'='/tmp/flip579/build','format'='debezium-json')");
env.executeSql(
"CREATE TABLE orders (order_id BIGINT, currency STRING, ts AS
LOCALTIMESTAMP,"
+ " WATERMARK FOR ts AS ts - INTERVAL '1' SECOND) WITH
('connector'='datagen')");
Table orders = env.from("orders");
Table rates = env.from("rates");
Table inner = orders.joinLateral(
call("SNAPSHOT", rates.asArgument("input"),
descriptor("update_time").asArgument("on_time")),
$("currency").isEqual($("r_currency")));
System.out.println(inner.explain());
Table left = orders.leftOuterJoinLateral(
call("SNAPSHOT", rates.asArgument("input"),
descriptor("update_time").asArgument("on_time")),
$("currency").isEqual($("r_currency")));
System.out.println(left.explain());
}
}
{code}
Expected: both plans contain a {{LateralSnapshotJoin(joinType=[InnerJoin])}}
and {{LateralSnapshotJoin(joinType=[LeftOuterJoin])}} node respectively.
Optionally replace {{explain()}} with {{executeInsert}} into a {{print}} table
and compare the behavior to sections 1 and 3 (the Table API job runs in a local
MiniCluster, so there is no Web UI).
h3. Exploratory
* Real sources: a probe side from Kafka and a build side from an
{{upsert-kafka}} topic, with watermark alignment
({{scan.watermark.alignment.group}}) on both sources and
{{table.exec.source.idle-timeout}}.
* A large build-side backlog and a probe-side start offset ({{/*+
OPTIONS('scan.startup.mode'=...) */}}) to see the amount of buffered probe rows
and the output spike at the flip.
* Failure recovery in the join phase (kill a TaskManager). Results after
recovery may differ from before, but must not contain half-applied update pairs.
* NULL join keys, composite keys, a view or CTE as build side, and
{{state_ttl}} without an equality predicate (the whole build side is evicted at
once when idle).
* RocksDB state backend ({{state.backend.type: rocksdb}}) for sections 1, 3 and
7.
h3. Reporting
Comment with the commit or RC and give pass or fail per section. File new bugs
as linked issues with Fix Version 2.4.0.
> Release Testing: Verify FLIP-579: LATERAL SNAPSHOT Join
> -------------------------------------------------------
>
> Key: FLINK-41368
> URL: https://issues.apache.org/jira/browse/FLINK-41368
> Project: Flink
> Issue Type: Sub-task
> Components: Release System, Tests
> Affects Versions: 2.4.0
> Reporter: Fabian Hueske
> Priority: Major
> Fix For: 2.4.0
>
>
> [FLIP-579|https://cwiki.apache.org/confluence/display/FLINK/FLIP-579%3A+LATERAL+SNAPSHOT+Join]
> adds {{LATERAL SNAPSHOT}} join, a stream enrichment join. It joins an
> append-only probe table with the *current* state of an updating build table.
> A load phase first fills the build-side state and buffers probe rows.
> Afterwards, every probe row is joined immediately, without waiting for
> build-side watermarks. Neither side needs a primary key. The result is
> non-deterministic in streaming mode and deterministic in batch mode.
> Docs: [LATERAL SNAPSHOT
> Join|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/sql/reference/queries/joins/#lateral-snapshot-join]
> and [Lateral Snapshot Join (Table
> API)|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/dev/table/tableapi/#lateral-snapshot-join].
> Planner, operator and restore tests are covered by unit and IT tests. This
> ticket is about the end-to-end path on a real cluster: phases, metrics,
> savepoints, batch mode and the Table API.
> h3. 0. Setup
> Build {{release-2.4}}, or use the 2.4.0 RC.
> {code:bash}
> ./mvnw clean install -DskipTests -Dfast -Pskip-webui-build -T1C
> cd build-target
> mkdir -p /tmp/flip579/build /tmp/flip579/savepoints /tmp/flip579/checkpoints
> {code}
> In {{conf/config.yaml}}, change {{numberOfTaskSlots: 1}} to {{4}}, and append:
> {code:yaml}
> execution.checkpointing.dir: file:///tmp/flip579/checkpoints
> execution.checkpointing.savepoint-dir: file:///tmp/flip579/savepoints
> execution.checkpointing.interval: 10s
> {code}
> Start the cluster with {{./bin/start-cluster.sh}}. Save the following as
> {{/tmp/flip579/ddl.sql}} and start the SQL Client with {{./bin/sql-client.sh
> \-i /tmp/flip579/ddl.sql}}. Run every query in the sections below in that
> session. Queries that start with {{INSERT INTO}} run detached on the cluster,
> and you can see their result in
> {{log/flink-<user>-taskexecutor-<n>-<host>.out}} (the {{print}} sink). Use
> {{tail \-f}} on it, and do not delete or truncate the file while the
> TaskManager is running (output is lost until the cluster restarts). Note the
> line count instead.
> The SQL Client only loads {{ddl.sql}} via {{\-i}} on start, and the tables
> and views created in section 4 live only in the session. If you restart the
> client, pass {{\-i}} again and recreate them.
> The build side is a directory of {{debezium-json}} files that the filesystem
> source monitors. A file you drop into the directory is read as build-side
> changes, including updates and deletes. The probe side is a datagen source
> with one order per second and the currency cycling A, B, C.
> {code:sql}
> SET 'table.local-time-zone' = 'UTC';
> SET 'sql-client.execution.result-mode' = 'tableau';
> CREATE TABLE rates (
> currency STRING, rate DOUBLE, update_time TIMESTAMP(3),
> WATERMARK FOR update_time AS update_time
> ) WITH (
> 'connector' = 'filesystem', 'path' = '/tmp/flip579/build',
> 'format' = 'debezium-json', 'source.monitor-interval' = '2s');
> CREATE TABLE orders (
> order_id BIGINT,
> currency AS CASE order_id % 3 WHEN 0 THEN 'A' WHEN 1 THEN 'B' ELSE 'C' END,
> ts AS LOCALTIMESTAMP,
> WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
> ) WITH ('connector' = 'datagen', 'rows-per-second' = '1',
> 'fields.order_id.kind' = 'sequence', 'fields.order_id.start' = '1',
> 'fields.order_id.end' = '100000');
> CREATE TABLE out_print (order_id BIGINT, currency STRING, rate DOUBLE) WITH
> ('connector' = 'print');
> {code}
> {{SET}} statements stay active for the rest of the SQL Client session. The
> sections below tell you when to {{RESET}} a setting. Alternatively, restart
> the SQL Client between sections.
> The operator metrics are visible in the Web UI ({{http://localhost:8081}}):
> open the job, click the {{LateralSnapshotJoin}} vertex and go to the Metrics
> tab (the metric names are prefixed with {{LateralSnapshotJoin[n].}}). The
> same values are available via REST, e.g. {{GET
> /jobs/<job-id>/vertices/<vertex-id>/subtasks/metrics?get=LateralSnapshotJoin[9].currentPhase}}.
> The operator number may differ. Metric values refresh only about every 10
> seconds, so a check such as "immediately" or "after a few seconds" can look
> stale for that long.
> *Build-side changes are applied when the build-side watermark passes their
> row time* (they are buffered until then, see the FLIP). A build row is
> therefore invisible to the join until a later build-side watermark arrives.
> All build files below end with a row for a dummy key {{Z}} with a later
> {{update_time}} for this reason. It pushes the watermark, and no probe row
> matches it. Without it, the last rows of a file stay invisible. In real
> pipelines the next build-side record does this.
> h3. 1. Load phase and flip via idle timeout
> Start with an empty {{/tmp/flip579/build}} and submit:
> {code:sql}
> INSERT INTO out_print
> SELECT o.order_id, o.currency, r.rate
> FROM orders AS o
> JOIN LATERAL SNAPSHOT(
> input => TABLE rates,
> on_time => DESCRIPTOR(update_time),
> load_completed_idle_timeout => INTERVAL '10' SECOND) AS r
> ON o.currency = r.currency;
> {code}
> The probe side runs, but nothing is emitted yet. Then write the initial build
> state:
> {code:bash}
> cat > /tmp/flip579/build/001.json <<'EOF'
> {"before":null,"after":{"currency":"A","rate":1.0,"update_time":"2026-01-01
> 10:00:00"},"op":"c"}
> {"before":null,"after":{"currency":"B","rate":2.0,"update_time":"2026-01-01
> 10:00:00"},"op":"c"}
> {"before":null,"after":{"currency":"Z","rate":0.0,"update_time":"2026-01-01
> 11:00:00"},"op":"c"}
> EOF
> {code}
> Expected:
> * Before the file is written, {{currentPhase}} = 0 and
> {{numProbeSideRecordsBuffered}} grows by about 1 per second. Nothing is
> printed.
> * About 10 seconds after the last build-side watermark advance, the join
> flips: {{currentPhase}} = 1 and {{numProbeSideRecordsBuffered}} drops to 0.
> * All buffered orders are printed at once: A orders with rate 1.0, B orders
> with rate 2.0. C orders are not printed (inner join), and
> {{numUnmatchedProbeRecords}} counts them.
> * New orders are printed immediately afterwards.
> * {{currentBuildSideWatermark}} = {{1767265200000}} ({{2026-01-01 11:00:00}}
> UTC). {{currentProbeSideWatermark}} is close to the current time in epoch
> milliseconds. {{maxJoinFanOut}} = 1.
> h3. 2. Stuck in the load phase
> Cancel the job and empty the build directory ({{rm /tmp/flip579/build/\*}}).
> Submit the same query *without* {{load_completed_idle_timeout}}, and drop
> {{001.json}} again. The default {{load_completed_time}} is the wall-clock
> time at query compilation, so the build-side watermark (January 2026) never
> reaches it.
> Expected: {{currentPhase}} stays 0, {{numProbeSideRecordsBuffered}} keeps
> growing, there is no output, and {{currentBuildSideWatermark}} stays at
> {{1767265200000}}.
> Then fix it with an explicit load completion time. Cancel the job, empty the
> build directory, and submit:
> {code:sql}
> INSERT INTO out_print
> SELECT o.order_id, o.currency, r.rate
> FROM orders AS o
> JOIN LATERAL SNAPSHOT(input => TABLE rates, on_time =>
> DESCRIPTOR(update_time),
> load_completed_time => CAST(TIMESTAMP '2026-01-02 00:00:00' AS
> TIMESTAMP_LTZ(3))) AS r
> ON o.currency = r.currency;
> {code}
> Drop {{001.json}} (its dummy {{Z}} row moves the watermark to 2026-01-01
> 11:00, which is below the load completion time). The join stays in the load
> phase. Then drop a file that moves the watermark past the load completion
> time:
> {code:bash}
> cat > /tmp/flip579/build/002.json <<'EOF'
> {"before":null,"after":{"currency":"Z","rate":0.0,"update_time":"2026-01-02
> 00:00:00"},"op":"c"}
> EOF
> {code}
> Expected: the join flips as soon as the watermark reaches 2026-01-02 00:00:00
> ({{currentBuildSideWatermark}} = {{1767312000000}}), {{currentPhase}} = 1,
> buffered orders are printed.
> h3. 3. Build-side updates, deletes and LEFT JOIN
> Cancel the job, empty the build directory and submit the section 1 query as
> {{LEFT JOIN LATERAL SNAPSHOT(...)}} (still with the idle timeout). Drop
> {{001.json}} from section 1 and wait for the flip. A orders have rate 1.0, B
> orders 2.0, and C orders are printed with a {{NULL}} rate. Then drop:
> {code:bash}
> cat > /tmp/flip579/build/003.json <<'EOF'
> {"before":{"currency":"A","rate":1.0,"update_time":"2026-01-01
> 10:00:00"},"after":{"currency":"A","rate":1.5,"update_time":"2026-02-01
> 10:00:00"},"op":"u"}
> {"before":{"currency":"B","rate":2.0,"update_time":"2026-01-01
> 10:00:00"},"after":null,"op":"d"}
> {"before":null,"after":{"currency":"C","rate":3.0,"update_time":"2026-02-01
> 10:00:00"},"op":"c"}
> {"before":null,"after":{"currency":"Z","rate":0.0,"update_time":"2026-02-02
> 10:00:00"},"op":"c"}
> EOF
> {code}
> Expected after a few seconds:
> * New A orders have rate 1.5. Rate 1.0 is never mixed with 1.5 in an
> inconsistent way: the update becomes visible atomically.
> * New B orders have a {{NULL}} rate (the row was deleted).
> * New C orders have rate 3.0.
> * Already printed rows are never changed. The result is append-only.
> * {{numUnmatchedBuildRetractions}} stays 0, because all retractions hit
> existing rows.
> * Note: {{numUnmatchedProbeRecords}} stays 0 for a LEFT JOIN, although
> unmatched rows are emitted null-padded. It only counts unmatched rows of
> INNER joins (the docs describe it more generally).
> h3. 4. Planning-time validation
> Create the extra tables and views (add to the SQL Client session):
> {code:sql}
> CREATE TABLE rates_nowm (currency STRING, rate DOUBLE, update_time
> TIMESTAMP(3))
> WITH
> ('connector'='filesystem','path'='/tmp/flip579/build','format'='debezium-json','source.monitor-interval'='2s');
> CREATE VIEW rates_rand AS SELECT currency, rate, RAND() AS r, update_time
> FROM rates;
> CREATE VIEW probe_agg AS SELECT currency, COUNT(*) AS c FROM orders GROUP BY
> currency;
> {code}
> Each of the following statements must be rejected with the quoted error. The
> full statement is {{EXPLAIN SELECT o.order_id FROM orders AS o <fragment> ON
> o.currency = r.currency;}}, with the {{<fragment>}} from the first column (in
> the {{probe_agg}} row the fragment replaces {{orders AS o}} as well, and the
> standalone {{SNAPSHOT}} row is a complete statement without {{ON}}). For
> {{RIGHT}}/{{FULL}} use the arguments {{(input => TABLE rates, on_time =>
> DESCRIPTOR(update_time))}}. A failing statement aborts a {{\-f}} script, so
> paste the statements one at a time into the interactive client.
> || Statement || Expected error ||
> | {{JOIN LATERAL SNAPSHOT(input => TABLE rates_nowm, on_time =>
> DESCRIPTOR(update_time)) AS r}} | {{Argument 'on_time' of SNAPSHOT must
> reference a rowtime attribute (a column with a watermark), but column
> 'update_time' is not one.}} |
> | {{JOIN LATERAL SNAPSHOT(input => TABLE rates) AS r}} (no {{on_time}}) |
> {{LATERAL SNAPSHOT requires the 'on_time' argument to identify the build-side
> rowtime attribute.}} |
> | {{on_time => DESCRIPTOR(rate)}} | {{Argument 'on_time' of SNAPSHOT must
> reference a TIMESTAMP or TIMESTAMP_LTZ column (up to precision 3), but column
> 'rate' has type 'DOUBLE'.}} |
> | {{FROM probe_agg AS o JOIN LATERAL SNAPSHOT(input => TABLE rates, on_time
> => DESCRIPTOR(update_time)) AS r}} | {{The probe (left) input of LATERAL
> SNAPSHOT join doesn't support consuming update changes ...}} |
> | {{RIGHT JOIN LATERAL SNAPSHOT(...)}} / {{FULL JOIN LATERAL SNAPSHOT(...)}}
> | {{LATERAL SNAPSHOT join only supports INNER JOIN and LEFT OUTER JOIN, but
> was RIGHT JOIN.}} (and {{FULL JOIN}}) |
> | {{SELECT * FROM TABLE(SNAPSHOT(input => TABLE rates, on_time =>
> DESCRIPTOR(update_time)))}} | {{The SNAPSHOT function can only be used as the
> build side (right-hand side) of a LATERAL join. ...}} |
> Non-deterministic build columns (FLINK-40890). The check is part of the
> non-deterministic update handling, so enable it first:
> {code:sql}
> SET 'table.optimizer.non-deterministic-update.strategy' = 'TRY_RESOLVE';
> EXPLAIN SELECT o.order_id, r.rate, r.r FROM orders AS o
> JOIN LATERAL SNAPSHOT(input => TABLE rates_rand, on_time =>
> DESCRIPTOR(update_time)) AS r ON o.currency = r.currency;
> -- TableException: The column(s): r(generated by non-deterministic function:
> RAND ) can not satisfy the determinism requirement ...
> RESET 'table.optimizer.non-deterministic-update.strategy';
> {code}
> The same statement is accepted with the default strategy ({{IGNORE}}), and
> also with {{TRY_RESOLVE}} if the non-deterministic column is not selected.
> The {{TABLE}} keyword in {{LATERAL}} calls is optional. Verify that {{JOIN
> LATERAL SNAPSHOT(...)}}, {{JOIN LATERAL TABLE(SNAPSHOT(...))}} and {{CROSS
> JOIN LATERAL SNAPSHOT(...) ... WHERE o.currency = r.currency}} all plan, and
> that {{EXPLAIN}} shows a {{LateralSnapshotJoin}} node.
> h3. 5. Join predicates and parallelism
> Run the section 1 query as an {{INSERT INTO}} with {{SET
> 'parallelism.default' = '2';}} and different {{ON}} clauses, and check the
> parallelism of the {{LateralSnapshotJoin}} vertex in the Web UI. At
> parallelism 2, one build-side source subtask never reads data, so aggregated
> values of {{currentBuildSideWatermark}} can show Long.MIN_VALUE. This is
> expected, so look at the metric per subtask. Cancel each job before
> submitting the next one, and run {{RESET 'parallelism.default';}} when you
> are done.
> * {{ON o.currency = r.currency AND o.order_id >= 0}} (equi and non-equi
> mixed): the join runs with parallelism 2.
> * {{ON o.currency <> r.currency}} (no equality predicate): the join runs with
> parallelism 1, while the sources run with 2. After flipping, each order is
> joined with all non-matching currencies, so {{maxJoinFanOut}} > 1 and
> {{avgJoinFanOut}} shows the average (the dummy {{Z}} row also matches).
> h3. 6. State TTL
> Run the section 1 query with an additional {{state_ttl => INTERVAL '20'
> SECOND}}, and drop {{001.json}}. After the flip and the TTL ({{t ~ 40-50 s}}):
> * {{numStateTtlEvictions}} = 1. The key {{Z}} was never accessed by a probe
> row and is evicted.
> * The keys {{A}} and {{B}} are accessed by probe rows all the time, so they
> are kept and orders are still matched with their rates.
> * {{numStateTtlEvictions}} stays 0 during the load phase.
> h3. 7. Savepoint and restore
> *Join phase.* Run the job from section 6 (or section 1) until
> {{currentPhase}} = 1, then:
> {code:bash}
> ./bin/flink stop <job-id>
> # prints: Savepoint completed. Path:
> file:/tmp/flip579/savepoints/savepoint-...
> {code}
> In the SQL Client, run {{SET 'execution.state-recovery.path' = '<savepoint
> path>';}} and submit the same {{INSERT INTO}} again. Then run {{RESET
> 'execution.state-recovery.path';}}, otherwise every later submission in this
> session restores from that savepoint. Expected:
> * The join is in the join phase right away ({{currentPhase}} = 1), without
> any new build-side data.
> * A and B orders are still matched. The build-side state was restored.
> * The metric values are not restored. Counters start at 0, and
> {{currentBuildSideWatermark}} is {{\-9.2E18}} (Long.MIN_VALUE) until the next
> build-side watermark arrives.
> *Load phase.* Cancel the job and empty the build directory. Submit the second
> query of section 2 (with {{load_completed_time}}) without build data. Let it
> buffer about 30 probe rows, then stop it with {{./bin/flink stop}}. Restore
> it with {{execution.state-recovery.path}} (and {{RESET}} it afterwards), and
> verify that {{currentPhase}} = 0 after the restore. Then drop {{001.json}}
> with the {{Z}} row at {{2026-01-03 00:00:00}}. Expected: the join flips, and
> the output contains the orders that were buffered *before* the savepoint
> ({{order_id}} starting at 1, with rates for A and B). The
> {{numProbeSideRecordsBuffered}} gauge only counts rows seen since the
> restore, so check the output instead.
> h3. 8. Batch mode
> {code:bash}
> mkdir -p /tmp/flip579/orders_b /tmp/flip579/rates_b
> printf '1,A\n2,B\n3,C\n4,A\n' > /tmp/flip579/orders_b/o.csv
> printf 'A,1.0\nB,2.0\nA,1.5\n' > /tmp/flip579/rates_b/r.csv
> {code}
> {code:sql}
> SET 'execution.runtime-mode' = 'batch';
> CREATE TABLE orders_b (order_id BIGINT, currency STRING) WITH ('connector' =
> 'filesystem', 'path' = '/tmp/flip579/orders_b', 'format' = 'csv');
> CREATE TABLE rates_b (currency STRING, rate DOUBLE) WITH ('connector' =
> 'filesystem', 'path' = '/tmp/flip579/rates_b', 'format' = 'csv');
> SELECT o.order_id, o.currency, r.rate
> FROM orders_b AS o
> LEFT JOIN LATERAL SNAPSHOT(input => TABLE rates_b, state_ttl => INTERVAL '1'
> HOUR) AS r
> ON o.currency = r.currency ORDER BY o.order_id, r.rate;
> RESET 'execution.runtime-mode';
> {code}
> Expected: the query finishes without watermarks and without {{on_time}}.
> {{state_ttl}} is accepted and ignored. The result is the regular left join
> against the complete build side (the SQL Client renders it as a table and
> prints {{<NULL>}}; the rows are the same):
> {noformat}
> 1 | A | 1.0
> 1 | A | 1.5
> 2 | B | 2.0
> 3 | C | NULL
> 4 | A | 1.0
> 4 | A | 1.5
> {noformat}
> Repeated runs return the same result. {{EXPLAIN}} of the inner join variant
> shows a regular join ({{HashJoin}}/{{AdaptiveJoin}}) and no
> {{LateralSnapshotJoin}}.
> h3. 9. Table API
> Save the following as {{LsjTableApi.java}} in {{build-target}} and run it
> from there with {{java -cp "lib/\*" LsjTableApi.java}} from {{build-target}}
> (Java 17 single-file launch). The columns of the two tables must have
> different names, hence the {{r_}} prefix.
> {code:java}
> import static org.apache.flink.table.api.Expressions.*;
> import org.apache.flink.table.api.*;
> public class LsjTableApi {
> public static void main(String[] args) throws Exception {
> TableEnvironment env =
> TableEnvironment.create(EnvironmentSettings.inStreamingMode());
> env.executeSql(
> "CREATE TABLE rates (r_currency STRING, r_rate DOUBLE,
> update_time TIMESTAMP(3),"
> + " WATERMARK FOR update_time AS update_time) WITH ("
> +
> "'connector'='filesystem','path'='/tmp/flip579/build','format'='debezium-json')");
> env.executeSql(
> "CREATE TABLE orders (order_id BIGINT, currency STRING, ts AS
> LOCALTIMESTAMP,"
> + " WATERMARK FOR ts AS ts - INTERVAL '1' SECOND)
> WITH ('connector'='datagen')");
> Table orders = env.from("orders");
> Table rates = env.from("rates");
> Table inner = orders.joinLateral(
> call("SNAPSHOT", rates.asArgument("input"),
> descriptor("update_time").asArgument("on_time")),
> $("currency").isEqual($("r_currency")));
> System.out.println(inner.explain());
> Table left = orders.leftOuterJoinLateral(
> call("SNAPSHOT", rates.asArgument("input"),
> descriptor("update_time").asArgument("on_time")),
> $("currency").isEqual($("r_currency")));
> System.out.println(left.explain());
> }
> }
> {code}
> Expected: both plans contain a {{LateralSnapshotJoin(joinType=[InnerJoin])}}
> and {{LateralSnapshotJoin(joinType=[LeftOuterJoin])}} node respectively.
> Optionally replace {{explain()}} with {{executeInsert}} into a {{print}}
> table and compare the behavior to sections 1 and 3 (the Table API job runs in
> a local MiniCluster, so there is no Web UI).
> h3. Exploratory
> * Real sources: a probe side from Kafka and a build side from an
> {{upsert-kafka}} topic, with watermark alignment
> ({{scan.watermark.alignment.group}}) on both sources and
> {{table.exec.source.idle-timeout}}.
> * A large build-side backlog and a probe-side start offset ({{/\*+
> OPTIONS('scan.startup.mode'=...) \*/}}) to see the amount of buffered probe
> rows and the output spike at the flip.
> * Failure recovery in the join phase (kill a TaskManager). Results after
> recovery may differ from before, but must not contain half-applied update
> pairs.
> * NULL join keys, composite keys, a view or CTE as build side, and
> {{state_ttl}} without an equality predicate (the whole build side is evicted
> at once when idle).
> * RocksDB state backend ({{state.backend.type: rocksdb}}) for sections 1, 3
> and 7.
> h3. Reporting
> Comment with the commit or RC and give pass or fail per section. File new
> bugs as linked issues with Fix Version 2.4.0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
