[
https://issues.apache.org/jira/browse/FLINK-40924?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
sepuri sai krishna updated FLINK-40924:
---------------------------------------
Release Note:
table.exec.sink.require-on-conflict is now also enforced when
table.exec.sink.upsert-materialize
is set to FORCE. A query whose upsert key differs from the sink's primary key
and that specifies
no ON CONFLICT clause previously planned under FORCE and will now fail with the
same
ValidationException already raised under AUTO. Add an ON CONFLICT clause, or set
require-on-conflict to false, to keep such a query planning.
upsert-materialize=NONE is
unaffected, because it removes the sink upsert materializer and therefore
applies no conflict
strategy.
was:
table.exec.sink.require-on-conflict is now also enforced when
table.exec.sink.upsert-materialize is set to NONE or FORCE. Queries whose
upsert key differs from
the sink's primary key and that specify no ON CONFLICT clause previously
planned successfully
under those two settings and will now fail with the same ValidationException
already raised under
AUTO. Set require-on-conflict to false to restore the previous behaviour.
Description:
{{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}}.
was:
{{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 {{NONE}} suppresses that
error, and the query
plans with no materializer.
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' = 'NONE';
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]}} |
| {{NONE}} | plans, {{upsertMaterialize=[false]}} |
Only {{AUTO}} raises the documented error.
h3. Why this looks unintended
The two options document separate concerns, and neither documents this
interaction.
{{require-on-conflict}} documents exactly one way to turn itself off:
{quote}
Set this to false to restore the old behavior where no ON CONFLICT clause was
required. Note that
disabling this check may lead to non-deterministic results in certain streaming
scenarios.
{quote}
{{upsert-materialize}} is described entirely in terms of the materialize
operator and shuffle
disorder, and says nothing about validation or {{ON CONFLICT}}.
So a user who sets {{NONE}} has opted out of materialization, not out of the
check. The
{{FORCE}} case also skips the error but still materializes, so its result stays
deterministic;
{{NONE}} skips the error and does not materialize.
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 at all.
Summary: table.exec.sink.upsert-materialize=FORCE silently bypasses
the require-on-conflict check (was: table.exec.sink.upsert-materialize=NONE
silently bypasses the require-on-conflict check)
> 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
(v8.20.10#820010)
