[Ballista] Introduce QueryStageScheduler for better managing the stage-based task scheduling

[yahoNanJing]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

The implementation for get_next_schedulable_task() is not efficient due to lack of classification of task stages and task status, especially when there's thousands of tasks to be scheduled. And it's also not easy to:

• stage-level priority-based scheduling
• stage-level retry
• speculative task scheduling

Describe the solution you'd like

To draw lessons from the Spark, it's better to divide the task scheduling into two levels' scheduling:

• to introduce QueryStageScheduler for scheduling stages
• to introduce StageManager for managing job stages and the state machine for each stage.
• to let TaskScheduler for fetching tasks from running stages in StageManager

An example sequential diagram is as follows:

[yahoNanJing]

It's quite a big change to achieve all of this. The whole change will be divided into the following tasks:

1. [Ballista] Refactor ballista scheduler server #1908
2. [Ballista] Refactor the event channel #1909
3. [Ballista] Refactor scheduler state #1910
4. [Ballista] Introduce query stage scheduler #1934
5. [Ballista] Introduce StageManager for managing tasks stage by stage #1936