Gustavo de Morais created FLINK-41066:
-----------------------------------------
Summary: MultiJoin produces wrong results when an upsert input
replaces a record without UPDATE_BEFORE
Key: FLINK-41066
URL: https://issues.apache.org/jira/browse/FLINK-41066
Project: Flink
Issue Type: Sub-task
Components: Table SQL / API
Affects Versions: 2.2.1, 2.3.0
Reporter: Gustavo de Morais
StreamingMultiJoinOperator can receive inputs with UPDATE_AFTER but without
UPDATE_BEFORE when the input's unique key contains the common join key. A
record that replaces the stored record with the same unique key only emits its
own joined rows and then overwrites the state. Nothing retracts what the old
version produced:
* With a non-equi condition, or an equi condition on an attribute that a later
input joins on, the old joined rows stay, and the null padding of LEFT joins is
not restored.
* If the update changes the input's own join key beyond the common key, the old
record stays in state under its old key and keeps joining.
* A delete that only carries the unique key is joined and removed with its null
fields, so it retracts nothing and the record stays in state.
```
SELECT * FROM A
JOIN B ON a.id = b.id
JOIN C ON a.id = c.id AND b.v = c.v
-- C is an upsert input with PRIMARY KEY (id, cid); an update of c.v without
UPDATE_BEFORE
-- keeps the old C row in state and its joined row in the output
```
We should look up the stored record with the same unique key, retract the
joined rows and null padding that only the old version produces, and join
retractions with the stored record.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
