DRILL-2170: For fragment parallelization, use max cost of operators i…

…nstead of total cost (this makes it consistent with what ExcessiveExchangeRemover uses).

exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/SimpleParallelizer.java

@@ -182,7 +182,9 @@ private void assignEndpoints(Collection<DrillbitEndpoint> allNodes, PlanningSet
 
       Stats stats = wrapper.getStats();
 
-      double targetSlices = stats.getTotalCost()/parallelizationThreshold;
+      // Use max cost of all operators in this fragment; this is consistent with the
+      // calculation that ExcessiveExchangeRemover uses
+      double targetSlices = stats.getMaxCost()/parallelizationThreshold;
       int targetIntSlices = (int) Math.ceil(targetSlices);
 
       // figure out width.

exec/java-exec/src/main/java/org/apache/drill/exec/planner/fragment/Stats.java

@@ -22,14 +22,14 @@ public class Stats {
   static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Stats.class);
 
   private int maxWidth = Integer.MAX_VALUE;
-  private double totalCost;
+  private double maxCost = 0.0;
 
   public void addMaxWidth(int maxWidth){
     this.maxWidth = Math.min(this.maxWidth, maxWidth);
   }
 
   public void addCost(double cost){
-    totalCost += cost;
+    maxCost = Math.max(maxCost, cost);
   }
 
   public int getMaxWidth() {
@@ -38,12 +38,11 @@ public int getMaxWidth() {
 
   @Override
   public String toString() {
-    return "Stats [maxWidth=" + maxWidth + ", totalCost=" + totalCost + "]";
+    return "Stats [maxWidth=" + maxWidth + ", maxCost=" + maxCost + "]";
   }
 
-  public double getTotalCost(){
-    return totalCost;
+  public double getMaxCost() {
+    return maxCost;
   }
 
-
 }

exec/java-exec/src/test/java/org/apache/drill/exec/fn/impl/TestAggregateFunctions.java

@@ -94,4 +94,23 @@ public void testDrill2092() throws Exception {
 
   }
 
+  @Test // DRILL-2170: Subquery has group-by, order-by on aggregate function and limit
+  public void testDrill2170() throws Exception {
+    String query =
+        "select count(*) as cnt from "
+        + "cp.`tpch/orders.parquet` o inner join\n"
+        + "(select l_orderkey, sum(l_quantity), sum(l_extendedprice) \n"
+        + "from cp.`tpch/lineitem.parquet` \n"
+        + "group by l_orderkey order by 3 limit 100) sq \n"
+        + "on sq.l_orderkey = o.o_orderkey";
+
+    testBuilder()
+    .sqlQuery(query)
+    .ordered()
+    .optionSettingQueriesForTestQuery("alter system set `planner.slice_target` = 1000")
+    .baselineColumns("cnt")
+    .baselineValues(100l)
+    .build().run();
+  }
+
 }