[
https://issues.apache.org/jira/browse/FLINK-41282?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18135260#comment-18135260
]
Anton Borisov edited comment on FLINK-41282 at 10/30/26, 3:46 PM:
------------------------------------------------------------------
[~gustavodemorais] [~snuyanzin] could you take a look?
It fixes the multi-join upsert keys added in FLINK-38401.
was (Author: aborisov):
[~gustavodemorais] [~snuyanzin] could you take a look?
It fixes the multi-join upsert keys from #27023 (FLINK-41282).
> Multi-join upsert key inference reports keys that are not unique
> ----------------------------------------------------------------
>
> Key: FLINK-41282
> URL: https://issues.apache.org/jira/browse/FLINK-41282
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.2.2, 2.3.1, 2.4.0, 2.4.1
> Reporter: Anton Borisov
> Priority: Major
> Labels: pull-request-available
>
> With table.optimizer.multi-join.enabled,
> FlinkRelMdUpsertKeys#getUpsertKeys(StreamPhysicalMultiJoin)
> can report an upsert key that is not unique in the output. The sink then
> skips the
> SinkUpsertMaterializer and doesn't ask for ON CONFLICT, and the sink ends up
> with wrong results.
> h3. Cause
> The method folds the inputs left to right, but uses getJoinKeyIndices(0) as
> the left join key at
> every step. MultiJoinUtil#createJoinAttributeMap adds an entry for each join
> column of input 0, so
> that is every column input 0 uses in any condition, not the columns joined to
> input i. At step 1
> the left side then counts as unique if input 0 is unique on columns that are
> only joined to a
> later input.
> h3. Repro
> {code:sql}
> SET 'table.optimizer.multi-join.enabled' = 'true';
> CREATE TABLE t0 (k STRING, y STRING, w0 STRING, PRIMARY KEY (k, y) NOT
> ENFORCED) WITH (...);
> CREATE TABLE t1 (k1 STRING, b1 STRING, w1 STRING, PRIMARY KEY (k1, b1) NOT
> ENFORCED) WITH (...);
> CREATE TABLE t2 (k2 STRING, y2 STRING, w2 STRING, PRIMARY KEY (k2, y2) NOT
> ENFORCED) WITH (...);
> CREATE TABLE snk (k1 STRING, b1 STRING, w0 STRING, w2 STRING,
> PRIMARY KEY (k1, b1) NOT ENFORCED) WITH (...);
> INSERT INTO snk
> SELECT t1.k1, t1.b1, t0.w0, t2.w2
> FROM t0
> JOIN t1 ON t0.k = t1.k1
> JOIN t2 ON t0.k = t2.k2 AND t0.y = t2.y2;
> {code}
> Data: t0 = +I(1, a, x), +I(1, b, y), -D(1, a, x); t1 = +I(1, B, z); t2 =
> +I(1, a, p), +I(1, b, q).
> The correct result is (1, B, y, q).
> On master, with the values connector:
> * Binary joins (multi-join off): planning fails asking for ON CONFLICT,
> since no upsert key
> matches the sink PK. With ON CONFLICT DO DEDUPLICATE the materializer is
> added and the sink
> ends with (1, B, y, q). Correct.
> * Multi-join on: the MultiJoin reports \{k1, b1} as an upsert key, so the
> query plans without
> ON CONFLICT and without the materializer. The sink receives
> +I(1, B, x, p), +I(1, B, y, q), -D(1, B, x, p) and ends empty. Same with ON
> CONFLICT DO DEDUPLICATE.
> A t1 row joins every t0 row with the same k, so \{k1, b1} is not unique in
> the output.
> h3. Suggested fix
> At step i, use the columns of inputs 0..i-1 that input i's conditions
> reference (the left side of
> joinAttributeMap.get (i)) as the left join key, instead of input 0's join
> columns.
> Found while working on FLINK-38753
> ([https://github.com/apache/flink/pull/29653]).
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
