Optimize PlanFragment Serialization

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

@@ -22,6 +22,7 @@
 import com.facebook.presto.sql.planner.plan.PlanFragmentId;
 import com.facebook.presto.sql.planner.plan.RemoteSourceNode;
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableList.Builder;
@@ -51,7 +52,7 @@ public class PlanFragment
     private final PartitioningScheme partitioningScheme;
     private final StageExecutionDescriptor stageExecutionDescriptor;
     private final boolean outputTableWriterFragment;
-    private final StatsAndCosts statsAndCosts;
+    private final Optional<StatsAndCosts> statsAndCosts;
     private final Optional<String> jsonRepresentation;
 
     @JsonCreator
@@ -64,7 +65,7 @@ public PlanFragment(
             @JsonProperty("partitioningScheme") PartitioningScheme partitioningScheme,
             @JsonProperty("stageExecutionDescriptor") StageExecutionDescriptor stageExecutionDescriptor,
             @JsonProperty("outputTableWriterFragment") boolean outputTableWriterFragment,
-            @JsonProperty("statsAndCosts") StatsAndCosts statsAndCosts,
+            @JsonProperty("statsAndCosts") Optional<StatsAndCosts> statsAndCosts,
             @JsonProperty("jsonRepresentation") Optional<String> jsonRepresentation)
     {
         this.id = requireNonNull(id, "id is null");
@@ -139,17 +140,15 @@ public boolean isOutputTableWriterFragment()
         return outputTableWriterFragment;
     }
 
-    @JsonProperty
-    public StatsAndCosts getStatsAndCosts()
+    @JsonIgnore
+    public Optional<StatsAndCosts> getStatsAndCosts()
     {
         return statsAndCosts;
     }
 
-    @JsonProperty
+    @JsonIgnore
     public Optional<String> getJsonRepresentation()
     {
-        // @reviewer: I believe this should be a json raw value, but that would make this class have a different deserialization constructor.
-        // workers don't need this, so that should be OK, but it's worth thinking about.
         return jsonRepresentation;
     }
 

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

@@ -387,7 +387,7 @@ private SubPlan buildFragment(PlanNode root, FragmentProperties properties, Plan
                     properties.getPartitioningScheme(),
                     StageExecutionDescriptor.ungroupedExecution(),
                     outputTableWriterFragment,
-                    statsAndCosts.getForSubplan(root),
+                    Optional.of(statsAndCosts.getForSubplan(root)),
                     Optional.of(jsonFragmentPlan(root, fragmentVariableTypes, metadata.getFunctionManager(), session)));
 
             return new SubPlan(fragment, properties.getChildren());

presto-main/src/main/java/com/facebook/presto/sql/planner/planPrinter/PlanPrinter.java

@@ -299,7 +299,7 @@ private static String formatFragment(FunctionManager functionManager, Session se
                 .flatMap(f -> f.getVariables().stream())
                 .distinct()
                 .collect(toImmutableList()));
-        builder.append(textLogicalPlan(fragment.getRoot(), typeProvider, Optional.of(fragment.getStageExecutionDescriptor()), functionManager, fragment.getStatsAndCosts(), session, planNodeStats, 1, verbose))
+        builder.append(textLogicalPlan(fragment.getRoot(), typeProvider, Optional.of(fragment.getStageExecutionDescriptor()), functionManager, fragment.getStatsAndCosts().orElse(StatsAndCosts.empty()), session, planNodeStats, 1, verbose))
                 .append("\n");
 
         return builder.toString();
@@ -317,7 +317,7 @@ public static String graphvizLogicalPlan(PlanNode plan, TypeProvider types, Sess
                 new PartitioningScheme(Partitioning.create(SINGLE_DISTRIBUTION, ImmutableList.of()), plan.getOutputVariables()),
                 StageExecutionDescriptor.ungroupedExecution(),
                 false,
-                StatsAndCosts.empty(),
+                Optional.empty(),
                 Optional.empty());
         return GraphvizPrinter.printLogical(ImmutableList.of(fragment), session, functionManager);
     }

