[
https://issues.apache.org/jira/browse/FLINK-41034?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18130255#comment-18130255
]
Gustavo de Morais commented on FLINK-41034:
-------------------------------------------
Fixed in apache/flink:master 8d01380ff5bf9892462ef9808b1b56fd9723f54d
> 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
> Assignee: Rion Williams
> Priority: Major
> Labels: pull-request-available
>
> 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
(v10.3.26#10030026)
