[
https://issues.apache.org/jira/browse/FLINK-41225?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18134276#comment-18134276
]
Gustavo de Morais commented on FLINK-41225:
-------------------------------------------
Thanks for the reproducers and bug report. The fold comes from the original
mini-batch join bundles (FLINK-34220). FLINK-40809 and FLINK-40831 fixed only
the count. cc [~xu_shuai_] [~lsy] [~xuyang1706] - could you take a look?
> 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
> Priority: Major
>
> 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)
