[
https://issues.apache.org/jira/browse/FLINK-40468?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Fabian Hueske reassigned FLINK-40468:
-------------------------------------
Assignee: Fabian Hueske
> Partial deletes cause NPE if value type is declared as NOT NULL
> ---------------------------------------------------------------
>
> Key: FLINK-40468
> URL: https://issues.apache.org/jira/browse/FLINK-40468
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.3.0
> Reporter: Fabian Hueske
> Assignee: Fabian Hueske
> Priority: Critical
>
> Partial deletes are an encoding format for upsert changelogs that encodes
> deletes only with (composite) key fields and all value fields as `NULL`.
> This is beneficial because 1) less data needs to be serialized and 2) some
> sources like Kafka internally encode deletes like this and converting to full
> deletes (incl. all value fields) requires a lot of state and is expensive.
> However, if data types are declared as NOT NULL, Flink's serializers do not
> expect NULL values and might fail if they process a partial delete.
> Test case to reproduce the issue:
> {code:java}
> @Test
> void testUpsertPartialDeleteIntoRowConstructor() throws Exception {
> final String id =
> TestValuesTableFactory.registerData(
> Arrays.asList(
> TestValuesTableFactory.changelogRow("+U", 1, new
> Integer[]{1, 2}),
> TestValuesTableFactory.changelogRow("+U", 2, new
> Integer[]{3}),
> // tombstone: key only, NOT NULL value column is
> null
> TestValuesTableFactory.changelogRow("-D", 1,
> null)));
> tEnv().executeSql(
> "CREATE TABLE UpsertSrc (k INT, v ARRAY<INT> NOT NULL, "
> + "PRIMARY KEY (k) NOT ENFORCED) WITH ("
> + "'connector' = 'values', 'data-id' = '"
> + id
> + "', 'changelog-mode' = 'UA,D')");
> tEnv().executeSql(
> "CREATE TABLE UpsertSnk (k INT, r ROW<a INT, b ARRAY<INT>>, "
> + "PRIMARY KEY (k) NOT ENFORCED) WITH ("
> + "'connector' = 'values', 'sink-insert-only' =
> 'false')");
> tEnv().executeSql("INSERT INTO UpsertSnk SELECT k, ROW(k, v) FROM
> UpsertSrc").await();
> // k=1 was inserted then deleted by the tombstone; k=2 survives.
> assertThat(TestValuesTableFactory.getResultsAsStrings("UpsertSnk"))
> .containsExactlyInAnyOrder("+I[2, +I[2, [3]]]");
> } {code}
> It's not clear yet what's the best approach to fix this issue.
> A very defensive approach would be to require full delete if any value field
> is NOT NULL. However, this might result in significant performance
> regressions.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
