Function align_splits 

pub fn align_splits(
    aligned_actors: impl IntoIterator<Item = (ActorId, ActorId)>,
    get_upstream_actor_splits: impl Fn(ActorId) -> Option<Vec<SplitImpl>>,
    fragment_id: FragmentId,
    upstream_source_fragment_id: FragmentId,
) -> Result<HashMap<ActorId, Vec<SplitImpl>>>

Assign splits to a new set of actors, according to existing assignment.

The get_upstream_actor_splits closure looks up the current splits for a given upstream actor ID. How exactly this lookup works depends on the caller:

• Inside the barrier worker, it reads from InflightDatabaseInfo.
• During reassignment, it reads from the pending upstream assignment.

illustration:

upstream                               new
actor x1 [split 1, split2]      ->     actor y1 [split 1, split2]
actor x2 [split 3]              ->     actor y2 [split 3]
...