[
https://issues.apache.org/jira/browse/FLINK-41443?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18144522#comment-18144522
]
Gustavo de Morais commented on FLINK-41443:
-------------------------------------------
Hey [~aborisov], thanks for the ticket. Agree it's a bug and it's somewhat
similar to the other ones. The difference here is that the lookup join doesn't
have stay so we can't easily consider a runtime fix.
How I see it: an inner lookup join is a stateless filter on the join key. Since
there's no shuffle, it keeps the input upsert key (\{{order_id}}), so the
planner drops UPDATE_BEFORE and assumes every update overwrites the previous
row. A miss emits nothing, so nothing gets overwritten.
We already have a rule for this: a filter on a column outside the upsert key
requires UPDATE_BEFORE (FLINK-9528 for Calc, FLINK-38579 for join conditions
and lookup table filters). I'd extend the same rule to the join key: an inner
lookup join whose join key isn't contained in the input upsert key requires
UPDATE_BEFORE. It costs a ChangelogNormalize or a second lookup, but that's the
same cost we already pay for a WHERE on a non-key column. If we add state to
the lookup join so we can fetch the past records, it's essentially doing the
same thing but only for one operator instead of doing it for the whole pipeline
once.
I'm not a fan of delete by key as the default: we will emit ghost deletes for
keys that were never emitted. It doesn't feel right. For an inner lookup join
used as a filter, every missed update becomes a tombstone. Correctness then
depends on every consumer tolerating deletes for unknown keys.
About TRY_RESOLVE: In general, TRY_RESOLVE is about determinism and solves a
different problem than the one we have here. KeyedLookupJoinWrapper isn't even
added for this query.
Wdyt? Happy if other people want to chime in
> 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)
