fhueske opened a new pull request, #29809:
URL: https://github.com/apache/flink/pull/29809
## What is the purpose of the change
A `LATERAL SNAPSHOT` join stores its build side as a row -> count multi-set
and matches retractions by full row. An upsert build source (`I,UA,D`, partial
deletes) therefore needs a `ChangelogNormalize` to produce full before-images,
and non-deterministic build columns are rejected (FLINK-40890).
This PR adds a build-side layout that stores rows by upsert key, so updates
overwrite and deletes (including partial ones carrying only the key) remove by
key. It is used when the build side has an upsert key containing all join keys;
otherwise the multi-set layout is used as before.
## Brief change log
- **Runtime:** `LateralSnapshotJoinOperator` becomes an abstract base
(phases, buffering, timers, metrics) with `MultiSetLateralSnapshotJoinOperator`
(existing behavior) and `UpsertLateralSnapshotJoinOperator` (new state
`build-table-upsert`, keyed by a projected `BinaryRowData` upsert key). The
base keeps its name because the `Phase` enum's binary name is persisted in
savepoints.
- **Planner:** `StreamPhysicalLateralSnapshotJoin#getBuildUpsertKey`
decides eligibility (smallest upsert key that contains the join keys).
`StreamExecLateralSnapshotJoin` has a new optional `buildUpsertKeys` field and
picks the operator from it.
- **Changelog mode:** `FlinkChangelogModeInferenceProgram` requests
`ONLY_UPDATE_AFTER` / `DELETE_BY_KEY` from an eligible build side (falling back
to the previous requirements), so `ChangelogNormalize` is dropped for upsert
sources.
- **Determinism:** `StreamNonDeterministicUpdatePlanVisitor` no longer
requires whole-row determinism for the build side with the upsert layout. It is
still required for the multi-set layout.
- **Compatibility:** no ExecNode version bump. The field is
`@Nullable`/`NON_NULL`, so existing plans serialize unchanged and plans without
it restore with the multi-set layout.
## Verifying this change
This change added tests and can be verified as follows:
- `UpsertLateralSnapshotJoinOperatorTest`: overwrite by key, partial
delete, unmatched retractions, outer join, multiple keys per join key, restore
on HEAP/RocksDB.
- `MultiSetLateralSnapshotJoinOperatorTest` (renamed from
`LateralSnapshotJoinOperatorTest`): unchanged coverage of the multi-set layout.
- `LateralSnapshotJoinTest`: no `ChangelogNormalize` for an upsert build
source, for a composite key `(bk, id)`, and a fallback to `ChangelogNormalize`
when the key does not contain the join key. Each case also asserts the chosen
operator and the serialized `buildUpsertKeys`. Non-deterministic non-key
columns are accepted with an upsert key and still rejected without one.
- `LateralSnapshotJoinRestoreTest`: new
`lateral-snapshot-join-upsert-build` program with a savepoint. The existing
restore resources are untouched and still pass.
- `LateralSnapshotJoinITCase`: composite primary key `(bk, id)` on HEAP
and RocksDB.
## 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, the
build-side change path of the new layout. The multi-set layout is unchanged.
- Anything that affects deployment or recovery: JobManager (and its
components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no. Existing plans and
savepoints are unaffected; only newly compiled plans with an eligible build
side use the new state.
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? no (planner optimization
of an existing feature)
- 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 4.8)
--
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]
