Mate Czagany created FLINK-40336:
------------------------------------
Summary: Avoid full state scans in event-time temporal join on
sorted state backends
Key: FLINK-40336
URL: https://issues.apache.org/jira/browse/FLINK-40336
Project: Flink
Issue Type: Improvement
Components: Table SQL / Runtime
Reporter: Mate Czagany
h1. Summary
On every watermark firing,
`TemporalRowTimeJoinOperator#emitResultAndCleanUpState` does per key:
* a full scan of `leftState`, including all probe records far beyond the
watermark, because the map is keyed by an arrival sequence number, and
* a full materialization + sort of `rightState` (`getRightRowSorted`), even
though only versions <= watermark are joinable.
With large state per key (e.g. probe records buffered far into the future, or
many build-side versions), this can become very slow and bottleneck the
application. The class has carried TODOs asking for ordered map state since its
introduction.
RocksDB and ForSt iterate MapState in serialized-key byte order, so with
order-preserving key serialization both scans can stop at the first entry newer
than the watermark. Other classes already depend on this behavior:
* InputSortBuffer
* WatermarkCompactingSinkMaterializer
* LateralSnapshotJoinOperator
h1. Proposal
* New class called `TemporalRowTimeJoinOperatorV2`:
** `leftState` keyed by (row time, arrival index) -> iterate in time order,
break at the watermark; the due subset is re-sorted by arrival index in memory
before emission, so output stays byte-identical to the current operator.
** `rightState` keyed with `SortedLongSerializer` -> build the sorted
right-version list by iterating up to the watermark; no sort, materialization
bounded to due versions. Join lookup and keep-latest-version cleanup are
unchanged and provably equivalent on the truncated list.
** Unordered backends (heap) fall back to full scans equivalent to today
* `StreamExecTemporalJoin` gets version 2 (state layout is incompatible with
v1)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
