[
https://issues.apache.org/jira/browse/FLINK-40477?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40477:
-----------------------------------
Labels: pull-request-available (was: )
> Partial deletes fail sink NOT NULL enforcement
> ----------------------------------------------
>
> Key: FLINK-40477
> URL: https://issues.apache.org/jira/browse/FLINK-40477
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
> Affects Versions: 2.3.0
> Reporter: Fabian Hueske
> Assignee: Fabian Hueske
> Priority: Critical
> Labels: pull-request-available
>
> In a partial-delete (key-only) pipeline, a -D tombstone carries the key and
> sets all non-key columns to null, "regardless of nullability constraints"
> (see ChangelogFunction / ChangelogMode#keyOnlyDeletes).
> When the sink has a NOT NULL value (non-key) column, the sink constraint
> enforcer (NotNullConstraint) inspects the tombstone and reacts to that null:
> * table.exec.sink.not-null-enforcer = ERROR (default): throws
> EnforcerException → the job fails on every by-key delete (crash-restart loop).
> * table.exec.sink.not-null-enforcer = DROP: the tombstone is silently
> dropped → the row is never deleted (stale data, no error).
> Root cause: NotNullConstraint#enforce (module flink-table-runtime, package
> ...operators.sink.constraint) checks input.isNullAt(index) for every row
> regardless of RowKind; it has no notion that a DELETE in a key-only-delete
> pipeline legitimately carries null non-key columns.
> This is distinct from FLINK-40468 (composite serializer NPE): that requires a
> null nested inside a composite constructor, whereas this affects any NOT NULL
> top-level value column and is caught earlier, at the enforcer.
> How to reproduce (scalar INT column, so no composite/serializer path is
> involved):
>
> {code:java}
> -- source produces key-only deletes; sink accepts key-only deletes
> CREATE TABLE src (id INT PRIMARY KEY NOT ENFORCED, v INT NOT NULL) WITH (
> 'connector' = 'values',
> 'changelog-mode' = 'I,UA,D',
> 'source.produces-delete-by-key' = 'true', ...);
> CREATE TABLE snk (id INT PRIMARY KEY NOT ENFORCED, v INT NOT NULL) WITH (
> 'connector' = 'values', 'sink-insert-only' = 'false',
> 'sink.supports-delete-by-key' = 'true');
> INSERT INTO snk SELECT id, v FROM src;
> -- Source data: +I(1, 10), +I(2, 20), -D(1, null) (tombstone: key only, v
> null). {code}
>
> * Expected: the tombstone is applied by key; final sink state is \{(2, 20)}.
> * Actual (ERROR): EnforcerException: Column 'v' is NOT NULL, however, a null
> value is being written into it.
> * Actual (DROP): tombstone dropped; final sink state is \{(1, 10), (2, 20)}
> (stale row 1).
> Suggested fix: make the enforcer skip the NOT NULL check for the non-key
> columns of a DELETE row when the input is a key-only-delete pipeline
> (ChangelogMode#keyOnlyDeletes). Key columns and non-DELETE rows stay fully
> enforced; full-delete pipelines are unaffected.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