presto-main/src/test/java/com/facebook/presto/cost/TestCostCalculator.java

@@ -564,7 +564,7 @@ private CostAssertionBuilder assertCostFragmentedPlan(
         CostProvider costProvider = new TestingCostProvider(costs, costCalculatorUsingExchanges, statsProvider, session, typeProvider);
         PlanNode plan = translateExpression(node, statsCalculator(stats), typeProvider);
         SubPlan subPlan = fragment(new Plan(plan, typeProvider, StatsAndCosts.create(node, statsProvider, costProvider)));
-        return new CostAssertionBuilder(subPlan.getFragment().getStatsAndCosts().getCosts().getOrDefault(node.getId(), PlanCostEstimate.unknown()));
+        return new CostAssertionBuilder(subPlan.getFragment().getStatsAndCosts().orElse(StatsAndCosts.empty()).getCosts().getOrDefault(node.getId(), PlanCostEstimate.unknown()));
     }
 
     private PlanNode translateExpression(PlanNode node, StatsCalculator statsCalculator, TypeProvider typeProvider)
@@ -700,7 +700,7 @@ private PlanCostEstimate calculateCostFragmentedPlan(PlanNode node, StatsCalcula
         CostProvider costProvider = new CachingCostProvider(costCalculatorUsingExchanges, statsProvider, Optional.empty(), session, typeProvider);
         node = translateExpression(node, statsCalculator, typeProvider);
         SubPlan subPlan = fragment(new Plan(node, typeProvider, StatsAndCosts.create(node, statsProvider, costProvider)));
-        return subPlan.getFragment().getStatsAndCosts().getCosts().getOrDefault(node.getId(), PlanCostEstimate.unknown());
+        return subPlan.getFragment().getStatsAndCosts().orElse(StatsAndCosts.empty()).getCosts().getOrDefault(node.getId(), PlanCostEstimate.unknown());
     }
 
     private static class CostAssertionBuilder

presto-main/src/test/java/com/facebook/presto/execution/MockRemoteTaskFactory.java

@@ -15,7 +15,6 @@
 
 import com.facebook.airlift.stats.TestingGcMonitor;
 import com.facebook.presto.Session;
-import com.facebook.presto.cost.StatsAndCosts;
 import com.facebook.presto.execution.NodeTaskMap.PartitionedSplitCountTracker;
 import com.facebook.presto.execution.buffer.LazyOutputBuffer;
 import com.facebook.presto.execution.buffer.OutputBuffer;
@@ -124,7 +123,7 @@ public MockRemoteTask createTableScanTask(TaskId taskId, InternalNode newNode, L
                 new PartitioningScheme(Partitioning.create(SINGLE_DISTRIBUTION, ImmutableList.of()), ImmutableList.of(variable)),
                 StageExecutionDescriptor.ungroupedExecution(),
                 false,
-                StatsAndCosts.empty(),
+                Optional.empty(),
                 Optional.empty());
 
         ImmutableMultimap.Builder<PlanNodeId, Split> initialSplits = ImmutableMultimap.builder();

presto-main/src/test/java/com/facebook/presto/execution/TaskTestUtils.java

@@ -15,7 +15,6 @@
 
 import com.facebook.airlift.json.ObjectMapperProvider;
 import com.facebook.presto.block.BlockEncodingManager;
-import com.facebook.presto.cost.StatsAndCosts;
 import com.facebook.presto.event.SplitMonitor;
 import com.facebook.presto.eventlistener.EventListenerManager;
 import com.facebook.presto.execution.buffer.OutputBuffers;
@@ -109,7 +108,7 @@ private TaskTestUtils()
                     .withBucketToPartition(Optional.of(new int[1])),
             StageExecutionDescriptor.ungroupedExecution(),
             false,
