[
https://issues.apache.org/jira/browse/FLINK-41443?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Anton Borisov updated FLINK-41443:
----------------------------------
Affects Version/s: 2.3.0
> Inner lookup join keeps a stale row when an update moves the join key to a
> row the lookup table doesn't have
> ------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-41443
> URL: https://issues.apache.org/jira/browse/FLINK-41443
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner, Table SQL / Runtime
> Affects Versions: 2.3.0
> Reporter: Anton Borisov
> Priority: Major
>
> For an inner lookup join whose input sends updates without UPDATE_BEFORE (an
> upsert source, or a CDC source whose UPDATE_BEFORE the planner drops because
> the output keeps the input's upsert key), an UPDATE_AFTER that finds no match
> emits nothing. Nothing retracts the joined row of the previous version, so it
> stays in the sink. A later DELETE with the new join key finds no match
> either, so the row is never removed.
> Reproducer:
> {code:sql}
> CREATE TABLE orders (order_id INT PRIMARY KEY NOT ENFORCED, customer_id INT,
> pt AS PROCTIME())
> WITH ('connector' = 'values', 'changelog-mode' = 'I,UB,UA,D');
> CREATE TABLE customers (id INT PRIMARY KEY NOT ENFORCED, name STRING) WITH
> ('connector' = 'values');
> CREATE TABLE sink (order_id INT PRIMARY KEY NOT ENFORCED, name STRING)
> WITH ('connector' = 'values', 'sink-insert-only' = 'false',
> 'sink-changelog-mode-enforced' = 'I,UA,D');
> INSERT INTO sink SELECT o.order_id, c.name FROM orders AS o
> JOIN customers FOR SYSTEM_TIME AS OF o.pt AS c ON o.customer_id = c.id;
> {code}
> Customers: {{(10, Alice)}}. Input: {{+I(1, 10)}}, {{-U(1, 10)}}, {{+U(1,
> 20)}}, {{-D(1, 20)}}. Expected sink content: empty. Actual: {{+I[1, Alice]}}.
> The same happens with an upsert source ({{'changelog-mode' = 'I,UA,D'}}),
> with an async lookup, and with
> {{table.optimizer.non-deterministic-update.strategy = TRY_RESOLVE}}. A left
> lookup join is correct, since it emits a null padded row instead.
> This is the same class of bug that FLINK-40841, FLINK-40842 and FLINK-41066
> fixed in the join operators, here in the lookup join, which keeps no state of
> its input.
> Requiring UPDATE_BEFORE, as FLINK-38579 does for join conditions and lookup
> table filters, would add a {{ChangelogNormalize}} for upsert inputs, or a
> second lookup per update for CDC inputs, to every lookup join on a non-key
> column.
> Instead, when the output keeps the input's upsert key and the input sends no
> UPDATE_BEFORE, an inner lookup join could emit a delete by key for an
> UPDATE_AFTER or DELETE that finds no match or fails the join condition. This
> needs no state and no extra lookup:
> * the changelog inference needs to know that the lookup join can produce
> deletes by key, and keep requiring UPDATE_BEFORE if the consumer needs full
> deletes
> * the compiled plan needs a flag for it
> * the sync and async runners need it, and {{KeyedLookupJoinWrapper}}
> (TRY_RESOLVE) could emit the full joined row from its state
> This would also cover the join conditions and lookup table filters, so the
> planner check from FLINK-38579 could be dropped for consumers that accept
> deletes by key.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
