Anton Borisov created FLINK-41282:
-------------------------------------
Summary: 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
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)