-            StatsAndCosts.empty(),
+            Optional.empty(),
             Optional.empty());
 
     public static LocalExecutionPlanner createTestingPlanner()

presto-main/src/test/java/com/facebook/presto/execution/TestSqlStageExecution.java

@@ -14,7 +14,6 @@
 package com.facebook.presto.execution;
 
 import com.facebook.presto.client.NodeVersion;
-import com.facebook.presto.cost.StatsAndCosts;
 import com.facebook.presto.execution.scheduler.SplitSchedulerStats;
 import com.facebook.presto.execution.scheduler.TableWriteInfo;
 import com.facebook.presto.failureDetector.NoOpFailureDetector;
@@ -173,7 +172,7 @@ private static PlanFragment createExchangePlanFragment()
                 new PartitioningScheme(Partitioning.create(SINGLE_DISTRIBUTION, ImmutableList.of()), planNode.getOutputVariables()),
                 StageExecutionDescriptor.ungroupedExecution(),
                 false,
-                StatsAndCosts.empty(),
+                Optional.empty(),
                 Optional.empty());
     }
 }

presto-main/src/test/java/com/facebook/presto/execution/TestStageExecutionStateMachine.java

@@ -13,7 +13,6 @@
  */
 package com.facebook.presto.execution;
 
-import com.facebook.presto.cost.StatsAndCosts;
 import com.facebook.presto.execution.scheduler.SplitSchedulerStats;
 import com.facebook.presto.operator.StageExecutionDescriptor;
 import com.facebook.presto.spi.QueryId;
@@ -324,7 +323,7 @@ private static PlanFragment createValuesPlan()
                 new PartitioningScheme(Partitioning.create(SINGLE_DISTRIBUTION, ImmutableList.of()), ImmutableList.of(variable)),
                 StageExecutionDescriptor.ungroupedExecution(),
                 false,
-                StatsAndCosts.empty(),
+                Optional.empty(),
                 Optional.empty());
 
         return planFragment;

presto-main/src/test/java/com/facebook/presto/execution/scheduler/TestPhasedExecutionSchedule.java

@@ -13,7 +13,6 @@
  */
 package com.facebook.presto.execution.scheduler;
 
-import com.facebook.presto.cost.StatsAndCosts;
 import com.facebook.presto.operator.StageExecutionDescriptor;
 import com.facebook.presto.spi.ConnectorId;
 import com.facebook.presto.spi.TableHandle;
@@ -269,7 +268,7 @@ private static PlanFragment createFragment(PlanNode planNode)
                 new PartitioningScheme(Partitioning.create(SINGLE_DISTRIBUTION, ImmutableList.of()), planNode.getOutputVariables()),
                 StageExecutionDescriptor.ungroupedExecution(),
                 false,
-                StatsAndCosts.empty(),
+                Optional.empty(),
                 Optional.empty());
     }
 }

presto-main/src/test/java/com/facebook/presto/execution/scheduler/TestSourcePartitionedScheduler.java

@@ -14,7 +14,6 @@
 package com.facebook.presto.execution.scheduler;
 
 import com.facebook.presto.client.NodeVersion;
-import com.facebook.presto.cost.StatsAndCosts;
 import com.facebook.presto.execution.LocationFactory;
 import com.facebook.presto.execution.MockRemoteTaskFactory;
 import com.facebook.presto.execution.MockRemoteTaskFactory.MockRemoteTask;
@@ -486,7 +485,7 @@ private static SubPlan createPlan()
                 new PartitioningScheme(Partitioning.create(SINGLE_DISTRIBUTION, ImmutableList.of()), ImmutableList.of(variable)),
                 StageExecutionDescriptor.ungroupedExecution(),
                 false,
-                StatsAndCosts.empty(),
+                Optional.empty(),
                 Optional.empty());
 
         return new SubPlan(testFragment, ImmutableList.of());