rionmonster opened a new pull request, #29583:
URL: https://github.com/apache/flink/pull/29583
> **Stacked on #29551 (FLINK-41055).** Only the last two commits belong to
this PR; I'll rebase once #29551 is merged.
## What is the purpose of the change
The planner lets an input of the multi-way join omit UPDATE_BEFORE and
delete by key if one of its keys contains the common join key.
`StreamingMultiJoinOperator` handled every accumulate as a new record and
joined deletes with their null fields. From SQL, with upsert sources and an
upsert sink:
- **Lost matches stay in the output.** A record replaced by a version that
matches differently didn't retract its joined rows. This happens with a
non-equi condition, or with an equi condition on an attribute that a later
input joins on. In a LEFT join, the null padding wasn't restored.
- **Old versions stay in state.** An update that changes a join key field
outside the common join key stored the new version under a new state key. The
old version stayed in state and kept joining.
- **Deletes by key retract nothing.** A delete by key was joined with its
null fields, so it retracted nothing, and with such a join key it wasn't
removed from state.
This follows the ticket and the approach of FLINK-40841 and FLINK-40858 for
the binary join:
- look up the stored record with the same key;
- retract the joined rows and null padding that only the old version
produces;
- join retractions with the stored record.
## Brief change log
**Commit 1: use the upsert keys of the inputs.**
- The multi-way join used `getUniqueKeys` for the inputs, both to decide
whether an input may omit UPDATE_BEFORE and to key its state.
- An input without UPDATE_BEFORE replaces the record with the same upsert
key. So the multi-way join now uses the upsert keys in the key group range of
the common join key, like `StreamPhysicalJoin` does.
- Compiled plans get a new field, `inputUpsertKeys`. Plans compiled before
Flink 2.4 contain only `inputUniqueKeys` and keep using them, so the state of
their inputs stays keyed as before.
**Commit 2: handle upsert inputs in `StreamingMultiJoinOperator`.**
- **Accumulates:** the operator looks up the stored record with the same
key. Only when a join condition reads a field that the new version changes, the
operator runs the old version as a retraction and the new version as an
accumulate into a buffer, then emits the difference:
- a retraction is dropped when the new version produces a joined row with
the same rows of the other inputs, which updates it downstream;
- null padding that the old version inserts and the new version retracts
is dropped from both sides.
The existing recursion is unchanged.
- **Deletes by key:** these are joined with the stored record and remove it
under its own join key.
- **Lookups only where they can change the result:**
- The lookup only happens if the join conditions read a field of the input
outside its key. Two versions of a record share their key, so no other field
can make the conditions match them differently.
- `JoinInputSideSpec#uniqueKeyContainsJoinKey` (new, set by
`JoinUtil.analyzeJoinInput`) tells the state view whether all versions of a
record have the same join key. It only scans the records of the current common
join key if they may not.
- **Plan format:** compiled plans get a new field, `inputChangelogModes`,
like `StreamExecJoin` since FLINK-40841. Plans compiled before Flink 2.4 don't
have it. Their inputs whose key contains the common join key are handled as
upsert inputs with deletes by key; the planner never let other inputs omit
UPDATE_BEFORE.
## Verifying this change
- **`StreamingMultiJoinUpsertOperatorTest`:** a new harness test, on heap
and RocksDB. A four-way join covers:
- an update that moves the join key;
- an insert that replaces a stored record;
- an update of the first input;
- deletes by key, with and without a non-equi condition;
- a non-equi update that loses its match (null padding is inserted);
- a non-equi update that keeps its match (only `+U`);
- an update that gains a deeper match (the old version's null padding is
retracted).
- **Red/green:**
- With the lookups disabled, all 7 bug cases fail on both backends. The 2
control cases pass either way.
- With the difference step disabled, exactly the two cases that need it
fail.
- **`MultiJoinRestoreTest`:** two new programs with the same data. After the
restore, an order moves to another region and an order is deleted by key, so
their full images are only in the restored state.
- `three-way-join-upsert-inputs-with-restore` uses the new plan format.
- `three-way-join-upsert-inputs-pre-2-4-with-restore` uses a plan and
savepoint generated with the code before this PR, which has `inputUniqueKeys`
and no changelog modes.
Both fail without the fix and pass with it. The existing multi-join
restore tests, all with plans that have `inputUniqueKeys`, pass unchanged.
- **`MultiJoinUtilTest`:** covers the new `createConditionFields`.
- **Existing suites:** the runtime join tests (415) and the planner's
multi-join, join, changelog-mode, NDU and restore tests pass.
## Performance
I ran an A/B harness benchmark of the operator before and after this PR, on
heap and RocksDB:
- Users ⋈ Orders ⋈ Payments on `user_id`, with 2,000 users, 10 orders each
and one payment each.
- 20,000 operations on Orders, whose key is `(user_id, order_id)`.
- Median of 5 runs after a warm-up, in ns per input record.
| Workload on Orders | Heap before | Heap after | RocksDB before | RocksDB
after |
|---|---|---|---|---|
| retract input, value updates | 1,331 | 1,270 | 4,225 | 4,118 |
| upsert, value updates | 1,421 | 1,403 | 4,289 | 4,202 |
| upsert, delete by key and re-insert | 1,238 | 1,263 | 4,090 | 4,126 |
| upsert, inserts of new orders | 1,367 | 1,226 | 4,228 | 4,383 |
| plan without changelog modes, retract input, value updates | 1,268 | 1,328
| 4,093 | 4,133 |
| upsert, value updates, `u.region = o.region` in the condition | 1,832 |
2,258 | 4,625 | 5,418 |
| upsert, region updates, `u.region = o.region` in the condition | 1,423 |
2,888 | 3,290 | 7,743 |
- **No change** when no lookup can change the result: retract inputs, or
conditions that only read key fields. These are the first five rows, all within
noise.
- **About +17% on RocksDB** when the conditions read a non-key field that
the update doesn't change. That's one point lookup and a field comparison per
record.
- **About 2.35x** when every update changes a condition field. The old
version's joined rows are retracted, which the code before this PR skipped and
so emitted wrong results. In this workload the order's join key `(user_id,
region)` isn't part of its key, so the lookup also scans the user's orders.
- **Unnecessary scans are gone.** An intermediate version scanned on every
lookup that missed. That cost up to 2.4x on inserts of new orders;
`JoinInputSideSpec#uniqueKeyContainsJoinKey` removes it.
## 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, see
Performance
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no. The plan format
changes in a compatible way, see Points worth a look 1 and 2.
- 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]
