[native pos] Change from Operator to Task Level Adapter

presto-main/src/main/java/com/facebook/presto/execution/StageExecutionInfo.java

@@ -16,7 +16,6 @@
 import com.facebook.airlift.stats.Distribution.DistributionSnapshot;
 import com.facebook.presto.common.RuntimeStats;
 import com.facebook.presto.operator.BlockedReason;
-import com.facebook.presto.operator.NativeExecutionInfo;
 import com.facebook.presto.operator.OperatorStats;
 import com.facebook.presto.operator.PipelineStats;
 import com.facebook.presto.operator.TaskStats;
@@ -131,21 +130,7 @@ public static StageExecutionInfo create(
             }
 
             TaskStats taskStats = taskInfo.getStats();
-
-            boolean isNativeTask = false;
-            for (PipelineStats pipeline : taskStats.getPipelines()) {
-                for (OperatorStats operatorStats : pipeline.getOperatorSummaries()) {
-                    if (operatorStats.getInfo() instanceof NativeExecutionInfo) {
-                        isNativeTask = true;
-                        allTaskStats.addAll(((NativeExecutionInfo) operatorStats.getInfo()).getTaskStats());
-                    }
-                }
-            }
-
-            // Prefer statistics from the native process.
-            if (!isNativeTask) {
-                allTaskStats.add(taskStats);
-            }
+            allTaskStats.add(taskStats);
 
             if (state == FINISHED && taskInfo.getTaskStatus().getState() == TaskState.FAILED) {
                 retriedCpuTime += taskStats.getTotalCpuTimeInNanos();

presto-main/src/main/java/com/facebook/presto/operator/OperatorInfo.java

@@ -26,7 +26,6 @@
 @JsonSubTypes({
         @JsonSubTypes.Type(value = ExchangeClientStatus.class, name = "exchangeClientStatus"),
         @JsonSubTypes.Type(value = LocalExchangeBufferInfo.class, name = "localExchangeBuffer"),
-        @JsonSubTypes.Type(value = NativeExecutionInfo.class, name = "NativeExecutionInfo"),
         @JsonSubTypes.Type(value = TableFinishInfo.class, name = "tableFinish"),
         @JsonSubTypes.Type(value = SplitOperatorInfo.class, name = "splitOperator"),
         @JsonSubTypes.Type(value = HashCollisionsInfo.class, name = "hashCollisionsInfo"),

presto-main/src/main/java/com/facebook/presto/sql/planner/BasePlanFragmenter.java

@@ -52,7 +52,6 @@
 import com.facebook.presto.sql.planner.plan.ExchangeNode;
 import com.facebook.presto.sql.planner.plan.ExplainAnalyzeNode;
 import com.facebook.presto.sql.planner.plan.MetadataDeleteNode;
-import com.facebook.presto.sql.planner.plan.NativeExecutionNode;
 import com.facebook.presto.sql.planner.plan.PlanFragmentId;
 import com.facebook.presto.sql.planner.plan.RemoteSourceNode;
 import com.facebook.presto.sql.planner.plan.SimplePlanRewriter;
@@ -78,7 +77,6 @@
 
 import static com.facebook.presto.SystemSessionProperties.getTaskPartitionedWriterCount;
 import static com.facebook.presto.SystemSessionProperties.isForceSingleNodeOutput;
-import static com.facebook.presto.SystemSessionProperties.isNativeExecutionEnabled;
 import static com.facebook.presto.SystemSessionProperties.isTableWriterMergeOperatorEnabled;
 import static com.facebook.presto.common.type.BigintType.BIGINT;
 import static com.facebook.presto.common.type.VarbinaryType.VARBINARY;
@@ -179,35 +177,6 @@ private SubPlan buildFragment(PlanNode root, FragmentProperties properties, Plan
                     tableWriterNodeIds);
         }
 
-        // Only delegate non-coordinatorOnly plan fragment to native engine
-        if (isNativeExecutionEnabled(session) && !properties.getPartitioningHandle().isCoordinatorOnly()) {
-            if (root instanceof OutputNode) {
-                // OutputNode is special in that it can have duplicate output variables since it
-                // does not get converted to a PhysicalOperation for execution.
-                // Regular plan nodes, including NativeExecutionNode, must have unique output variables.
-                // Check if OutputNode has duplicate output variables and remove these.
-                // This is safe because OutputNode variables are not used by the workers.
-                OutputNode outputNode = (OutputNode) root;
-                List<VariableReferenceExpression> outputVariables = outputNode.getOutputVariables();
-                List<VariableReferenceExpression> newOutputVariables = new ArrayList<>();
-                List<String> newColumnNames = new ArrayList<>();
-                Set<VariableReferenceExpression> uniqueOutputVariables = new HashSet<>();
-                for (int i = 0; i < outputVariables.size(); ++i) {
-                    VariableReferenceExpression variable = outputVariables.get(i);
-                    if (uniqueOutputVariables.add(variable)) {
-                        newOutputVariables.add(variable);
-                        newColumnNames.add(outputNode.getColumnNames().get(i));
-                    }
-                }
-
-                if (uniqueOutputVariables.size() < outputVariables.size()) {
-                    root = new OutputNode(outputNode.getSourceLocation(), outputNode.getId(), outputNode.getSource(), newColumnNames, newOutputVariables);
-                }
-            }
-            root = new NativeExecutionNode(root);
-            schedulingOrder = scheduleOrder(root);
-        }
-
         PlanFragment fragment = new PlanFragment(
                 fragmentId,
                 root,

presto-main/src/main/java/com/facebook/presto/sql/planner/InputExtractor.java

@@ -29,7 +29,6 @@
 import com.facebook.presto.sql.planner.plan.IndexSourceNode;
 import com.facebook.presto.sql.planner.plan.InternalPlanVisitor;
 import com.facebook.presto.sql.planner.plan.JoinNode;
-import com.facebook.presto.sql.planner.plan.NativeExecutionNode;
 import com.facebook.presto.sql.planner.plan.SemiJoinNode;
 import com.facebook.presto.sql.planner.plan.SpatialJoinNode;
 import com.google.common.collect.ImmutableList;
@@ -155,13 +154,6 @@ public Void visitIndexSource(IndexSourceNode node, Context context)
             return null;
         }
 
-        @Override
-        public Void visitNativeExecution(NativeExecutionNode node, Context context)
-        {
-            node.getSubPlan().accept(this, context);
-            return null;
-        }
-
         @Override
         public Void visitPlan(PlanNode node, Context context)
         {

presto-main/src/main/java/com/facebook/presto/sql/planner/OutputExtractor.java

@@ -18,7 +18,6 @@
 import com.facebook.presto.spi.SchemaTableName;
 import com.facebook.presto.spi.plan.PlanNode;
 import com.facebook.presto.sql.planner.plan.InternalPlanVisitor;
-import com.facebook.presto.sql.planner.plan.NativeExecutionNode;
 import com.facebook.presto.sql.planner.plan.TableWriterNode;
 import com.google.common.base.VerifyException;
 
@@ -62,13 +61,6 @@ public Void visitTableWriter(TableWriterNode node, Void context)
             return null;
         }
 
-        @Override
-        public Void visitNativeExecution(NativeExecutionNode node, Void context)
-        {
-            node.getSubPlan().accept(this, context);
-            return null;
-        }
-
         @Override
         public Void visitPlan(PlanNode node, Void context)
         {

presto-main/src/main/java/com/facebook/presto/sql/planner/PlanFragment.java

@@ -20,7 +20,6 @@
 import com.facebook.presto.spi.plan.PlanNode;
 import com.facebook.presto.spi.plan.PlanNodeId;
 import com.facebook.presto.spi.relation.VariableReferenceExpression;
-import com.facebook.presto.sql.planner.plan.NativeExecutionNode;
 import com.facebook.presto.sql.planner.plan.PlanFragmentId;
 import com.facebook.presto.sql.planner.plan.RemoteSourceNode;
 import com.fasterxml.jackson.annotation.JsonCreator;
@@ -214,11 +213,6 @@ private static void findRemoteSourceNodes(PlanNode node, Builder<RemoteSourceNod
             findRemoteSourceNodes(source, builder);
         }
 
-        if (node instanceof NativeExecutionNode) {
-            findRemoteSourceNodes(((NativeExecutionNode) node).getSubPlan(), builder);
-            return;
-        }
-
         if (node instanceof RemoteSourceNode) {
             builder.add((RemoteSourceNode) node);
         }

presto-main/src/main/java/com/facebook/presto/sql/planner/SchedulingOrderVisitor.java

@@ -20,7 +20,6 @@
 import com.facebook.presto.sql.planner.plan.IndexJoinNode;
 import com.facebook.presto.sql.planner.plan.InternalPlanVisitor;
 import com.facebook.presto.sql.planner.plan.JoinNode;
-import com.facebook.presto.sql.planner.plan.NativeExecutionNode;
 import com.facebook.presto.sql.planner.plan.SemiJoinNode;
 import com.facebook.presto.sql.planner.plan.SpatialJoinNode;
 import com.google.common.collect.ImmutableList;
@@ -89,12 +88,5 @@ public Void visitTableScan(TableScanNode node, Consumer<PlanNodeId> schedulingOr
             schedulingOrder.accept(node.getId());
             return null;
         }
-
-        @Override
-        public Void visitNativeExecution(NativeExecutionNode node, Consumer<PlanNodeId> schedulingOrder)
-        {
-            schedulingOrder.accept(node.getId());
-            return null;
-        }
     }
 }

presto-main/src/main/java/com/facebook/presto/sql/planner/SplitSourceFactory.java

@@ -48,7 +48,6 @@
 import com.facebook.presto.sql.planner.plan.JoinNode;
 import com.facebook.presto.sql.planner.plan.MergeJoinNode;
 import com.facebook.presto.sql.planner.plan.MetadataDeleteNode;
-import com.facebook.presto.sql.planner.plan.NativeExecutionNode;
 import com.facebook.presto.sql.planner.plan.RemoteSourceNode;
 import com.facebook.presto.sql.planner.plan.RowNumberNode;
 import com.facebook.presto.sql.planner.plan.SampleNode;
@@ -398,12 +397,6 @@ public Map<PlanNodeId, SplitSource> visitExchange(ExchangeNode node, Context con
             return processSources(node.getSources(), context);
         }
 
-        @Override
-        public Map<PlanNodeId, SplitSource> visitNativeExecution(NativeExecutionNode node, Context context)
-        {
-            return processSources(ImmutableList.of(node.getSubPlan()), context);
-        }
-
         private Map<PlanNodeId, SplitSource> processSources(List<PlanNode> sources, Context context)
         {
             ImmutableMap.Builder<PlanNodeId, SplitSource> result = ImmutableMap.builder();

presto-main/src/main/java/com/facebook/presto/sql/planner/iterative/Plans.java

@@ -15,7 +15,6 @@
 
 import com.facebook.presto.spi.plan.PlanNode;
 import com.facebook.presto.sql.planner.plan.InternalPlanVisitor;
-import com.facebook.presto.sql.planner.plan.NativeExecutionNode;
 
 import java.util.List;
 import java.util.stream.Collectors;
@@ -55,13 +54,6 @@ public PlanNode visitGroupReference(GroupReference node, Void context)
         {
             return lookup.resolve(node).accept(this, context);
         }
-
-        @Override
-        public PlanNode visitNativeExecution(NativeExecutionNode node, Void context)
-        {
-            node.getSubPlan().accept(this, context);
-            return node;
-        }
     }
 
     private Plans() {}

presto-main/src/main/java/com/facebook/presto/sql/planner/plan/InternalPlanVisitor.java

@@ -176,9 +176,4 @@ public R visitStatsEquivalentPlanNodeWithLimit(StatsEquivalentPlanNodeWithLimit
     {
         return visitPlan(node, context);
     }
-
-    public R visitNativeExecution(NativeExecutionNode node, C context)
-    {
-        return visitPlan(node, context);
-    }
 }