Small PlanFragmenter refactorings
@@ -51,7 +51,6 @@ import com.facebook.presto.sql.planner.plan.ValuesNode;
import com.facebook.presto.sql.planner.plan.WindowNode;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Maps;
import javax.inject.Inject;
@@ -69,6 +68,7 @@ import static com.facebook.presto.spi.StandardErrorCode.QUERY_HAS_TOO_MANY_STAGE
import static com.facebook.presto.spi.StandardWarningCode.TOO_MANY_STAGES;
import static com.facebook.presto.spi.connector.NotPartitionedPartitionHandle.NOT_PARTITIONED;
import static com.facebook.presto.sql.planner.SchedulingOrderVisitor.scheduleOrder;
+import static com.facebook.presto.sql.planner.SymbolsExtractor.extractOutputSymbols;
import static com.facebook.presto.sql.planner.SystemPartitioningHandle.COORDINATOR_DISTRIBUTION;
import static com.facebook.presto.sql.planner.SystemPartitioningHandle.SINGLE_DISTRIBUTION;
import static com.facebook.presto.sql.planner.SystemPartitioningHandle.SOURCE_DISTRIBUTION;
@@ -80,6 +80,7 @@ import static com.google.common.base.Preconditions.checkState;
import static com.google.common.base.Predicates.in;
import static com.google.common.collect.ImmutableList.toImmutableList;
import static com.google.common.collect.Iterables.getOnlyElement;
+import static com.google.common.collect.Maps.filterKeys;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
@@ -238,24 +239,24 @@ public class PlanFragmenter
private SubPlan buildFragment(PlanNode root, FragmentProperties properties, PlanFragmentId fragmentId)
{
- Set<Symbol> dependencies = SymbolsExtractor.extractOutputSymbols(root);
-
List<PlanNodeId> schedulingOrder = scheduleOrder(root);
- boolean equals = properties.getPartitionedSources().equals(ImmutableSet.copyOf(schedulingOrder));
- checkArgument(equals, "Expected scheduling order (%s) to contain an entry for all partitioned sources (%s)", schedulingOrder, properties.getPartitionedSources());
-
- Map<Symbol, Type> symbols = Maps.filterKeys(types.allTypes(), in(dependencies));
+ checkArgument(
+ properties.getPartitionedSources().equals(ImmutableSet.copyOf(schedulingOrder)),
+ "Expected scheduling order (%s) to contain an entry for all partitioned sources (%s)",
+ schedulingOrder,
+ properties.getPartitionedSources());
+ Map<Symbol, Type> fragmentSymbolTypes = filterKeys(types.allTypes(), in(extractOutputSymbols(root)));
PlanFragment fragment = new PlanFragment(
fragmentId,
root,
- symbols,
+ fragmentSymbolTypes,
properties.getPartitioningHandle(),
schedulingOrder,
properties.getPartitioningScheme(),
StageExecutionDescriptor.ungroupedExecution(),
statsAndCosts.getForSubplan(root),
- Optional.of(jsonFragmentPlan(root, symbols, metadata.getFunctionManager(), session)));
+ Optional.of(jsonFragmentPlan(root, fragmentSymbolTypes, metadata.getFunctionManager(), session)));
return new SubPlan(fragment, properties.getChildren());
}
