[
https://issues.apache.org/jira/browse/FLINK-41258?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18133965#comment-18133965
]
Rion Williams commented on FLINK-41258:
---------------------------------------
Hey [~gustavodemorais]!
This was a follow up of one of our many, many JOIN-related improvements. Feel
free to assign to me, if you can, and I had one quick question as I was
planning this one out. I figured I'd ask here to reduce the churn in the PR.
In the regular join, a tombstone DELETE for a key that was never inserted still
matches the other side and emits a {{-D}} for a joined row we never produced.
The semi/anti trick (skipping the lookup when the join key contains the unique
key) doesn't carry over: there, a count of 0 means nothing is emitted, but the
regular join emits a row for every match whatever the counts are.
Which would you prefer?
- *Look up the stored record for every DELETE, only on known-upsert inputs
with a unique key.* That's one extra point read per DELETE, and it fixes the
tombstone case. The first case of {{testLeftOuterJoinDeletesWithoutLookup}}
would then expect the full stored row instead of the key-only nulls.
- {*}Keep the no-lookup path{*}, which leaves the stray {{-D}} for tombstones
of keys that were never inserted
> Streaming join miscounts matches on DELETEs of records that are not stored
> --------------------------------------------------------------------------
>
> Key: FLINK-41258
> URL: https://issues.apache.org/jira/browse/FLINK-41258
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Runtime
> Reporter: Rion Williams
> Priority: Major
>
> In {{StreamingJoinOperator}}, with an upsert input:
> * A DELETE of a record that was never inserted, such as a tombstone, is still
> joined. When the outer side's join key contains the unique key of the other
> side, the outer record's number of matches can go negative. Since
> {{isNewMatch}} doesn't increase a non-zero count, the record then never gets
> its null padding back or matches correctly.
> * A key-only DELETE of a record that is not stored is joined with its null
> fields. A NULL-tolerant non-equi condition (for example {{r.v IS NULL OR
> ...}}) can then match it and retract rows it never produced.
> h3. Proposed approach
> As in FLINK-40842 for the semi/anti join: for inputs whose changelog mode is
> known to be upsert, skip such DELETEs, or treat them as matching nothing.
> Plans compiled before 2.4 keep the current handling, because a missing record
> there may have expired under state TTL.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
