Martijn Visser created FLINK-40681:
--------------------------------------
Summary: Streaming outer join drops the null-padded row after an
inner-side update followed by delete when UPDATE_BEFORE is not required
Key: FLINK-40681
URL: https://issues.apache.org/jira/browse/FLINK-40681
Project: Flink
Issue Type: Bug
Components: Table SQL / Runtime
Affects Versions: 1.20.5, 2.2.1, 2.3.0, 2.4.0
Reporter: Martijn Visser
When the planner does not require UPDATE_BEFORE on the inputs of a regular
outer join (both inputs have an upsert key that contains the join key and the
sink's primary key is satisfied by the query's upsert key), the inner side
sends a bare UPDATE_AFTER for a changed row, and no ChangelogNormalize sits in
front of the join. {{StreamingJoinOperator}} treats every accumulate message
alike: it overwrites the row in the {{JoinKeyContainsUniqueKey}} state but
still increments {{numOfAssociations}} of every matching outer row (line 274 on
master). The later DELETE of that inner row finds the counter at 2 instead of 1
(line 321) and never emits the null-padded outer row again, so the outer row
disappears from an upsert sink. A repeated INSERT for the same key has the same
effect, so the defect is "an accumulate that replaces a row on a unique-key
side is counted as a new association", not the UPDATE_AFTER kind specifically.
The planner's assumption that the operator can handle upsert input once it can
identify the record is not honored by the outer-join bookkeeping.
Reproducer with two upsert tables and an upsert sink, all keyed on {{k}},
parallelism 1:
{code:sql}
CREATE TABLE left_t (k INT NOT NULL, v STRING, PRIMARY KEY (k) NOT ENFORCED)
WITH ('connector' = 'values', 'changelog-mode' = 'I,UA,D');
CREATE TABLE right_t (k INT NOT NULL, w STRING, PRIMARY KEY (k) NOT ENFORCED)
WITH ('connector' = 'values', 'changelog-mode' = 'I,UA,D');
CREATE TABLE sink (k INT NOT NULL, v STRING, w STRING, PRIMARY KEY (k) NOT
ENFORCED) WITH ('connector' = 'values', 'sink-insert-only' = 'false');
INSERT INTO sink SELECT l.k, l.v, r.w FROM left_t l LEFT JOIN right_t r ON l.k
= r.k;
{code}
left_t receives {{+I(1, a)}}; right_t receives {{+I(1, x)}}, {{+U(1, y)}},
{{-D(1, y)}}. The {{+U}} must arrive after the left row has been counted (if
the left row arrives after the {{+U}} it is stored with the right count and the
delete behaves); the attached ITCase enforces that by gating the right source
on the sink contents. The plan is
{code}
Join(joinType=[LeftOuterJoin], where=[=(k, k0)],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey], changelogMode=[I,UA,D])
:- Exchange(distribution=[hash[k]], changelogMode=[I,UA,D])
: +- TableSourceScan(table=[[default_catalog, default_database, left_t]],
fields=[k, v], changelogMode=[I,UA,D])
+- Exchange(distribution=[hash[k]], changelogMode=[I,UA,D])
+- TableSourceScan(table=[[default_catalog, default_database, right_t]],
fields=[k, w], changelogMode=[I,UA,D])
{code}
The raw changelog the sink receives (parallelism 1, HEAP and ROCKSDB alike):
{code}
+I[1, a, null]
-D[1, a, null]
+I[1, a, x]
+I[1, a, y]
-D[1, a, y]
{code}
Expected is the same sequence followed by {{+I[1, a, null]}}. The materialized
sink content is therefore empty instead of {{[1, a, null]}}, and a later delete
of the left row reaches the sink as a DELETE for a key it no longer holds.
As a control, not a workaround: with the sink primary key changed to {{(k, v)}}
the planner requires UPDATE_BEFORE, adds a ChangelogNormalize in front of each
input, the join sees an explicit {{-U}} that decrements the counter, and the
result is correct.
The same arithmetic is in {{JoinHelper}} (async state) and in
{{StreamingSemiAntiJoinOperator}} (an anti join never re-emits the left row
after {{+I, +U, -D}} on the right side). {{MiniBatchStreamingJoinOperator}} has
a second problem: {{JoinKeyContainsUniqueKeyBundle}} folds an UPDATE_AFTER
followed by a DELETE within one bundle into nothing while the previous version
of the row stays in state, so nothing is emitted at all.
Three existing tests assert the wrong output:
{{StreamingJoinOperatorTest#testLeftOuterJoinWithStateRetentionDisabled}},
{{StreamingSemiAntiJoinOperatorTest#testLeftSemiJoinWithDifferentStateRetentionTime}}
and {{#testLeftSemiJoinWithStateRetentionDisabled}}.
I'll open a draft PR with failing tests for all variants (operator harness for
LEFT/RIGHT/FULL with sync and async state, duplicate INSERT, mini-batch, anti
join; ITCases asserting the raw changelog; plan tests pinning the changelog
modes)
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
