Arvid Heise created FLINK-21378:
-----------------------------------
Summary: Rescale pointwise connection during unaligned checkpoint
recovery
Key: FLINK-21378
URL: https://issues.apache.org/jira/browse/FLINK-21378
Project: Flink
Issue Type: Improvement
Components: Runtime / Checkpointing
Affects Versions: 1.13.0
Reporter: Arvid Heise
Assignee: Arvid Heise
FLINK-19801 added support for rescaling of unaligned checkpoints through
virtual channels: A mapping of old to new channel infos helped to create a
virtual channel that demultiplexes buffers from different original channel over
the same physical channel.
The calculation of FLINK-19801, however, assumes that subpartition = channel
index, which holds for all fully connected exchanges, but not for point-wise
connection. For point-wise connections, there are few channels per subtask and
they correspond to one particular subpartition.
A possible approach is to actually use the subpartition information while
constructing {{InflightDataRescalingDescriptor}} in {{TaskStateAssignment}}.
Thus, instead of taking subtask index as the channel index, we should take the
subpartition as the channel index. The easiest way to implement it is, by
translating subtask index to subpartition index and then calculate the channel
index from it.
For that, the following changes are needed:
* {{StateAssignmentOperation}} attaches the (upstream/downstream) ->
subpartition mapping to all assignments of pointwise exchanges. The information
can be derived through {{ExecutionEdge}} ->
{{IntermediateResultPartition.partitionNumber}} (note that on execution graph
level subpartition is named partition).
* For non-pointwise exchanges, this mapping is the identity function.
* {{TaskStateAssignment}} uses this additional lookup to translate subtask
mapping to subpartition mappings, which can then be used to calculate the
channel indexes both on input and output side.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
