apache / apache/uniffle

[Umbrella] Refactor BlockId semantics

Open
#1,621 2 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
454
Forks
172
Avg merge
5d 17h
Merged PRs (30d)
5

Description

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

### Search before asking

- [X] I have searched in the [issues](https://github.com/apache/incubator-uniffle/issues?q=is%3Aissue) and found no similar issues.

### Describe the proposal

The block id encodes the task attempt id, the partition id, and the sequence number of a shuffle block. Block data are stored along with this block id, while some of these information are explicitly stored as well (task attempt id). For reading, interfaces use block id, partition id encoded in block id, and explicit task attempt ids to identify relevant blocks. Reading blocks explicitly by their block ids implicitly reads particular partition id, task attempt id and sequence number tuples.

Further, the block id is 63 bits long, while task attempt id and partition id are 31 bits long. This poses a conflict as bits have to be balanced between task attempt ids, partition ids and sequence number. There are potential situations where one of these bits are insufficient, which kills an application using the shuffle service.

The block id should be refactored to simplify interfaces, to make reading blocks explicit, and to avoid potential situations where bits are insufficient.

The block id is effectively a triple of (task attempt id, partition id, sequence number), truncated and encoded in a `long`.

Given each block id is stored together with its task attempt id, we can make block id a triple of (task index, partition id, sequence number). Blocks a read giving the set of task attempt ids that wrote the desired map index, we can reduce the bit foot print of the block id.

The structure of the block id should then be made explicit:

BlockId(int mapIndex, int partitionId, int sequenceNo)

The block id is often used in large sets, e.g.

- the set of all block ids written by a task attempt (register shuffle result)
- the set of all block ids that constitute a partition (reading blocks)

A task attempt can register blocks using a `Map`, which maps a partition id to the number of blocks (#1399). This should also reduce the memory footprint storing registered block ids on the shuffle server (or the Spark driver #1538).

Reading blocks should be as easy as reading blocks written by a set of task attempt ids that produced block ids that match `Map`, which maps a partition id to the task index and their number of blocks. Given map index and partition index are ranges, we can store these information efficiently as arrays `blockNums[partitionRangeIndex][mapRangeIndex]`.

While reading those blocks, the reader has to maintain a set of block sequence numbers from `0` to `blockNum - 1` (#1399) for each partition id and task index to identify duplicates. This can be done via `Map`, where the `Roaring64NavigableMap` stores read (map index, sequence number) tuples.

The `BufferSegment` stores the block id (`long`) and already the task attempt id (`long`). Reworking the block id will increases the size of `BufferSegment` only by one 32 bits (block id long turns into three ints).

Primary objective:
- Any possible task attempt id, partition id and sequence number is supported.

Benefits:
- the block id field information are not truncated any more
- concept of block id layout and its configuration can be removed
- block id field information are accessible server and client side
- id helper providing access to the partition id can be removed
- storing shuffle result registration server side does not require a number of bitmaps argument
- reduce redundancies in `BufferSegment`

### Task list

- [ ] Register shuffle result using `Map`
- [ ] Replace `taskAttemptId` with `mapIndex` in block id
- [ ] Hide set-of-block-ids implementation behind interface
- [ ] Use `BlockId` instead of `long` everywhere
- [ ] Turn `BlockId(long blockId)` into `BlockId(int mapIndex, int partitionId, int sequenceNo)`

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

Contributor guide

Open the contributing guide

Research direction

Start by tracing the BlockId and BufferSegment representations, then inspect the register-shuffle-result and block-reading interfaces described in the proposal. Use the task list to map the affected areas. Done means BlockId is explicit and untruncated, registration and reading use the proposed structures, and long-based block IDs and obsolete layout configuration are removed.

Written by the indexing model from the issue text.

Assessment

Tech stack
java, spark
Domain
backend, distributed-systems
Issue type
Refactor
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.