code: public static StageScheduler newSourcePartitionedSchedulerAsStageScheduler(
              SqlStageExecution stage,
              PlanNodeId partitionedNode,
              SplitSource splitSource,
              SplitPlacementPolicy splitPlacementPolicy,
              int splitBatchSize)
      {
          SourcePartitionedScheduler sourcePartitionedScheduler = new SourcePartitionedScheduler(stage, partitionedNode, splitSource, splitPlacementPolicy, splitBatchSize, false);
          sourcePartitionedScheduler.startLifespan(Lifespan.taskWide(), NOT_PARTITIONED);
  
          return new StageScheduler() {
              @Override
              public ScheduleResult schedule()
              {
                  ScheduleResult scheduleResult = sourcePartitionedScheduler.schedule();
                  sourcePartitionedScheduler.drainCompletelyScheduledLifespans();
                  return scheduleResult;
              }
  
              @Override
              public void close()
              {
                  sourcePartitionedScheduler.close();
              }
          };
      }
docs: Obtains an instance of {@code SourcePartitionedScheduler} suitable for use as a
  stage scheduler.
  <p>
  This returns an ungrouped {@code SourcePartitionedScheduler} that requires
  minimal management from the caller, which is ideal for use as a stage scheduler.