[
https://issues.apache.org/jira/browse/FLINK-41440?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Timo Walther updated FLINK-41440:
---------------------------------
Affects Version/s: 2.0.0
> Invalid InputProperty for various StreamExecNodes
> -------------------------------------------------
>
> Key: FLINK-41440
> URL: https://issues.apache.org/jira/browse/FLINK-41440
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / Planner
> Affects Versions: 2.0.0
> Reporter: Timo Walther
> Assignee: Timo Walther
> Priority: Critical
>
> h2. Problem
> Some streaming SQL operators have no key and must run at a single parallel
> instance:
> * an interval join without equi-join keys
> * {{MATCH_RECOGNIZE}} without {{PARTITION BY}}
> * {{ORDER BY}} on a time attribute (temporal sort)
> The optimizer requests a SINGLETON distribution for their inputs and inserts
> {{Exchange(distribution=[single])}} (a {{{}GlobalPartitioner{}}}) in front of
> them. The operator inherits parallelism 1 from that exchange. However, {*}its
> max parallelism is not set to 1{*}, so it keeps the job-wide default
> ({{{}pipeline.max-parallelism{}}}).
> Nothing then stops the vertex from being rescaled, for example by:
> * the adaptive scheduler
> * {{pipeline.jobvertex-parallelism-overrides}}
> * resource requirement updates through the REST API
> After a rescale, the {{GlobalPartitioner}} still routes every record to
> subtask 0. The operator is keyed by the empty key, though, and the empty
> key's key group is generally owned by a different subtask. The job fails on
> the first record:
> {noformat}
> java.lang.IllegalArgumentException: Key group 24843 is not in
> KeyGroupRange{startKeyGroup=0, endKeyGroup=16383}.
> Unless you're directly using low level state access APIs, this is most likely
> caused by non-deterministic shuffle key (hashCode and equals implementation).
> at
> org.apache.flink.runtime.state.KeyGroupRangeOffsets.newIllegalKeyGroupException(KeyGroupRangeOffsets.java)
> at
> org.apache.flink.runtime.state.InternalKeyContextImpl.setCurrentKeyGroupIndex(InternalKeyContextImpl.java)
> at
> org.apache.flink.runtime.state.AbstractKeyedStateBackend.setCurrentKey(AbstractKeyedStateBackend.java)
> at
> org.apache.flink.streaming.api.operators.StreamOperatorStateHandler.setCurrentKey(StreamOperatorStateHandler.java)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.setCurrentKey(AbstractStreamOperator.java)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.setKeyContextElement1(AbstractStreamOperator.java)
> ...
> {noformat}
> Other singleton streaming operators (regular join without keys, global
> aggregation, OVER/Top-N/deduplication without {{{}PARTITION BY{}}}, window
> aggregation without key) are *not* affected. They are covered by a fallback
> in {{ExecNodeBase}} (see History).
> h2. How to reproduce
> {code:sql}
> SET 'pipeline.max-parallelism' = '32768';
> SELECT a.id
> FROM a JOIN b
> ON b.ts >= a.ts + INTERVAL '5' MINUTE
> AND b.ts < a.ts + INTERVAL '10' MINUTE;
> {code}
> Inspect the StreamGraph and you get:
> ||Operator||parallelism||max parallelism||
> |IntervalJoin|1|*32768*|
> |TemporalSort ({{{}ORDER BY ts{}}})|1|*32768*|
> |Match ({{{}MATCH_RECOGNIZE{}}} without {{{}PARTITION BY{}}})|1|*32768*|
> |Join (keyless regular join)|1|1|
> |GroupAggregate (global)|1|1|
> |OverAggregate / Rank / Deduplicate (no {{{}PARTITION BY{}}})|1|1|
> To trigger the exception, rescale the interval join's vertex to parallelism
> >= 2 (for example via {{{}pipeline.jobvertex-parallelism-overrides{}}}) and
> run the job.
> * With {{{}pipeline.max-parallelism = 32768{}}}, the empty key falls into
> key group 24843, so any parallelism >= 2 fails.
> * With the default max parallelism (128), the empty key falls into key group
> 11. Subtask 0 owns that key group up to parallelism 11 ({{{}11 * p / 128 <
> 1{}}}), so small rescales go unnoticed and the job fails from parallelism 12
> on.
> h2. Root cause
> Two mechanisms are supposed to pin singleton operators to parallelism 1 and
> max parallelism 1:
> # *The generic fallback* in {{{}ExecNodeBase#translateToPlan{}}}. It pins
> the returned transformation if one of its direct inputs is a
> {{PartitionTransformation}} with a {{{}GlobalPartitioner{}}}. It only runs
> for nodes implementing {{{}SingleTransformationTranslator{}}}.
> # *Explicit {{inputsContainSingleton()}} checks* in individual nodes. Since
> FLINK-25995 these inspect the node's *own*
> {{{}InputProperty#getRequiredDistribution(){}}}.
> In the streaming planner, the distribution requirement only exists as a
> Calcite trait during optimization; it is satisfied by inserting
> {{{}StreamPhysicalExchange{}}}. When converting to ExecNodes, almost all
> streaming physical nodes pass {{InputProperty.DEFAULT}} ({{{}UNKNOWN{}}} in
> the compiled plan). Only {{{}StreamPhysicalExchange{}}},
> {{StreamPhysicalMultiJoin}} and {{StreamPhysicalProcessTableFunction}} (for
> broadcast table arguments only) declare a real requirement. Mechanism 2 is
> therefore dead for streaming.
> {{{}StreamExecIntervalJoin{}}}, {{StreamExecMatch}} and
> {{StreamExecTemporalSort}} implement
> {{{}MultipleTransformationTranslator{}}}, so mechanism 1 doesn't run for them
> either. Nothing pins them.
> Batch physical nodes declare their {{requiredDistribution}} correctly,
> because the batch exec-graph processors rely on it
> ({{{}ForwardHashExchangeProcessor{}}},
> {{{}MultipleInputNodeCreationProcessor{}}}, input-priority/deadlock
> handling). The streaming planner has no such processors, so the missing
> information never surfaced.
> h2. History
> * *FLINK-21730 (1.13,
> [apache/flink#15147|https://github.com/apache/flink/pull/15147]).* Introduced
> {{SingleTransformationTranslator}} / {{MultipleTransformationTranslator}} and
> moved the "singleton -> parallelism 1 / max parallelism 1" boilerplate into
> {{ExecNodeBase}} for single-transformation nodes. Multi-transformation nodes
> (IntervalJoin, Match, TemporalSort) kept an explicit check. At that time
> {{inputsContainSingleton()}} checked whether the *upstream node* was a
> {{CommonExecExchange}} with SINGLETON distribution, so it worked for
> streaming even though streaming nodes used {{{}InputProperty.DEFAULT{}}}.
> * *FLINK-25995 (1.15,
> [apache/flink#18707|https://github.com/apache/flink/pull/18707]).* "Make
> implicit assumption of SQL local hash explicit" (batch). Gave batch physical
> nodes a correct {{requiredDistribution}} and changed
> {{inputsContainSingleton()}} to check the node's *own* input properties. This
> was correct for batch, but disabled singleton pinning for *all* streaming
> nodes.
> * *FLINK-30814 (1.17,
> [apache/flink#21777|https://github.com/apache/flink/pull/21777]).* "Fix
> maxParallelism is not forced to 1 after global partitioner" (motivated by the
> adaptive batch scheduler). Added the {{GlobalPartitioner}} fallback, again
> only for single-transformation nodes. This incidentally repaired streaming
> single-transformation nodes. IntervalJoin, Match and TemporalSort remain
> unfixed.
> h2. Proposed fix
> Make streaming physical nodes declare the input distribution they require,
> consistent with batch physical nodes:
> * SINGLETON when the node requested a singleton distribution: no join keys /
> no partition keys / temporal sort.
> * HASH(keys) where the node requested a hash distribution on its inputs.
> {{{}StreamExecIntervalJoin{}}}, {{StreamExecMatch}} and
> {{StreamExecTemporalSort}} already call {{{}inputsContainSingleton(){}}}.
> With a correct {{InputProperty}} they get parallelism 1 and max parallelism 1
> without changes to the ExecNodes or {{{}ExecNodeBase{}}}. HASH has no runtime
> effect today, but makes the ExecNode graph and compiled plans
> self-describing. It also lets the planner later validate that each keyed
> operator is fed by a matching exchange.
> h2. Compatibility
> * *Existing compiled plans.* They contain {{requiredDistribution: UNKNOWN}}
> and are restored unchanged. Their operators keep their current max
> parallelism, so their state stays restorable. Pinning max parallelism 1 onto
> existing state with a different max parallelism would fail restore, because
> an explicitly configured max parallelism cannot be adapted to the state's.
> Old plans therefore keep the current, rescale-unsafe behavior. Users can fix
> them by recompiling the plan and starting without state.
> * *Newly compiled plans.* They contain the new input properties. Because
> {{inputProperties}} are already part of the plan JSON and SINGLETON/HASH are
> already serializable (used by {{{}StreamExecExchange{}}}), no ExecNode
> version bump or plan migration is needed. Plans compiled by a fixed planner
> also run on older runtimes.
> * *Fixing {{inputsContainSingleton()}} instead.* Restoring its pre-1.15
> upstream-exchange check would also fix the issue. It runs at translation
> time, though, so it would change max parallelism for *existing* compiled
> plans and break their state restore. Declaring the property in the physical
> nodes avoids that.
> * *Test baselines.* Compiled-plan JSON test files will need to be
> regenerated.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
