Yoni created FLINK-41225:
----------------------------
Summary: Mini-batch join: the unique-key bundles fold an
UPDATE_AFTER and the following DELETE into nothing, so the stored row is never
retracted
Key: FLINK-41225
URL: https://issues.apache.org/jira/browse/FLINK-41225
Project: Flink
Issue Type: Bug
Components: Table SQL / Runtime
Affects Versions: 2.2.1
Reporter: Yoni
FLINK-40681 describes this as the mini-batch join's second problem, next to the
association count. Its sub-tasks FLINK-40809 and FLINK-40831 fixed the count,
and none of its sub-tasks covers the fold. The bundles are unchanged on master.
This issue tracks the fold on its own, with deterministic reproducers, and is
linked to FLINK-40681. FLINK-40681 names only
{{JoinKeyContainsUniqueKeyBundle}}; the second reproducer below shows that
{{InputSideHasUniqueKeyBundle}} is affected too. cc [~gustavodemorais]
When nothing downstream requires UPDATE_BEFORE, for example an upsert sink
keyed on the query's upsert key, the planner puts no ChangelogNormalize in
front of a regular join's input that has a unique key. The join then reads
upsert input {{\[I,UA,D\]}} directly, so an accumulate can replace a row the
join already holds: a bare UPDATE_AFTER, or a repeated INSERT for the same key.
Upsert sources such as upsert-kafka produce exactly this. Both unique-key input
specs are affected:
* {{JoinKeyContainsUniqueKey}}: the join key contains the input's unique key
({{JoinKeyContainsUniqueKeyBundle}}).
* {{HasUniqueKey}}: the unique key is wider than the join key, as in
FLINK-40842's example ({{InputSideHasUniqueKeyBundle}}).
In mini-batch mode, both bundles' {{foldRecord}} apply the rule "+I/+U followed
by -U/-D: clear both". That is correct only when the accumulate inserted a row
that was not in state before. When it replaced a stored row, clearing the pair
removes that row's only retraction:
* the stored row stays in the join state;
* the joined row it produced stays in the sink;
* nothing later corrects it.
Both a bare {{+U}} and a repeated {{+I}} followed by {{-D}} reproduce it.
Semi and anti joins are not affected by the fold: {{StreamExecJoin}} builds
them as {{StreamingSemiAntiJoinOperator}}, which has no bundle, even when the
plan shows {{miniBatch=\[true\]}}. Their count issue is FLINK-40842.
h3. Reproducer 1: JoinKeyContainsUniqueKey
Stock Flink 2.2.1, {{values}} connector, parallelism 1. {{allow-latency}} is
set high so that only {{mini-batch.size}} closes bundles. Size 2 puts the first
two left records in bundle 1, and the update and delete in bundle 2.
Deterministic: 10 of 10 runs at each size gave the results below.
{code:sql}
SET 'parallelism.default' = '1';
SET 'table.exec.mini-batch.enabled' = 'true';
SET 'table.exec.mini-batch.allow-latency' = '1 h';
SET 'table.exec.mini-batch.size' = '2';
CREATE TABLE left_t (k INT NOT NULL, v STRING, PRIMARY KEY (k) NOT ENFORCED)
WITH ('connector' = 'values', 'changelog-mode' = 'I,UA,D', 'data-id' =
'<left>');
CREATE TABLE right_t (k INT NOT NULL, w STRING, PRIMARY KEY (k) NOT ENFORCED)
WITH ('connector' = 'values', 'changelog-mode' = 'I,UA,D', 'data-id' =
'<right, empty>');
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}
The {{data-id}}s come from {{TestValuesTableFactory.registerData}}, so the SQL
runs inside a test, for example:
{code:java}
String left = TestValuesTableFactory.registerData(List.of(
Row.ofKind(RowKind.INSERT, 1, "a"),
Row.ofKind(RowKind.INSERT, 2, "c"),
Row.ofKind(RowKind.UPDATE_AFTER, 1, "b"), // RowKind.INSERT folds the
same way
Row.ofKind(RowKind.DELETE, 1, "b")));
String right = TestValuesTableFactory.registerData(List.of());
{code}
Plan ({{EXPLAIN CHANGELOG_MODE}}). There is no ChangelogNormalize; the join
reads {{\[I,UA,D\]}}:
{noformat}
Sink(table=[default_catalog.default_database.sink], fields=[k, v, w],
changelogMode=[NONE])
+- Join(joinType=[LeftOuterJoin], where=[=(k, k0)], select=[k, v, k0, w],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey], miniBatch=[true],
changelogMode=[I,UA,D])
:- Exchange(distribution=[hash[k]], changelogMode=[I,UA,D])
: +- MiniBatchAssigner(interval=[3600000ms], mode=[ProcTime],
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])
+- MiniBatchAssigner(interval=[3600000ms], mode=[ProcTime],
changelogMode=[I,UA,D])
+- TableSourceScan(table=[[default_catalog, default_database,
right_t]], fields=[k, w], changelogMode=[I,UA,D])
{noformat}
Result:
||mini-batch.size||sink changelog||final sink||
|2|+I\[2, c, null\], +I\[1, a, null\]|\[1, a, null\], \[2, c, null\]: *key 1 is
never retracted, and keeps the old value a*|
|1 (no folding)|+I\[1, a, null\], +I\[2, c, null\], +I\[1, b, null\], -D\[1, b,
null\]|\[2, c, null\]: correct|
Expected with size 2: the same final result as with size 1.
h3. Reproducer 2: HasUniqueKey (InputSideHasUniqueKeyBundle)
Same settings, size 2. {{wide_t}} has a unique key {{(id, k)}} that is wider
than the join key {{k}}, and the sink is keyed on {{(id, k)}}. {{narrow_t}} is
empty, so only {{wide_t}}'s records fill the bundles.
{code:sql}
CREATE TABLE wide_t (id INT NOT NULL, k INT NOT NULL, w STRING, PRIMARY KEY
(id, k) NOT ENFORCED)
WITH ('connector' = 'values', 'changelog-mode' = 'I,UA,D', 'data-id' =
'<wide>');
CREATE TABLE narrow_t (k INT NOT NULL, v STRING, PRIMARY KEY (k) NOT ENFORCED)
WITH ('connector' = 'values', 'changelog-mode' = 'I,UA,D', 'data-id' =
'<narrow, empty>');
CREATE TABLE sink2 (id INT NOT NULL, k INT NOT NULL, w STRING, v STRING,
PRIMARY KEY (id, k) NOT ENFORCED)
WITH ('connector' = 'values', 'sink-insert-only' = 'false');
INSERT INTO sink2 SELECT r.id, r.k, r.w, l.v FROM wide_t r LEFT JOIN narrow_t l
ON l.k = r.k;
{code}
{{wide_t}} data, in order: {{+I(10, 1, x)}}, {{+I(20, 2, z)}}, {{+U(10, 1,
y)}}, {{-D(10, 1, y)}}.
{noformat}
Sink(table=[default_catalog.default_database.sink2], fields=[id, k, w, v],
changelogMode=[NONE])
+- Calc(select=[id, k, w, v], changelogMode=[I,UA,D])
+- Join(joinType=[LeftOuterJoin], where=[=(k0, k)], select=[id, k, w, k0,
v], leftInputSpec=[HasUniqueKey], rightInputSpec=[JoinKeyContainsUniqueKey],
miniBatch=[true], changelogMode=[I,UA,D])
:- Exchange(distribution=[hash[k]], changelogMode=[I,UA,D])
: +- MiniBatchAssigner(interval=[3600000ms], mode=[ProcTime],
changelogMode=[I,UA,D])
: +- TableSourceScan(table=[[default_catalog, default_database,
wide_t]], fields=[id, k, w], changelogMode=[I,UA,D])
+- Exchange(distribution=[hash[k]], changelogMode=[I,UA,D])
+- MiniBatchAssigner(interval=[3600000ms], mode=[ProcTime],
changelogMode=[I,UA,D])
+- TableSourceScan(table=[[default_catalog, default_database,
narrow_t]], fields=[k, v], changelogMode=[I,UA,D])
{noformat}
With size 2, 10 of 10 runs give the sink changelog {{+I\[20, 2, z, null\]}},
{{+I\[10, 1, x, null\]}}, and the final sink {{\[10, 1, x, null\]}}, {{\[20, 2,
z, null\]}}: *row (10, 1) is never retracted, and keeps the old value x*.
Expected: {{\[20, 2, z, null\]}} only.
h3. Impact
In our deployment this showed up as rows for deleted keys that stay in the
output forever, with no error.
* *Trigger:* the update and the delete of a key land in one bundle. In steady
state that happens when they arrive within {{allow-latency}}; ours were 22 ms
apart, delivered by one upsert-kafka poll.
* *Much more likely after a checkpoint restore or a stateless rebuild,* because
the catch-up packs records into full bundles.
* *Workarounds:*
** {{table.exec.mini-batch.enabled = false}} removes the fold, but it gives up
mini-batch for the whole job.
** {{table.exec.mini-batch.size = 1}} also removes it. It keeps mini-batch's
operators and overhead without any batching.
** Rewrite such a join as a {{GROUP BY}} over a {{UNION ALL}}. An aggregate
requires UPDATE_BEFORE, so the ChangelogNormalize stays. This also avoids the
association-count bug on versions without FLINK-40809 and FLINK-40831.
h3. Suggested fix
The bundle cannot tell whether an accumulate inserted a new row or replaced a
stored one. The planner knows at plan time whether an input carries
UPDATE_BEFORE. Two options:
* *In the operator:* when the input does not carry UPDATE_BEFORE, do not fold
an accumulate with the retraction that follows it; process both, in order. The
planner can pass that flag to {{MiniBatchStreamingJoinOperator}}.
* *In the planner:* require UPDATE_BEFORE, keeping the ChangelogNormalize, for
a mini-batch join input with {{JoinKeyContainsUniqueKey}} or {{HasUniqueKey}}.
Folding the pair down to only the retraction is NOT safe. It breaks the case
the fold exists for, a fresh {{+I}} followed by {{-D}}: a lone retraction on
the outer side emits a null-padded DELETE even when no row was stored, which is
a spurious DELETE downstream.
Clearing both remains correct when the input does produce UPDATE_BEFORE. The
bundle then also holds the stored row's {{-U}}, which survives the fold.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
