Rion Williams created FLINK-41067:
-------------------------------------
Summary: 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
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)
