[jira] [Created] (FLINK-41066) MultiJoin produces wrong results when an upsert input replaces a record without UPDATE_BEFORE

["Gustavo de Morais (Jira)" <ji...@apache.org>]

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)