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