Timo Walther created FLINK-41440:
------------------------------------
Summary: 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
Reporter: Timo Walther
Assignee: Timo Walther
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)
