[
https://issues.apache.org/jira/browse/FLINK-41282?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Anton Borisov updated FLINK-41282:
----------------------------------
Description:
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]).
was:
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).
> 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
>
> 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)
