[
https://issues.apache.org/jira/browse/FLINK-41067?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41067:
-----------------------------------
Labels: pull-request-available (was: )
> Window and interval joins store rows whose join key can never match
> -------------------------------------------------------------------
>
> Key: FLINK-41067
> URL: https://issues.apache.org/jira/browse/FLINK-41067
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / Runtime
> Affects Versions: 2.3.0
> Reporter: Rion Williams
> Priority: Major
> Labels: pull-request-available
>
> Like the regular streaming join before FLINK-41034, the window join and the
> interval join store every incoming row, including rows with a NULL in a join
> key field that isn't null-safe. Such rows can never match. The NULL check
> only runs when the join condition is evaluated:
> * {{WindowJoinOperator}} (and {{AsyncStateWindowJoinOperator}}):
> {{WindowJoinHelper#processElement}} adds every accumulate row to the left or
> right window state. The check is in {{JoinConditionWithNullFilters}}.
> * {{TimeIntervalJoin}} ({{RowTimeIntervalJoin}}, {{ProcTimeIntervalJoin}}):
> {{processElement1}}/{{processElement2}} add every row to the left or right
> cache. The check is in {{IntervalJoinFunction#join}}.
> The state is keyed by the join key, so all NULL-key rows go to the same
> subtask. Unlike the regular join, this state is bounded: window state is
> cleared when the window fires, and the interval caches are cleaned when the
> interval expires. So the impact is a hot subtask and larger state for the
> duration of a window or interval, not state that grows without bound.
> h3. Proposed change
> Don't store a row whose join key can never match, using
> {{JoinConditionWithNullFilters#currentKeyCannotMatch}} from FLINK-41034 (and
> the same check in {{IntervalJoinFunction}}).
> * Inner joins, and the non-preserved side of outer joins: drop the row.
> * Preserved side of outer joins: today the null-padded row is emitted when
> the window fires, or when the interval expires. Emitting it immediately
> instead would avoid storing the row, but it changes when the row is emitted.
> This ticket should decide whether that is acceptable, in particular for the
> interval join's early-fire padding and for downstream operators that rely on
> window-end emission. The conservative option is to keep storing rows on the
> preserved side.
> Both operators only receive append-only input (the window join rejects
> retractions), so there are no retractions of skipped rows to handle.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
