[
https://issues.apache.org/jira/browse/FLINK-41355?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41355:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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)
