[
https://issues.apache.org/jira/browse/FLINK-41055?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Gustavo de Morais reassigned FLINK-41055:
-----------------------------------------
Assignee: Gustavo de Morais
> StreamingMultiJoinOperator stores rows with a NULL join key in state, even
> though they can never match
> ------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41055
> URL: https://issues.apache.org/jira/browse/FLINK-41055
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / Runtime
> Reporter: Rion Williams
> Assignee: Gustavo de Morais
> Priority: Major
>
> h3. What happens
> This is the multi-way join part of FLINK-41034. That ticket's fix covers the
> binary streaming joins (StreamingJoinOperator, its mini-batch and async-state
> variants, and StreamingSemiAntiJoinOperator), which check
> JoinConditionWithNullFilters before storing a row. StreamingMultiJoinOperator
> doesn't use JoinConditionWithNullFilters: processElement always calls
> addRecordToState, which adds (or retracts) the row under its join key no
> matter what the key holds.
> A row whose join key has a NULL in an equality that isn't null-safe can never
> pass its join condition, so nothing ever reads it from state again. It stays
> there until state TTL removes it. All of these rows share the NULL key, so
> they all land in the same key group and on the same subtask, which is the
> same skew FLINK-41034 describes.
> This only affects plans with table.optimizer.multi-join.enabled = true
> (default: false).
> h3. Example
> {code:sql}
> SET 'table.optimizer.multi-join.enabled' = 'true';
> -- every order with customer_id = NULL is stored in the multi-join's state
> for the orders input
> SELECT o.order_id, c.name, p.amount
> FROM orders o
> JOIN customers c ON o.customer_id = c.customer_id
> LEFT JOIN payments p ON o.customer_id = p.customer_id;
> {code}
> h3. Expected behaviour
> A row whose join key can never match shouldn't be stored. The output stays
> exactly the same, including null padding and retractions for rows on the
> preserved side of a LEFT join.
> h3. Why this needs its own fix
> The binary-join fix doesn't carry over directly:
> * The multi-join gets its join keys from AttributeBasedJoinKeyExtractor, not
> from a JoinConditionWithNullFilters, so the planner doesn't pass it a per-key
> filterNulls / null-safe flag. We need to check whether the multi-join can
> contain IS NOT DISTINCT FROM keys at all. If it can, the planner has to pass
> that information to the operator so those rows keep being stored.
> * An input's row is used in two ways: it's looked up by its own join key, and
> it acts as the left side when later inputs are looked up
> (getLeftSideJoinKey). Before skipping a row, we have to make sure that a NULL
> in its stored key means the row can't take part in any output row, not only
> that it can't match the earlier input.
> * For input 0 (and the preserved side of a LEFT join in general) we have to
> make sure that emitting the null-padded row, and later retracting it, doesn't
> depend on the row being in state. In the binary joins that was already true.
> In the multi-join, the association tracking in recursiveMultiJoin /
> processInputRecord needs to be checked.
> h3. Proposed fix
> In addRecordToState, once the points above are confirmed, skip adding rows
> whose state key contains a NULL in a field that isn't null-safe. Retractions
> keep going to state as before: removing a row that was never stored is a
> no-op, and a row that an earlier version stored is still removed by its
> retraction.
> h3. Tests
> * StreamingMultiJoinOperator tests for inner and LEFT joins with NULL keys on
> each input position: state stays empty for those rows, and the output
> (including null padding and retractions) is unchanged.
> * A case where a row that an earlier version stored (with a NULL key) is
> retracted.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
