fresh-borzoni opened a new pull request, #29653: URL: https://github.com/apache/flink/pull/29653
Linked issue: https://issues.apache.org/jira/browse/FLINK-38753 Supersedes #27372, which was auto-closed as stale (earlier review discussion is there). Rebased on current master. ## What is the purpose of the change Recognizes upsert keys through equi-join column equivalences to avoid unnecessary `upsertMaterialize` in streaming sinks. Consider: `src1` pk `{a1}`, `src2` pk `{a2, b2}`, join on `a1 = a2`, sink pk `{a1, b2}`. Currently the join produces upsert keys `{a2, b2}` and `{a1, a2, b2}` - neither contains `{a1, b2}`, so Flink adds `upsertMaterialize`. But since `a1 = a2` in the join, `{a1, b2}` is semantically equivalent to `{a2, b2}`. Since `table.exec.sink.require-on-conflict` defaults to `true` (Flink 2.3), such a query no longer just gets `upsertMaterialize`: it fails planning with "The query has an upsert key that differs from the primary key of the sink table ... Please specify an ON CONFLICT clause" unless an `ON CONFLICT` clause is added. This PR makes the planner recognize that equivalence. ## Brief change log - Added `UpsertKeyEquivalenceUtil.upsertKeysWithin(rel, target, mq)`: finds the input's upsert keys that fall within the sink primary key, substituting equi-join-equivalent columns where null-safe (only toward a non-nullable column) - Used it in the two sink-side checks: `FlinkChangelogModeInferenceProgram#canUpsertKeysWithImmutableColsSatisfyPk` and `StreamPhysicalSink#primaryKeysContainsUpsertKey` - `FlinkRelMdUpsertKeys` / `getUpsertKeys` is unchanged: equivalent key variants are not added to the upsert key set ## Verifying this change New plan tests in `JoinTest`: - `testJoinUpsertKeyEquivalenceInnerJoinBasic` - basic case from FLINK-38753 - `testJoinUpsertKeyEquivalenceInnerJoinReverse` - substitution in the opposite direction - `testJoinUpsertKeyEquivalenceLeftJoin` - left outer join - `testJoinUpsertKeyEquivalenceRightJoin` - right outer join - `testJoinUpsertKeyEquivalenceMultipleEquiConditions` - `a1 = a2 AND b1 = b2`, multi-step substitution - `testJoinUpsertKeyEquivalenceThreeWayJoin` - the satisfying equality is in the deeper of two chained joins - `testJoinUpsertKeyEquivalenceMultiJoinOperator` - same three-way join with the fused multi-join operator - `testJoinUpsertKeyEquivalenceSemiJoin` / `testJoinUpsertKeyEquivalenceAntiJoin` - `EXISTS` / `NOT EXISTS` (left-only output) - `testJoinUpsertKeyEquivalenceNegativeCase` - confirms `upsertMaterialize` when sink pk uses a non-join column - `testJoinUpsertKeyEquivalenceOneToManyNotSatisfied` - one-to-many left join keeps `upsertMaterialize` - `testJoinUpsertKeyEquivalenceThreeInputMultiJoinNotSatisfied` - 3-input star multi-join with a one-to-many input keeps `upsertMaterialize` ## 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): no - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no - 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 ## Rationale Equi-join conditions create value equivalences between join columns, so multiple column sets may represent the same logical row identity and therefore be valid upsert keys. Rather than expanding the upsert key set with every equivalent variant (which can grow combinatorially), this PR resolves equivalences on demand at the sink. The sink primary key is a small fixed target that is threaded down the plan: at each join, keys are repaired via null-safe equi-substitution, and the resolver recurses into the side the join preserves (the other side is unique on the join key and not null-padded). This keeps `getUpsertKeys` unchanged and the work bounded. Resolution happens at the join rather than only at the sink because an intermediate projection between joins usually drops the redundant equated column, so by the time the input reaches the sink the equivalence is no longer visible. Multi-step substitution is handled: with `a1 = a2 AND b1 = b2`, base key `{a2, b1}` resolves to `{a1, b2}` (`{a2, b1}` → `{a1, b1}` → `{a1, b2}`). Covers binary joins (inner/outer, chained) and the fused MultiJoin operator. ##### Was generative AI tooling used to co-author this PR? - [x] Yes (please specify the tool below) 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
