[
https://issues.apache.org/jira/browse/FLINK-41368?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Mika Naylor reassigned FLINK-41368:
-----------------------------------
Assignee: Chanhae Oh
> 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
> Assignee: Chanhae Oh
> 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)
