[
https://issues.apache.org/jira/browse/FLINK-40780?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-40780:
-----------------------------------
Labels: pull-request-available (was: )
> FORWARD edge silently converted to REBALANCE on parallelism mismatch corrupts
> changelog (upsert/retract) streams
> ----------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40780
> URL: https://issues.apache.org/jira/browse/FLINK-40780
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Coordination
> Reporter: Rui Fan
> Assignee: Rui Fan
> Priority: Major
> Labels: pull-request-available
>
> h2. Problem
> When a FORWARD (pointwise) edge connects two operators whose parallelism
> differs, the runtime silently replaces the ForwardPartitioner with a
> REBALANCE (round-robin) partitioner (introduced in FLINK-30213). This is
> correct for append-only streams, but for order-sensitive changelog streams
> (upsert/retract) it silently produces wrong, non-deterministic results — no
> error, no failing job.
> A FORWARD edge preserves the upstream partitioning, and the planner relies on
> this to skip a redundant keyed exchange when the input is already partitioned
> by the required key. Downgrading it to REBALANCE breaks that guarantee:
> records for the same key are scattered across downstream subtasks, so a keyed
> operator or upsert sink no longer sees a key's changelog in order.
> h2. Example
> A streaming global aggregation into an upsert sink:
> INSERT INTO upsert_sink SELECT 1 AS k, COUNT(*) FROM src;
> Compiled at parallelism 1, the whole path (GroupAggregate → Calc → Sink)
> chains via FORWARD — correct. If the sink is later run at a higher
> parallelism than the parallelism-1 aggregate (e.g. via a per-vertex
> parallelism override / rescale), the edge into the sink stays FORWARD but now
> spans 1→N: the single key's changelog is spread round-robin across N writers,
> which interleave writes to the same key and yield a non-deterministic final
> value.
> h2. Proposed fix
> Add a config option controlling the behavior when a FORWARD edge spans
> mismatched parallelism:
> pipeline.forward-edge.parallelism-mismatch-mode (enum)
> - REBALANCE — current behavior (downgrade to REBALANCE). Default
> (backward-compatible).
> - FAIL — reject the job with a clear exception.
> - KEEP_FORWARD — keep the 1-to-1 mapping (order-preserving; the
> pre-FLINK-30213 behavior).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
