Aaron He created FLINK-41355:
--------------------------------
Summary: Introduce async state version of StreamExecIntervalJoin
Key: FLINK-41355
URL: https://issues.apache.org/jira/browse/FLINK-41355
Project: Flink
Issue Type: Sub-task
Components: Table SQL / Planner, Table SQL / Runtime
Reporter: Aaron He
h2. Motivation
FLIP-473 (FLINK-36157) moves SQL operators to the async state API, selected by
table.exec.async-state.enabled. The interval join has no async version:
StreamExecIntervalJoin always builds RowTimeIntervalJoin or
ProcTimeIntervalJoin on synchronous state, and both places still carry "TODO:
add async version ... to use AsyncKeyedCoProcessOperator". A job with an
interval join therefore falls back to synchronous state on a disaggregated
state backend such as ForSt, even when its other stateful operators are async.
The DataStream interval join already has an async version (FLINK-37003).
h2. Proposal
Add AsyncStateRowTimeIntervalJoin and AsyncStateProcTimeIntervalJoin (sharing a
base class) in
org.apache.flink.table.runtime.operators.join.interval.asyncprocessing, run by
the existing AsyncKeyedCoProcessOperator and
AsyncKeyedCoProcessOperatorWithWatermarkDelay, and select them in
StreamExecIntervalJoin when table.exec.async-state.enabled is true.
* State: the same descriptors as the synchronous functions
(IntervalJoinLeftCache / IntervalJoinRightCache as MapState<Long,
List<Tuple2<RowData, Boolean>>>, and the two cleanup-timer value states),
accessed through the async state API.
* Logic: the same join, outer padding and cleanup-timer rules as
TimeIntervalJoin, including FLINK-41067 (rows whose join key can never match
are not cached).
* Join key: records of several keys are in flight at a time, so
IntervalJoinFunction gets a join overload that takes the join key explicitly.
* Cleanup: after the changed buckets are written, all expired buckets of a key
are removed with one MapState#asyncPutAll that maps each expired timestamp to
null. When no row of the key survives, the timer state is cleared and the
buckets are bulk-deleted, or cleared with MapState#asyncClear above 65,536 row
times. No configuration option.
* Rows at time zero or earlier are kept and joined like later rows, and cleanup
times saturate instead of overflowing.
* Early fire (FLIP-497): with an EARLY_FIRE hint, the planner keeps the
synchronous functions. Async early fire follows as a separate subtask.
* No new configuration option, no JSON plan change and no ExecNode version
change.
* Docs: add the interval join to the supported operators on the disaggregated
state page.
h2. Dependency
The bulk removal relies on MapState#asyncPutAll removing keys mapped to null,
as its contract states. ForSt (FLINK-41091) and the adaptor used on HashMap and
RocksDB (FLINK-41095) do not honor that yet. This subtask should be merged
after both.
h2. Compatibility
The async functions use the same state names and serializers, so a job can
switch between them where the backend restores the same descriptors (HashMap,
RocksDB). On ForSt, the synchronous backend keeps timers in ForSt by default
while the async backend keeps them in a heap queue, so switching a timer-using
operator between synchronous and async state on ForSt is not covered, as for
every async operator with timers.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
