public DistributedQueryPlan planFragments() {
    PlanNode rootAfterRewrite = rewriteSource();

    PlanNode rootWithExchange = addExchangeNode(rootAfterRewrite);
    PlanNode optimizedRootWithExchange = optimize(rootWithExchange);
    if (analysis.getTreeStatement() != null && analysis.getTreeStatement().isQuery()) {
      analysis
          .getRespDatasetHeader()
          .setTreeColumnToTsBlockIndexMap(optimizedRootWithExchange.getOutputColumnNames());
    }

    SubPlan subPlan = splitFragment(optimizedRootWithExchange);
    // Mark the root Fragment of root SubPlan as `root`
    subPlan.getPlanFragment().setRoot(true);

    List<FragmentInstance> fragmentInstances = planFragmentInstances(subPlan);

    // Only execute this step for READ operation
    if (context.getQueryType() == QueryType.READ) {
      setSinkForRootInstance(subPlan, fragmentInstances);
    }

    return new DistributedQueryPlan(subPlan, fragmentInstances);
  }
Return true if we need to use ShuffleSinkNode instead of IdentitySinkNode.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/distribution/DistributionPlanner.java