[
https://issues.apache.org/jira/browse/FLINK-40528?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40528:
-----------------------------------
Labels: pull-request-available (was: )
> Codegen should be tolerate to partial deletes
> ---------------------------------------------
>
> Key: FLINK-40528
> URL: https://issues.apache.org/jira/browse/FLINK-40528
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner, Table SQL / Runtime
> Reporter: Sergey Nuyanzin
> Assignee: Sergey Nuyanzin
> Priority: Major
> Labels: pull-request-available
>
> The problem is that there is partial deletes which might lead to {{null}}
> records without paying attention whether schema allows it or not.
> example of semantic test leading to NPE in such cases (especially for the
> case of nested structure)
> {code:java}
> public static final TableTestProgram
> INSERT_SELECT_DELETE_BY_KEY_WITH_NESTED_NOT_NULL_ROW =
> TableTestProgram.of(
>
> "select-delete-on-key-to-delete-on-key-with-nested-not-null-row",
> "No ChangelogNormalize: a delete-by-key tombstone
> carries null for a NOT"
> + " NULL ROW column wrapped in a ROW(...)
> projection")
> .setupTableSource(
> SourceTestStep.newBuilder("source_t")
> .addSchema(
> "id INT PRIMARY KEY NOT ENFORCED",
> "nested ROW<x INT, y INT> NOT
> NULL")
> .addOption("changelog-mode", "I,UA,D")
>
> .addOption("source.produces-delete-by-key", "true")
> .producedValues(
> Row.ofKind(RowKind.INSERT, 1,
> Row.of(1, 10)),
> Row.ofKind(RowKind.INSERT, 2,
> Row.of(2, 20)),
> // Delete by key: NOT NULL row
> column is null
> Row.ofKind(RowKind.DELETE, 1,
> null),
> // Update after only
> Row.ofKind(RowKind.UPDATE_AFTER,
> 2, Row.of(2, 30)))
> .build())
> .setupTableSink(
> SinkTestStep.newBuilder("sink_t")
> .addSchema(
> "id INT PRIMARY KEY NOT ENFORCED",
> "r ROW<a INT, b ROW<x INT, y
> INT>>")
> .addOption("changelog-mode", "I,UA,D")
> .addOption("sink.supports-delete-by-key",
> "true")
> .consumedValues(
> "+I[1, +I[1, +I[1, 10]]]",
> "+I[2, +I[2, +I[2, 20]]]",
> "-D[1, +I[1, null]]",
> "+U[2, +I[2, +I[2, 30]]]")
> .build())
> .runSql("INSERT INTO sink_t SELECT id, ROW(id, nested)
> FROM source_t")
> .build();
> {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
