[jira] [Created] (FLINK-40468) Partial deletes cause NPE if value type is declared as NOT NULL

[Fabian Hueske (Jira)]

Fabian Hueske created FLINK-40468:
-------------------------------------

             Summary: 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


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)