[
https://issues.apache.org/jira/browse/FLINK-40997?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18127756#comment-18127756
]
wind.wang commented on FLINK-40997:
-----------------------------------
We hit this in production on Flink 1.20.5 (application mode on
flink-kubernetes-operator, RocksDB incremental checkpoints, Paimon 1.3.1
sources/sinks), so it also affects 1.20.x.
*Shape of the query*
{code:sql}
-- left input of the join
SELECT order_id, turn_no, order_no, CAST(turn_no AS STRING) AS turn_no0
FROM (SELECT order_id, turn_no, order_no FROM t GROUP BY order_id, turn_no,
order_no)
-- joined as
... LEFT JOIN r ON turn_no0 = r.turn_no
{code}
The left side has two upsert keys of the same length, [0,1,2] = (order_id,
turn_no VARCHAR(64), order_no) and [0,2,3] = (order_id, order_no, turn_no0
VARCHAR(2147483647)); the join key [3] contains neither, so
{{OuterJoinRecordStateViews$InputSideHasUniqueKey}} is used.
*What happened*
* Two fresh (stateless) submissions of the job created state keyed by [0,1,2].
We verified this from the savepoints' {{_metadata}}: the left-records MapState
key serializer is {{RowData(VARCHAR(2147483647) NOT NULL, VARCHAR(64),
VARCHAR(10))}}.
* Restoring the same SQL from those savepoints (twice, hours apart) compiled to
{{leftUpsertKeys=[[0,2,3],[0,1,2]]}}, picked [0,2,3], and failed with
{noformat}
StateMigrationException: The new state serializer (MapSerializer) must not be
incompatible with the old state serializer (MapSerializer)
at OuterJoinRecordStateViews$InputSideHasUniqueKey.<init>
at StreamingJoinOperator.open
{noformat}
* Five further compilations of the same SQL (sql-client, different schema
versions of an unrelated source table, parallelism 4 and 16) all gave
[[0,2,3],[0,1,2]], so in our case the order looked stable within one
environment and only differed from the submissions that wrote the state. We
could not identify what made those submissions differ.
* Of the 27 joins in this job, Join[14] was the only one with tied upsert keys,
and the only operator that failed to restore.
*Workaround that worked*
# COMPILE PLAN the same SQL, swap the {{leftUpsertKeys}} order of that join
node to [[0,1,2],[0,2,3]].
# Run the job with {{EXECUTE PLAN}} and {{table.exec.uid.generation =
DISABLED}} so that operator IDs stay identical to the original SQL job.
# Restore from the existing savepoint. All 496 tasks came up and checkpoints
completed. A later stop-with-savepoint and restore of the plan-based job also
worked.
Side note: restoring the plan failed with "Table options do not contain an
option key 'connector'" for Paimon catalog tables. We worked around it by
adding {{connector=paimon}} and {{path}} to the persisted table options in the
plan; {{table.plan.restore.catalog-objects=IDENTIFIER}} did not help. We have
not checked whether that part is a Flink or Paimon issue.
+1 for making the tie-break deterministic, or persisting the chosen state-key
spec in the plan or state, so that a plain SQL resubmission restores safely.
The silent-corruption case in the description (both keys have the same types)
worries us more than the exception.
> Regular 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
> Priority: Major
> Labels: pull-request-available
>
> 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)
