[jira] [Created] (FLINK-40842) Streaming semi/anti join over-counts matches for right-side updates without UPDATE_BEFORE

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

Gustavo de Morais created FLINK-40842:
-----------------------------------------

             Summary: Streaming semi/anti join over-counts matches for right-side updates without UPDATE_BEFORE
                 Key: FLINK-40842
                 URL: https://issues.apache.org/jira/browse/FLINK-40842
             Project: Flink
          Issue Type: Sub-task
          Components: Table SQL / API
            Reporter: Gustavo de Morais
            Assignee: Gustavo de Morais


{{StreamingSemiAntiJoinOperator}} keeps, for every left row, the number of matching right rows. An accumulate message on the right side always increments this count. When the right input is an upsert stream (UPDATE_AFTER without UPDATE_BEFORE), an update that replaces a right row with the same unique key is counted as a second match. A later delete then only decrements back to 1. This is the same bug as FLINK-40809, in the semi/anti join operator.

Example: {{SELECT * FROM L WHERE L.k IN (SELECT R.k FROM R)}} (semi) or {{NOT IN}} / {{NOT EXISTS}} (anti), where R is an upsert stream with PK {{{}(id, k){}}}.


{code:java}
+I L(k=1)
+I R(id=1, k=1, v=a)    count 1   semi: +I[L]    anti: -D[L]
+U R(id=1, k=1, v=b)    count 2   <- should stay 1
-D R(id=1, k=1)         count 1   semi: L is not retracted
                                  anti: L is not emitted again{code}
{{ }}
Expected: the semi join emits {{-D[L]}} and the anti join emits {{+I[L]}} after the delete. Actual: nothing is emitted, and L stays wrong.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)