1.根据SqlQueryExecution::analyzeQuery方法生成subPlan
    2.根据获取subPlan信息，在SqlQueryExecution::planDistribution 方法调用了DistributedExecutionPlanner的plan方法

    1. private void planDistribution(PlanRoot plan)
    2.     {
    3.         // time distribution planning  时间分配计划
    4.         stateMachine.beginDistributedPlanning();
    6.         // plan the execution on the active nodes   计划活动节点上的执行
    7.         DistributedExecutionPlanner distributedPlanner = new DistributedExecutionPlanner(splitManager, metadata);
    8.         StageExecutionPlan outputStageExecutionPlan;
    9.         Session session = stateMachine.getSession();
    10.         Map<Integer, Integer> parallelSources;
    11.         if (SystemSessionProperties.isSnapshotEnabled(session)) {
    13.             // Snapshot: UNION statement creates situation where multiple table scan stages communicate with the
    14.             // same ExchangeOperator, we need to be aware of that in order to adjust the taskCount on Markers coming from
    15.             // those table scan stages to have successful snapshots.
    16.             // The key represents the table scan stage's stageId and value represents how many table scan stages
    17.             // are in parallel in total.
    18.             //快照：UNION语句创建了多个表扫描阶段与同一个ExchangeOperator进行通信的情况，我们需要注意这一点，
    19.             // 以便调整来自那些表扫描阶段的Marks上的taskCount以获得成功的快照。
    20.             // 该键表示表扫描阶段的stageId，而值表示总共有多少个表扫描阶段并行。
    22.             parallelSources = new HashMap<>();
    23.             // Snapshot: need to plan different when snapshot is enabled.  启用快照时需要进行不同的计划
    24.             // See the "plan" method for difference between the different modes.  不同模式的区别见“计划”方法
    25.             MarkerAnnouncer announcer = splitManager.getMarkerAnnouncer(session);
    26.             outputStageExecutionPlan = distributedPlanner.plan(plan.getRoot(), session, SNAPSHOT, null, announcer.currentSnapshotId(), parallelSources);
    27.         }
    28.         else {
    29.             parallelSources = null;
    30.             // 将子计划生成对应的 阶段执行计划
    31.             outputStageExecutionPlan = distributedPlanner.plan(plan.getRoot(), session, NORMAL, null, 0, null);
    32.         }
    33.         // 结束分发计划
    34.         stateMachine.endDistributedPlanning();
    36.         // ensure split sources are closed 确保分片数据源已经关闭
    37.         stateMachine.addStateChangeListener(state -> {
    38.             if (state.isDone()) {
    39.                 closeSplitSources(outputStageExecutionPlan);
    40.             }
    41.         });
    43.         // if query was canceled, skip creating scheduler
    44.         if (stateMachine.isDone()) {
    45.             return;
    46.         }
    48.         // record output field  获取输出字段 -- 》  <字段名，字段类型>  映射关系
    49.         stateMachine.setColumns(outputStageExecutionPlan.getFieldNames(), outputStageExecutionPlan.getFragment().getTypes());
    51.         PartitioningHandle partitioningHandle = plan.getRoot().getFragment().getPartitioningScheme().getPartitioning().getHandle();
    52.         OutputBuffers rootOutputBuffers = createInitialEmptyOutputBuffers(partitioningHandle)
    53.                 .withBuffer(OUTPUT_BUFFER_ID, BROADCAST_PARTITION_ID)
    54.                 .withNoMoreBufferIds();
    56.         // build the stage execution objects (this doesn't schedule execution)  构建阶段执行对象（不是调度执行）
    57.         SqlQueryScheduler scheduler = createSqlQueryScheduler(
    58.                 stateMachine,
    59.                 locationFactory,
    60.                 outputStageExecutionPlan,
    61.                 nodePartitioningManager,
    62.                 nodeScheduler,
    63.                 remoteTaskFactory,
    64.                 stateMachine.getSession(),
    65.                 plan.isSummarizeTaskInfos(),
    66.                 scheduleSplitBatchSize,
    67.                 queryExecutor,
    68.                 schedulerExecutor,
    69.                 failureDetector,
    70.                 rootOutputBuffers,
    71.                 nodeTaskMap,
    72.                 executionPolicy,
    73.                 schedulerStats,
    74.                 dynamicFilterService,
    75.                 heuristicIndexerManager,
    76.                 snapshotManager,
    77.                 null,
    78.                 parallelSources);
    80.         queryScheduler.set(scheduler);
    82.         // if query was canceled during scheduler creation, abort the scheduler
    83.         // directly since the callback may have already fired
    84.         if (stateMachine.isDone()) {
    85.             scheduler.abort();
    86.             queryScheduler.set(null);
    87.         }
    88.     }