rionmonster opened a new pull request, #29532:
URL: https://github.com/apache/flink/pull/29532
## What is the purpose of the change
A streaming equi-join row whose join key has a NULL in a field that is not
null-safe can never match. The streaming joins already never match it, but they
still added it to their state, where it stayed until state TTL removed it.
Every NULL key is the same key, so all of these rows end up in the state of one
subtask. Fact tables with NULL foreign keys (e.g. TPC-DS `store_sales`) put all
their unmatched rows on a single join subtask.
This PR stops storing such rows. Output is unchanged.
**Why in the runtime and not as an `IS NOT NULL` filter in the planner:** a
filter cannot drop the preserved side of an outer join or the left side of an
anti join, because those rows must still be emitted. `JoinDeriveNullFilterRule`
is also only in the batch rule sets, only handles inner joins, and only fires
above 2,000,000 NULLs from statistics. The runtime check covers every join type
and side without changing plans.
## Brief change log
- `JoinConditionWithNullFilters#currentKeyCannotMatch()`, extracted from
`apply()` and reusing `NullAwareJoinHelper.shouldFilter`.
- `StreamingJoinOperator` (and so `MiniBatchStreamingJoinOperator`): an
accumulate message whose key can never match returns early. It is null padded
on an outer side, not stored, and the state lookups for it are skipped.
- `StreamingSemiAntiJoinOperator` and `AsyncStateStreamingJoinOperator`:
such rows are not stored.
- Retractions are unchanged, so a row stored by an earlier version is still
removed.
- `[hotfix]` commit: `HandwrittenKeySelector` checked the null bit at the
key index instead of the key field.
## Verifying this change
This change added tests:
- `StreamingJoinOperatorTest` covers inner, left, right and full outer
joins, with each input side spec (no unique key, unique key, unique key
contained by the join key), sync and async state:
- NULL-key rows are emitted as before and not stored;
- a NULL-safe join still stores and matches them;
- a NULL-key row restored from state is still removed by its retraction.
- `StreamingSemiAntiJoinOperatorTest` (semi and anti) and
`StreamingMiniBatchJoinOperatorTest` (a bundle with NULL keys).
- `JoinConditionWithNullFiltersTest` covers a key mixing `=` and `IS NOT
DISTINCT FROM` fields.
Each test asserts that a row with a non-null key *is* stored. All new tests
except the NULL-safe, restore and unit tests fail without the fix.
Existing coverage: `JoinRestoreTest` runs `JoinTestPrograms`, which contain
NULL join keys for inner, outer, semi and anti joins and a NULL-safe join,
against savepoints taken before this change.
Ran the full `flink-table-runtime` test suite and the planner's join
restore, semantic, plan and ITCase tests.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with
`@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): yes. One
null-bit check on the current key per accumulate message. A harness
micro-benchmark of a left outer join on keys that are never NULL (200k records,
three interleaved runs per variant) showed no measurable difference: a median
of 20.6–20.9 µs/record with the change and 20.7–20.9 without.
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no. The state layout
and the compiled plan are unchanged, and existing savepoints restore. NULL-key
rows stored by an earlier version stay until their retraction or state TTL
removes them.
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no
- If yes, how is the feature documented? not applicable
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Opus 5.5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
