Weiqing Yang created FLINK-41451:
------------------------------------
Summary: Release Testing: Verify FLIP-497: Early Fire Support for
Flink SQL Interval Join
Key: FLINK-41451
URL: https://issues.apache.org/jira/browse/FLINK-41451
Project: Flink
Issue Type: Sub-task
Reporter: Weiqing Yang
Fix For: 2.4.0
[FLIP-497|https://cwiki.apache.org/confluence/display/FLINK/FLIP-497%3A+Early+Fire+Support+for+Flink+SQL+Interval+Join]
adds the {{EARLY_FIRE}} query hint for outer interval joins. Without it, an
unmatched outer row is emitted padded with NULLs only once its time window has
closed. With {{/*+ EARLY_FIRE('delay'='5s') */}}, that padded row is emitted
after the delay instead, and if a match arrives later inside the window it is
corrected with {{\-U}} (the padded row) followed by {{+U}} (the joined row).
Each outer row fires at most once. The join output becomes an updating stream,
so the sink must accept updates. {{time-mode}} chooses whether the delay is
measured against the watermark ({{rowtime}}) or wall-clock time ({{proctime}}).
Docs: [Early Fire for Outer Interval
Joins|https://nightlies.apache.org/flink/flink-docs-release-2.4/docs/sql/reference/queries/joins/#early-fire-for-outer-interval-joins].
Planner, operator and restore tests cover the code paths. This ticket is about
the end-to-end behavior on a real cluster: timing, corrections, sinks,
savepoints and compiled plans.
h3. Known issue (already filed, fix in progress)
This came up in pre-testing and is tracked for 2.4.0. There is no need to file
it again. If you see different behavior, comment on the ticket.
* FLINK-41446: in batch mode, any query with the hint, including an inner join,
fails at planning with {{Unknown join strategy : EARLY_FIRE}}. Without the hint
the same query plans and runs. Step 12 shows it.
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/flip497/savepoints /tmp/flip497/checkpoints
{code}
In {{conf/config.yaml}}, change {{numberOfTaskSlots: 1}} to {{8}} and the
TaskManager {{process.size: 1728m}} to {{4096m}}, and append:
{code:yaml}
execution.checkpointing.dir: file:///tmp/flip497/checkpoints
execution.checkpointing.savepoint-dir: file:///tmp/flip497/savepoints
execution.checkpointing.interval: 10s
{code}
Start the cluster with {{./bin/start-cluster.sh}}. Save the following as
{{/tmp/flip497/ddl.sql}} and start the SQL Client with {{./bin/sql-client.sh
\-i /tmp/flip497/ddl.sql}}. Run every statement below in that session. Each
{{INSERT INTO}} runs detached on the cluster and writes to a {{print}} sink.
Watch the output with {{tail \-f log/flink\-*\-taskexecutor\-*.out}}. Every
line starts with the sink's {{print-identifier}}, so several jobs can share the
file. Cancel jobs you are done with ({{STOP JOB '<id>';}}) to free slots.
The data: one order and one shipment per second. The shipment for order N is
generated about 10 seconds after order N, and the join only accepts even ids,
so even orders get a match about 10 seconds late and odd orders never match.
{{order_time}} and {{ship_time}} are the wall-clock time (UTC) at which each
row was generated, so you can compare them with when a row shows up. The window
is one minute, so without the hint an unmatched order waits about a minute.
{code:sql}
SET 'table.local-time-zone' = 'UTC';
CREATE TABLE orders (
order_id BIGINT,
order_time AS LOCALTIMESTAMP,
WATERMARK FOR order_time AS order_time - 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 shipments (
seq BIGINT,
order_id AS seq - 10,
ship_time AS LOCALTIMESTAMP,
WATERMARK FOR ship_time AS ship_time - INTERVAL '1' SECOND
) WITH ('connector' = 'datagen', 'rows-per-second' = '1',
'fields.seq.kind' = 'sequence', 'fields.seq.start' = '1', 'fields.seq.end' =
'100000');
-- Event time never advances on this input, so the join's watermark stalls
(step 6).
CREATE TABLE shipments_stalled (
seq BIGINT,
order_id AS seq - 10,
ship_time AS CAST(TIMESTAMP '2026-01-01 00:00:00' AS TIMESTAMP(3)),
WATERMARK FOR ship_time AS ship_time
) WITH ('connector' = 'datagen', 'rows-per-second' = '1',
'fields.seq.kind' = 'sequence', 'fields.seq.start' = '1', 'fields.seq.end' =
'100000');
-- Every fifth order has a NULL join key, which can never match (step 8).
CREATE TABLE orders_nk (
order_id BIGINT,
k AS CASE WHEN MOD(order_id, 5) = 0 THEN CAST(NULL AS BIGINT) ELSE order_id
END,
order_time AS LOCALTIMESTAMP,
WATERMARK FOR order_time AS order_time - 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');
-- Processing-time inputs (step 5).
CREATE TABLE orders_pt (order_id BIGINT, pt AS PROCTIME()) WITH ('connector' =
'datagen',
'rows-per-second' = '1', 'fields.order_id.kind' = 'sequence',
'fields.order_id.start' = '1', 'fields.order_id.end' = '100000');
CREATE TABLE shipments_pt (seq BIGINT, order_id AS seq - 10, pt AS PROCTIME())
WITH ('connector' = 'datagen',
'rows-per-second' = '1', 'fields.seq.kind' = 'sequence', 'fields.seq.start' =
'1', 'fields.seq.end' = '100000');
CREATE TABLE out_baseline (order_id BIGINT, ot TIMESTAMP(3), st TIMESTAMP(3))
WITH ('connector' = 'print', 'print-identifier' = 'baseline');
CREATE TABLE out_early (order_id BIGINT, ot TIMESTAMP(3), st TIMESTAMP(3)) WITH
('connector' = 'print', 'print-identifier' = 'early');
CREATE TABLE out_stalled_rt (order_id BIGINT, ot TIMESTAMP(3), st TIMESTAMP(3))
WITH ('connector' = 'print', 'print-identifier' = 'stalled-rowtime');
CREATE TABLE out_stalled_pt (order_id BIGINT, ot TIMESTAMP(3), st TIMESTAMP(3))
WITH ('connector' = 'print', 'print-identifier' = 'stalled-proctime');
CREATE TABLE out_full (o_id BIGINT, s_id BIGINT, ot TIMESTAMP(3), st
TIMESTAMP(3)) WITH ('connector' = 'print', 'print-identifier' = 'full');
CREATE TABLE out_nullkey (order_id BIGINT, k BIGINT, ot TIMESTAMP(3), st
TIMESTAMP(3)) WITH ('connector' = 'print', 'print-identifier' = 'nullkey');
CREATE TABLE out_plan (order_id BIGINT, ot TIMESTAMP(3), st TIMESTAMP(3)) WITH
('connector' = 'print', 'print-identifier' = 'plan');
CREATE TABLE out_csv (order_id BIGINT, ot TIMESTAMP(3), st TIMESTAMP(3)) WITH
('connector' = 'filesystem', 'path' = 'file:///tmp/flip497/csv', 'format' =
'csv');
{code}
Keep the datagen {{fields.*.end}} values small: the sequence generator holds
its whole range in memory. The two timestamps are cast to plain
{{TIMESTAMP(3)}} in the queries below because a sink cannot take two rowtime
columns.
h3. 1. Baseline vs. EARLY_FIRE
{code:sql}
SET 'pipeline.name' = 'flip497-baseline';
INSERT INTO out_baseline
SELECT o.order_id, CAST(o.order_time AS TIMESTAMP(3)), CAST(s.ship_time AS
TIMESTAMP(3))
FROM orders o LEFT JOIN shipments s
ON o.order_id = s.order_id AND MOD(s.order_id, 2) = 0
AND o.order_time BETWEEN s.ship_time - INTERVAL '1' MINUTE AND s.ship_time;
SET 'pipeline.name' = 'flip497-early';
INSERT INTO out_early
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, CAST(o.order_time AS
TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o LEFT JOIN shipments s
ON o.order_id = s.order_id AND MOD(s.order_id, 2) = 0
AND o.order_time BETWEEN s.ship_time - INTERVAL '1' MINUTE AND s.ship_time;
{code}
*Expected* (keep both running, steps 9 and 10 use them):
* {{early}}: every order appears as {{+I\[N, <order_time>, null\]}} 4 to 5
seconds after its {{order_time}}: the 3 s delay, the 1 s watermark lag, and up
to one more second because each input's watermark only moves when its next row
arrives (one per second) and the join waits for the slower input. About 10
seconds after {{order_time}}, every even order is corrected:
{noformat}
early> +I[2, 2026-11-12T01:01:43.221, null]
early> -U[2, 2026-11-12T01:01:43.221, null]
early> +U[2, 2026-11-12T01:01:43.221, 2026-11-12T01:01:53.231]
{noformat}
Odd orders are never corrected.
* {{baseline}}: even orders appear once, already joined, about 10 seconds after
{{order_time}}. Odd orders appear as {{+I\[N, <order_time>, null\]}} only 61 to
62 seconds after {{order_time}}, when the window closes.
h3. 2. The same query interactively (optional)
{code:sql}
SET 'sql-client.execution.result-mode' = 'tableau';
SET 'pipeline.name' = 'flip497-select';
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, CAST(o.order_time AS
TIMESTAMP(3)) AS ot, CAST(s.ship_time AS TIMESTAMP(3)) AS st
FROM orders o LEFT JOIN shipments s
ON o.order_id = s.order_id AND MOD(s.order_id, 2) = 0
AND o.order_time BETWEEN s.ship_time - INTERVAL '1' MINUTE AND s.ship_time;
{code}
*Expected:* the {{op}} column shows the same {{+I}}, {{\-U}}, {{+U}} pattern.
The query holds the session until you stop it with Ctrl+C. Note that {{EXPLAIN
CHANGELOG_MODE}} on a bare {{SELECT}} shows the join as {{\[I,UA\]}}, as it
does for any updating query such as a {{GROUP BY}}. Run it on the {{INSERT INTO
out_early ...}} statement instead to see the {{\[I,UB,UA\]}} the sink receives.
h3. 3. The sink must accept updates
{code:sql}
INSERT INTO out_csv
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, CAST(o.order_time AS
TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o LEFT JOIN shipments s
ON o.order_id = s.order_id AND o.order_time BETWEEN s.ship_time - INTERVAL
'1' MINUTE AND s.ship_time;
{code}
*Expected:* planning fails with {{Table sink
'default_catalog.default_database.out_csv' doesn't support consuming update
changes, but the EARLY_FIRE hint makes this outer interval join produce update
changes ...}}
h3. 4. The hint is ignored on an inner join
{code:sql}
EXPLAIN CHANGELOG_MODE
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, CAST(o.order_time AS
TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o JOIN shipments s
ON o.order_id = s.order_id AND o.order_time BETWEEN s.ship_time - INTERVAL
'1' MINUTE AND s.ship_time;
{code}
*Expected:* the {{IntervalJoin(joinType=\[InnerJoin\] ...)}} node and the whole
plan stay {{changelogMode=\[I\]}}. The node still lists
{{earlyFireDelay=\[3000\]}}: the hint is attached but has no effect, as
documented for inner joins and joins with a negative window span.
h3. 5. Invalid hints fail at planning
Put each hint from the table in place of the hint in the first statement below.
For the last row, use the second statement, which joins the processing-time
inputs. {{EXPLAIN}} is enough, since the errors are raised during planning.
{code:sql}
EXPLAIN SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, CAST(o.order_time AS
TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o LEFT JOIN shipments s
ON o.order_id = s.order_id AND o.order_time BETWEEN s.ship_time - INTERVAL
'1' MINUTE AND s.ship_time;
EXPLAIN SELECT /*+ EARLY_FIRE('delay'='3s', 'time-mode'='rowtime') */ o.order_id
FROM orders_pt o LEFT JOIN shipments_pt s
ON o.order_id = s.order_id AND o.pt BETWEEN s.pt - INTERVAL '1' MINUTE AND
s.pt;
{code}
The first six errors are reported as {{java.lang.AssertionError}}. That is
expected: every SQL hint option check (also {{LOOKUP}} and {{STATE_TTL}})
reports errors this way.
||Hint||Expected message contains||
|{{EARLY_FIRE('time-mode'='rowtime')}}|{{incomplete required option(s): \[Key:
'delay'}}|
|{{EARLY_FIRE('delay'='0s')}}|{{delay value should be at least 1 millisecond}}|
|{{EARLY_FIRE('delay'='1ns')}}|{{delay value should be at least 1 millisecond}}|
|{{EARLY_FIRE('delay'='3s', 'target'='window_join')}}|{{only 'interval_join' is
supported currently}}|
|{{EARLY_FIRE('delay'='3s', 'foo'='bar')}}|{{Unsupported EARLY_FIRE hint
option(s) \[foo\]}}|
|{{EARLY_FIRE('3s')}}|{{only support key-value options}}|
|{{EARLY_FIRE('delay'='3s', 'time-mode'='rowtime')}} (second
statement)|{{Row-time triggering requires a row-time interval join}}|
h3. 6. proctime mode keeps firing when the watermark stalls
{code:sql}
SET 'pipeline.name' = 'flip497-stalled-rowtime';
INSERT INTO out_stalled_rt
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, CAST(o.order_time AS
TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o LEFT JOIN shipments_stalled s
ON o.order_id = s.order_id AND o.order_time BETWEEN s.ship_time - INTERVAL
'1' MINUTE AND s.ship_time;
SET 'pipeline.name' = 'flip497-stalled-proctime';
INSERT INTO out_stalled_pt
SELECT /*+ EARLY_FIRE('delay'='3s', 'time-mode'='proctime') */ o.order_id,
CAST(o.order_time AS TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o LEFT JOIN shipments_stalled s
ON o.order_id = s.order_id AND o.order_time BETWEEN s.ship_time - INTERVAL
'1' MINUTE AND s.ship_time;
{code}
*Expected:* {{stalled-rowtime}} prints nothing, because the watermark never
passes the orders. {{stalled-proctime}} prints {{+I\[N, <order_time>, null\]}}
3 seconds after each order, on wall-clock time.
h3. 7. FULL OUTER join
{code:sql}
SET 'pipeline.name' = 'flip497-full';
INSERT INTO out_full
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, s.order_id,
CAST(o.order_time AS TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o FULL JOIN shipments s
ON o.order_id = s.order_id AND MOD(s.order_id, 2) = 0
AND o.order_time BETWEEN s.ship_time - INTERVAL '1' MINUTE AND s.ship_time;
{code}
*Expected:* orders behave as in step 1 ({{+I\[N, null, <order_time>, null\]}}
after 4 to 5 s, even ones corrected later). Shipments that never match, such as
{{s_id}} -9 to 0 and the odd ids, appear as {{+I\[null, M, null,
<ship_time>\]}} 1 to 2 seconds after {{ship_time}}. That is not early fire. A
shipment can only match orders at or before its own time, so its window closes
as soon as the watermark passes it, and the same query without the hint gives
the same 1 to 2 seconds. Early fire only shortens the wait for rows whose
window is still open.
h3. 8. NULL join key on the preserved side
{code:sql}
SET 'pipeline.name' = 'flip497-nullkey';
INSERT INTO out_nullkey
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, o.k, CAST(o.order_time AS
TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders_nk o LEFT JOIN shipments s
ON o.k = s.order_id AND MOD(s.order_id, 2) = 0
AND o.order_time BETWEEN s.ship_time - INTERVAL '1' MINUTE AND s.ship_time;
{code}
*Expected:* orders 5, 10, 15, ... have a NULL key. Each appears once as
{{+I\[N, null, <order_time>, null\]}} 4 to 5 seconds after its {{order_time}}
and is never corrected, including the even ones. The other orders behave as in
step 1.
h3. 9. Stop with a savepoint and resume
Let {{flip497-early}} from step 1 run for at least 30 seconds. Find its id with
{{SHOW JOBS;}} (the {{RUNNING}} row with that name), then:
{code:sql}
STOP JOB '<id of flip497-early>' WITH SAVEPOINT;
SET 'execution.state-recovery.path' = '<the savepoint path printed above>';
SET 'pipeline.name' = 'flip497-early-restored';
INSERT INTO out_early
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, CAST(o.order_time AS
TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o LEFT JOIN shipments s
ON o.order_id = s.order_id AND MOD(s.order_id, 2) = 0
AND o.order_time BETWEEN s.ship_time - INTERVAL '1' MINUTE AND s.ship_time;
RESET 'execution.state-recovery.path';
{code}
Resume within about 30 seconds, since the window is one minute. *Expected:*
even orders that were early-fired before the stop are corrected with
{{\-U}}/{{+U}} after the resume, and no order is padded a second time. This
must print nothing:
{code:bash}
grep -oE 'early> \+I\[[0-9]+,' log/flink-*-taskexecutor-*.out | sort | uniq -d
{code}
h3. 10. Add the hint to a running job through a savepoint
Stop {{flip497-baseline}} from step 1 with a savepoint, then resume it from
that savepoint with the hint added:
{code:sql}
STOP JOB '<id of flip497-baseline>' WITH SAVEPOINT;
SET 'execution.state-recovery.path' = '<the savepoint path printed above>';
SET 'pipeline.name' = 'flip497-baseline-with-hint';
INSERT INTO out_baseline
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, CAST(o.order_time AS
TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o LEFT JOIN shipments s
ON o.order_id = s.order_id AND MOD(s.order_id, 2) = 0
AND o.order_time BETWEEN s.ship_time - INTERVAL '1' MINUTE AND s.ship_time;
RESET 'execution.state-recovery.path';
{code}
*Expected:* the job restores. Rows that were waiting in state at the savepoint
keep their original timing: even orders arrive joined with a plain {{+I}}, and
odd orders are padded when their window closes. Rows that arrive after the
resume are early-fired and corrected as in step 1. The duplicate check from
step 9 with {{baseline>}} instead of {{early>}} must print nothing.
h3. 11. Compiled plan
{code:sql}
SET 'pipeline.name' = 'flip497-plan';
COMPILE PLAN 'file:///tmp/flip497/plan.json' IF NOT EXISTS FOR INSERT INTO
out_plan
SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, CAST(o.order_time AS
TIMESTAMP(3)), CAST(s.ship_time AS TIMESTAMP(3))
FROM orders o LEFT JOIN shipments s
ON o.order_id = s.order_id AND MOD(s.order_id, 2) = 0
AND o.order_time BETWEEN s.ship_time - INTERVAL '1' MINUTE AND s.ship_time;
EXECUTE PLAN 'file:///tmp/flip497/plan.json';
{code}
*Expected:* {{grep earlyFire /tmp/flip497/plan.json}} shows {{"earlyFireDelay"
: 3000}} and {{"earlyFireTimeMode" : "ROWTIME"}} (plus the join node's
description line, which mentions both), and the {{plan}} output follows the
step 1 pattern.
h3. 12. Batch mode
The step 1 tables are unbounded, so this step uses two small bounded ones.
{{EXPLAIN}} is enough.
{code:sql}
CREATE TEMPORARY TABLE b_orders (order_id BIGINT, order_time TIMESTAMP(3)) WITH
('connector' = 'datagen', 'number-of-rows' = '100');
CREATE TEMPORARY TABLE b_shipments (order_id BIGINT, ship_time TIMESTAMP(3))
WITH ('connector' = 'datagen', 'number-of-rows' = '100');
SET 'execution.runtime-mode' = 'batch';
EXPLAIN SELECT /*+ EARLY_FIRE('delay'='3s') */ o.order_id, o.order_time,
s.ship_time
FROM b_orders o LEFT JOIN b_shipments s
ON o.order_id = s.order_id AND o.order_time BETWEEN s.ship_time - INTERVAL
'1' MINUTE AND s.ship_time;
SET 'execution.runtime-mode' = 'streaming';
{code}
*Expected* until FLINK-41446 is fixed: planning fails with {{Unknown join
strategy : EARLY_FIRE}}, while the same {{EXPLAIN}} without the hint plans a
{{HashJoin}}. Once the fix is in, re-run it: the query with the hint should
plan as well.
h3. Exploratory (optional)
* Run step 1 with {{SET 'parallelism.default' = '2';}} and check that every
{{\-U}} is directly followed by its {{+U}} and that no order is padded twice.
* Repeat steps 1 and 9 with {{SET 'state.backend.type' = 'rocksdb';}}.
* Feed the early-fire output into a {{GROUP BY}} and check that the final
aggregates match the same query without the hint.
* Try a {{RIGHT}} join, a delay longer than the window, or extra non-equi
conditions in {{ON}}.
h3. Known behavior (not bugs)
* On inner joins and joins with a negative window span the hint is ignored and
the output stays append-only. {{EXPLAIN}} still shows {{earlyFireDelay}} on the
join node.
* Each outer row fires at most once. The hint does not emit periodically.
* Early-fire timers are not removed when a row matches or expires first. They
fire later and find nothing to do.
* {{'time-mode'='proctime'}} on an event-time join keeps extra state on each
input to track pending early fires. Matching and state cleanup still follow
event time.
* Writing the output to a sink with a primary key needs an {{ON CONFLICT}}
clause on the {{INSERT}}, because the join's upsert key differs from the sink's
primary key. The planner's error message lists the options. The same {{INSERT}}
without the hint is accepted as is.
* The hint applies only to the interval join in its own query block. A hint on
an outer {{SELECT}} does not reach a join inside a subquery. Early-fire output
is an updating stream and an interval join only accepts append-only input, so
two interval joins in one block with the hint fail at planning.
* On a regular join without a time condition, the hint is ignored.
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)
