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