apache / apache/pinot

[multistage] decouple plan/runtime API abstracts.

Open
#10,657 3 comments 6 reactions 0 assignees View on GitHub
enhancement multi-stage
Dominant language
Java
Stars
6.1k
Forks
1.5k
Avg merge
2d 55m
Merged PRs (30d)
182

Description

Background
===

Currently we have multiple abstractions reused with different components in planner and runtime. it causes several problems
- when trying to add partition-based routing and planning it is super complex
- information only required in plan time and dispatch time get leaked to runtime which is not useful, but somehow usage is mixed and hard to change
- mailbox uses information way more than necessary and makes it hard to identify b/c the mailboxIdentifier equal requires all those to be identical.
- ... many other issues

These are related with partition strategy, worker assignment and mailbox-pipeline breaker efforts

Proposed changes
===
Several abstract is being introduced and will replace the current abstract
1. Step 1a: replace `VirtualServer`
`VirtualServer` is now a `ServerInstance + VirtualID`, it will be replaced with
`Worker` which is indicating parallelism of work. It:
(1) is globally indexed per stage;
(2) mapped to a single `ServerInstance` stored in `StageMetadata`,
(3) contains partition or segment info which will be put into a new abstract called: `WorkerMetadata`

with this `VirtualServer` is completely removed, and we decoupled `ServerInstance` which is not useful in runtime from `VirtualID`/`workerID` which is used in runtime.

- Step 1b: replace identifiers:
- `MailboxIdentifier` will use `workerID` which is globally indexed to uniquely identify a stream as:
`reqID|sendingStageID|sendingWorkerID|receivingStageID|receivingStageWorkerID`
- `OpChainID` will use `WorkerID` as well `reqID|stageID|workerID`

- Step 2: rename the API abstractions once it is clean

API/Class abstraction definitions
===
Here is a global view of what we need in terms of primitives
![Primitives](https://user-images.githubusercontent.com/3581352/235488698-98e8760c-d456-41e9-86a1-1e68ac001eca.png)

The Yellow objects with POJO definitions are what we plan to introduce, specifically:

Broker
----
- `Plan` (or `QueryPlan`) - represents a converted format from Calcite's `RelNode`
- Calcite's `RelNode` is being 1-1 converted to `PlanNode` (originally named as `StageNode` we decided to rename it b/c it is not necessarily associated with a `Stage`
- As we walk the tree we will attach `PlanMetadata`
- `SubPlan` - represents a part of the entire `Plan` that will be executed at a time. (This is a placeholder for now, it is used to pipelined execution in the future)
- all the children `SubPlan` should be executed first before the current subPlanRoot `PlanFragment`
- `PlanFragment` - represents a logical execution unit that can be executed at a time. (originally named as a `Stage`)
- unlike Stage definition nowadays, it contains children which is reference to all the depending `PlanFragment`s.
- `DispatchablePlan` - represents a `PlanFragment` + physical dispatch info to a specific `Worker`
- contains a list of `WorkerMetadata` that contains the physical execution info.

Server
----
- `StagePlan` - represents the format of the `DispatchablePlan` on server (originally named `DistributedStagePlan`)
- it includes only a subset of the information from `DispatchablePlan` that's needed by server.
- `OpChain` - this is not changed

CC @Jackie-Jiang @xiangfu0 @ankitsultana @somandal @siddharthteotia

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.