Fabian Hueske created FLINK-40477:
-------------------------------------
Summary: 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
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)
