Mate Czagany created FLINK-40997:
------------------------------------
Summary: Streaming join picks a random state key among
equal-length upsert keys
Key: FLINK-40997
URL: https://issues.apache.org/jira/browse/FLINK-40997
Project: Flink
Issue Type: Bug
Components: Table SQL / Planner
Reporter: Mate Czagany
The same SQL can produce different {{leftUpsertKeys}} / {{rightUpsertKeys}}
order on {{StreamExecJoin}} nodes.
If the input has two or more upsert keys that tie for the fewest columns, and
none of them is contained in that side's join key, otherwise
{{JoinKeyContainsUniqueKey}} would be used.
Example:
{code:sql}
CREATE TABLE items (id INT, score INT, brand STRING, PRIMARY KEY (id) NOT
ENFORCED) WITH (...);
CREATE TABLE z (k STRING, z STRING) WITH (...);
INSERT INTO snk
SELECT t.id, t.rn, z.z
FROM (SELECT * FROM (SELECT *, ROW_NUMBER() OVER (ORDER BY score DESC) AS rn
FROM items)
WHERE rn <= 3) t
JOIN z ON t.brand = z.k;
{code}
The left input of the join is {{{}(id, brand, rn){}}}. It has two upsert keys,
{{id}} and {{{}rn{}}}, and the join key {{brand}} contains neither.
After compiling this several times, {{leftUpsertKeys}} of the join switches
between [[0],[2]] (state keyed by {{{}id{}}}) and [[2],[0]] (keyed by
{{{}rn{}}}). This can break state restore using a previous snapshot with a
{{{}StateMigrationException{}}}.
With id {{BIGINT}} instead of {{{}INT{}}}, both keys are {{{}BIGINT NOT
NULL{}}}, a state restore can succeed. Retractions then look up the wrong key,
and stale rows stay in the join state and keep joining. No error is reported.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
