[
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-*-taskexecutor-*.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|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-*-taskexecutor-*.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|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)
