[multistage] decouple plan/runtime API abstracts.

[walterddr]

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

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 PlanFragments.
• 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

[siddharthteotia]

Nice. Will review

FYI @vvivekiyer

[walterddr]

step 1a and 1b are done via (#10665, #10673) and (#10681, #10669)

[walterddr]

collecting some feedback from folks, please feel free to comment below as well:

• Parser should be part of the external of planner so that

  • it can be used in controller and broker to parse out tables for routing
  • parser is also used by v1 engine so it make sense to single it out (and it is invariant with the Planner)

• We should consider making the component modules so that it doesn't need to force dependency on some other packages, for example:

  • parser doens't need to depend on planner
  • dispatcher doesn't need to depend on calcite.
  • decision here is to make sure interfaces are clear, whether to group them into separated modules can be refactor later if interfaces have clear boundaries.

• Planner -> LogicalPlanner:

  • there should be a conversion regarding such as RelTrait, or RelExchange is being the only node looked at by dispatcher

• where does some planning info to

  • information that only affects the node->operator then this should be node-level info and make it used in PhysicalPlanner (on the server) (e.g. use concurrent index map group-by or 2-way reducer/merger group-by, this decision might not be able to be made on broker b/c it might need info like column cardinality on segment level)
  • Information that not only affects the node but also the upstream/downstream then this decision should be made on the broker (e.g. using sort-merge join requires sorting operation on both inputs; using hash join requires materializing hash table on either left or right side)
  • hint can always be attached, but the decision on where to use that hint is decided with the rule above^

• PinotQueryPlanner and Plan should be a separate module

  • the only goal achieved here is de-calcite --> convert RelNode into format that doesn't depend on Calcite
  • PinotQueryPlanner + LogicalPlanner can be replaced with a Cache later on if we decide to run periodic tasks that doesn't have changes to the SQL itself (e.g. speed-up planning)