[
https://issues.apache.org/jira/browse/FLINK-40924?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Yuepeng Pan updated FLINK-40924:
--------------------------------
Fix Version/s: (was: 2.5.0)
> table.exec.sink.upsert-materialize=FORCE silently bypasses the
> require-on-conflict check
> ----------------------------------------------------------------------------------------
>
> Key: FLINK-40924
> URL: https://issues.apache.org/jira/browse/FLINK-40924
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.3.0, 2.4.0
> Reporter: sepuri sai krishna
> Assignee: sepuri sai krishna
> Priority: Major
> Labels: pull-request-available
>
> {{table.exec.sink.require-on-conflict}} defaults to {{true}} and is
> documented to throw when the
> query's upsert key differs from the sink's primary key and no {{ON CONFLICT}}
> clause is given.
> Setting {{table.exec.sink.upsert-materialize}} to {{FORCE}} suppresses that
> error, and the query
> plans with {{upsertMaterialize=[true]}}.
> h3. Reproducer
> The upsert key is the grouping key {{c}}, which is never written to the sink,
> so it cannot equal
> the primary key {{x}}.
> {code: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;
> {code}
> With {{require-on-conflict}} left at its default of {{true}}:
> || upsert-materialize || result ||
> | {{AUTO}} (default) | {{ValidationException}}: "The query has an upsert key
> that differs from the primary key of the sink table ... Please specify an ON
> CONFLICT clause" |
> | {{FORCE}} | plans, {{upsertMaterialize=[true]}} |
> The check is called from inside the {{AUTO}} branch of
> {{analyzeUpsertMaterializeStrategy}} in
> {{FlinkChangelogModeInferenceProgram}}, so the other settings never reach it.
> h3. Why FORCE should be covered
> {{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 built only on that
> path. So the clause does
> take effect under {{FORCE}}, and FLIP-558 states the requirement without
> tying it to a
> materialization setting:
> {quote}
> 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.
> {quote}
> Nothing documents {{FORCE}} as a way to opt out. The documented ways are
> {{require-on-conflict=false}} and {{upsert-materialize=NONE}}.
> h3. Why NONE is not part of this
> {{NONE}} makes {{analyzeUpsertMaterializeStrategy}} return {{false}}, so no
> materializer is built
> and an {{ON CONFLICT}} clause has no effect. Requiring one would ask for a
> clause that is then
> ignored. It is also documented as the way to opt out entirely, in
> {{docs/content/docs/sql/reference/dml/insert.md}} under "When is ON CONFLICT
> required?":
> {quote}
> Alternatively, if you do not need consistency guarantees for conflicting
> keys, you can disable the
> sink upsert materializer entirely by setting
> table.exec.sink.upsert-materialize to NONE. This
> removes the materializer operator from the pipeline, so no buffering,
> compaction, or conflict
> resolution is performed. Records are passed directly to the sink as they
> arrive.
> {quote}
> So {{NONE}} stays as it is.
> h3. Affects
> {{master}} and {{release-2.3}}, where the relevant code is unchanged.
> Related to FLINK-40899, which reports the opposite direction on the same
> validation: there it
> fires too early and masks an NDU error. This one is the check not firing
> under {{FORCE}}.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
