SEPURI-SAI-KRISHNA opened a new pull request, #29491:
URL: https://github.com/apache/flink/pull/29491
## What is the purpose of the change
`table.exec.sink.require-on-conflict` defaults to `true` and rejects a query
whose upsert key
differs from the sink table's primary key when no `ON CONFLICT` clause is
given. The check was
called from inside the `AUTO` branch of `analyzeUpsertMaterializeStrategy`,
so setting
`table.exec.sink.upsert-materialize` to `FORCE` skipped it and the query
planned.
Reproducer. The upsert key is the grouping key `c`, which is never written
to the sink, so it can
never equal the primary key `x`:
```sql
CREATE TABLE src (a INT, b BIGINT, c STRING)
WITH ('connector' = 'datagen', 'number-of-rows' = '5');
CREATE TABLE snk (x INT, y BIGINT, PRIMARY KEY (x) NOT ENFORCED)
WITH ('connector' = 'blackhole');
SET 'table.exec.sink.upsert-materialize' = 'FORCE';
INSERT INTO snk SELECT MAX(a), COUNT(*) FROM src GROUP BY c;
```
With `require-on-conflict` left at its default:
| upsert-materialize | before | after |
|---|---|---|
| `AUTO` (default) | ValidationException | unchanged |
| `FORCE` | planned, `upsertMaterialize=[true]` | ValidationException |
| `NONE` | planned, `upsertMaterialize=[false]` | unchanged, on purpose |
## Why FORCE and not NONE
`FORCE` adds the sink upsert materializer, and the materializer is the only
thing that applies a
conflict strategy:
- `CommonExecSink` gates it with
`upsertMaterialize && (!inputInsertOnly ||
detectsDuplicateKeys(conflictStrategy))`
- `DO ERROR` and `DO NOTHING` are implemented by
`WatermarkCompactingSinkMaterializer`, and
`DO DEDUPLICATE` by `SinkUpsertMaterializer`, both created only on that
path
So under `FORCE` the clause does take effect, and requiring it is the same
decision the planner
already makes under `AUTO`. This is also what FLIP-558 asks for: "If no ON
CONFLICT DO ... is
provided and the upsert key is different from the primary key of the sink we
would fail during the
planning."
`NONE` is deliberately left as it is. It makes
`analyzeUpsertMaterializeStrategy` return `false`,
so no materializer is built and an `ON CONFLICT` clause would have no effect
at all. Requiring one
there would ask the user to write a clause that is then ignored. That
opt-out is documented in
`docs/content/docs/sql/reference/dml/insert.md` under "When is ON CONFLICT
required?", so no
documentation change is needed and a test pins the behaviour.
## Brief change log
- Run `validateOnConflictSpecifiedIfRequired` in the `FORCE` branch of
`analyzeUpsertMaterializeStrategy`, with the same condition the `AUTO`
branch uses
- Extract that condition into `sinkIsUpsertWithPk` so both branches state
it once
- Say in the method's scaladoc which settings the check applies to, and
why `NONE` is excluded
- Set `require-on-conflict = false` in
`testForcedMaterializeWithAppendOnlyInput`, which covers
what `FORCE` materializes rather than `ON CONFLICT` enforcement
## Verifying this change
This change added tests and can be verified as follows:
- `TableSinkTest#testOnConflictIsRequiredWithForcedMaterialize` asserts
the `ValidationException`
under `FORCE`. It fails without the change, because the query plans
- `TableSinkTest#testOnConflictIsNotRequiredWithoutMaterialize` asserts
that `NONE` still plans
without a clause. It passes before and after, and guards the documented
opt-out
- `TableSinkTest` and `DeltaJoinTest` pass in full
`testForcedMaterializeWithAppendOnlyInput` needs the opt-out because its
input is append only and
carries no upsert key, so the clause becomes required. That is not new
behaviour introduced here:
its `AUTO` sibling `testAppendOnlyInputWithoutOnConflict` already sets the
same flag for the same
reason. No other test in the repository sets `upsert-materialize` to a
non-default value, and all
restore-test plans pin `AUTO`.
## 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
This is a user-visible behaviour change for `upsert-materialize = FORCE`, so
the Release Notes
field on FLINK-40924 is filled in. The `ON CONFLICT` documentation needs no
change, because the
only opt-out it names besides `require-on-conflict` is `upsert-materialize =
NONE`, which this PR
does not touch.
Two things worth a reviewer's attention:
- FLINK-40899 reports that this same validation already fires too early
for a lookup join with a
non-deterministic condition and masks the NDU error. Enforcing it under
`FORCE` as well widens
that case slightly. Keeping `NONE` out of scope keeps the exposure small
- The same gap exists in `release-2.3`, where this patch applies cleanly.
Happy to open the
backport if you want it, as was done for FLINK-40324 and FLINK-40334
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Opus 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]
