Anton Borisov created FLINK-41443:
-------------------------------------
Summary: 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
Reporter: Anton Borisov
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)
