[
https://issues.apache.org/jira/browse/FLINK-41034?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Rion Williams updated FLINK-41034:
----------------------------------
Description:
h3. What happens
In SQL, {{NULL = NULL}} is never true. So in an equi-join, a row whose join key
is NULL can never find a match. The streaming join operators already know this
and never emit a match for such a row, but they still add the row to their
state. Nothing ever reads it there again, and unless state TTL is configured,
it stays forever.
There's a second effect. Every NULL key is the same key, so all of these rows
hash to the same key group and land on the same subtask. A table with many NULL
foreign keys therefore piles all of its unmatched rows into the state of a
single join subtask. TPC-DS fact tables are an example: columns like
{{ss_sold_date_sk}} and {{ss_customer_sk}} are NULL for part of the rows.
Streaming plans don't remove these rows beforehand either. The rule that adds
{{IS NOT NULL}} filters on join keys ({{JoinDeriveNullFilterRule}}) only exists
in the batch rule sets.
h3. Example
{code:sql}
-- every order with customer_id = NULL is stored in the join's state,
-- all of them on the same subtask
SELECT o.order_id, c.name
FROM orders o LEFT JOIN customers c ON o.customer_id = c.customer_id;
{code}
Measured with
[StreamingJoinNullKeyStateITCase|https://gist.github.com/rionmonster/ada38929729779d339457e3a556e0b1a]
(parallelism 4, keyed state bytes of the join per subtask):
{noformat}
1,000 orders that match a customer: [8314, 7339, 7050, 7508]
the same plus 20,000 orders with a NULL key: [8314, 7336, 7053, 167728]
{noformat}
Every NULL-key order ends up on subtask 3, and none of them can ever join.
h3. Who is affected
* Regular streaming joins: inner, left/right/full outer. This includes the
mini-batch and async-state variants.
* Semi and anti joins ({{IN}} / {{EXISTS}} / {{NOT EXISTS}} subqueries).
* Multi-way joins ({{StreamingMultiJoinOperator}}).
* NULL-safe joins ({{IS NOT DISTINCT FROM}}) are not affected, and must keep
storing these rows.
h3. Why it matters
* *Unbounded state:* state grows with every NULL-key row, and state TTL is the
only thing that ever removes them.
* *Skew:* all of that state sits on one subtask, which makes its checkpoints
and restores slower than the rest of the job and limits how far the join scales.
h3. Expected behaviour
A row whose join key can never match shouldn't be stored. The output stays
exactly the same:
* an outer-join row with a NULL key is still emitted with NULL padding, and its
retraction is still emitted;
* an anti-join row with a NULL key is still emitted and retracted;
* inner and semi joins still emit nothing for it.
h3. Proposed fix
Before storing a row, run the same check that
{{JoinConditionWithNullFilters#apply}} already uses for matching (the key isn't
null-safe and has a NULL in a filtered field). If the row can never match,
don't store it.
This changes no output:
* the row is never returned as a match anyway;
* a later retraction of a row that was never stored is already a no-op in the
join state views;
* NULL padding an outer row doesn't depend on state.
{{StreamingMultiJoinOperator}} doesn't use {{JoinConditionWithNullFilters}}, so
it needs its own check.
h3. Workaround
Filter out NULL keys before the join:
* *Inner and semi joins:* add {{WHERE o.customer_id IS NOT NULL}}.
* *Outer and anti joins:* split the preserved side, so the output stays the
same:
{code:sql}
SELECT o.order_id, c.name
FROM orders o LEFT JOIN customers c ON o.customer_id = c.customer_id
WHERE o.customer_id IS NOT NULL
UNION ALL
SELECT order_id, CAST(NULL AS STRING) FROM orders WHERE customer_id IS NULL;
{code}
h3. Tests
*
[StreamingJoinNullKeyStateITCase|https://gist.github.com/rionmonster/ada38929729779d339457e3a556e0b1a]
reproduces this on master; it asserts the expected behaviour and currently
fails.
* Operator-level tests (inner, outer, semi and anti joins) will come with the
fix PR.
was:todo
> Streaming joins store rows with a NULL join key in state, even though they
> can never match
> ------------------------------------------------------------------------------------------
>
> Key: FLINK-41034
> URL: https://issues.apache.org/jira/browse/FLINK-41034
> Project: Flink
> Issue Type: Improvement
> Components: Table SQL / Runtime
> Affects Versions: 2.3.0
> Reporter: Rion Williams
> Priority: Major
>
> h3. What happens
> In SQL, {{NULL = NULL}} is never true. So in an equi-join, a row whose join
> key is NULL can never find a match. The streaming join operators already know
> this and never emit a match for such a row, but they still add the row to
> their state. Nothing ever reads it there again, and unless state TTL is
> configured, it stays forever.
> There's a second effect. Every NULL key is the same key, so all of these rows
> hash to the same key group and land on the same subtask. A table with many
> NULL foreign keys therefore piles all of its unmatched rows into the state of
> a single join subtask. TPC-DS fact tables are an example: columns like
> {{ss_sold_date_sk}} and {{ss_customer_sk}} are NULL for part of the rows.
> Streaming plans don't remove these rows beforehand either. The rule that adds
> {{IS NOT NULL}} filters on join keys ({{JoinDeriveNullFilterRule}}) only
> exists in the batch rule sets.
> h3. Example
> {code:sql}
> -- every order with customer_id = NULL is stored in the join's state,
> -- all of them on the same subtask
> SELECT o.order_id, c.name
> FROM orders o LEFT JOIN customers c ON o.customer_id = c.customer_id;
> {code}
> Measured with
> [StreamingJoinNullKeyStateITCase|https://gist.github.com/rionmonster/ada38929729779d339457e3a556e0b1a]
> (parallelism 4, keyed state bytes of the join per subtask):
> {noformat}
> 1,000 orders that match a customer: [8314, 7339, 7050, 7508]
> the same plus 20,000 orders with a NULL key: [8314, 7336, 7053, 167728]
> {noformat}
> Every NULL-key order ends up on subtask 3, and none of them can ever join.
> h3. Who is affected
> * Regular streaming joins: inner, left/right/full outer. This includes the
> mini-batch and async-state variants.
> * Semi and anti joins ({{IN}} / {{EXISTS}} / {{NOT EXISTS}} subqueries).
> * Multi-way joins ({{StreamingMultiJoinOperator}}).
> * NULL-safe joins ({{IS NOT DISTINCT FROM}}) are not affected, and must keep
> storing these rows.
> h3. Why it matters
> * *Unbounded state:* state grows with every NULL-key row, and state TTL is
> the only thing that ever removes them.
> * *Skew:* all of that state sits on one subtask, which makes its checkpoints
> and restores slower than the rest of the job and limits how far the join
> scales.
> h3. Expected behaviour
> A row whose join key can never match shouldn't be stored. The output stays
> exactly the same:
> * an outer-join row with a NULL key is still emitted with NULL padding, and
> its retraction is still emitted;
> * an anti-join row with a NULL key is still emitted and retracted;
> * inner and semi joins still emit nothing for it.
> h3. Proposed fix
> Before storing a row, run the same check that
> {{JoinConditionWithNullFilters#apply}} already uses for matching (the key
> isn't null-safe and has a NULL in a filtered field). If the row can never
> match, don't store it.
> This changes no output:
> * the row is never returned as a match anyway;
> * a later retraction of a row that was never stored is already a no-op in the
> join state views;
> * NULL padding an outer row doesn't depend on state.
> {{StreamingMultiJoinOperator}} doesn't use {{JoinConditionWithNullFilters}},
> so it needs its own check.
> h3. Workaround
> Filter out NULL keys before the join:
> * *Inner and semi joins:* add {{WHERE o.customer_id IS NOT NULL}}.
> * *Outer and anti joins:* split the preserved side, so the output stays the
> same:
> {code:sql}
> SELECT o.order_id, c.name
> FROM orders o LEFT JOIN customers c ON o.customer_id = c.customer_id
> WHERE o.customer_id IS NOT NULL
> UNION ALL
> SELECT order_id, CAST(NULL AS STRING) FROM orders WHERE customer_id IS NULL;
> {code}
> h3. Tests
> *
> [StreamingJoinNullKeyStateITCase|https://gist.github.com/rionmonster/ada38929729779d339457e3a556e0b1a]
> reproduces this on master; it asserts the expected behaviour and currently
> fails.
> * Operator-level tests (inner, outer, semi and anti joins) will come with the
> fix PR.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
