fix merging of groupBy subtotal spec results

core/src/main/java/org/apache/druid/java/util/common/collect/Utils.java

@@ -70,4 +70,15 @@ public static <T> List<T> nullableListOf(@Nullable T... elements)
     }
     return list;
   }
+
+  public static <T> boolean isPrefix(List<T> small, List<T> big)
+  {
+    for (int i = 0; i < small.size(); i++) {
+      if (!small.get(i).equals(big.get(i))) {
+        return false;
+      }
+    }
+
+    return true;
+  }
 }

processing/src/main/java/org/apache/druid/query/groupby/GroupByQuery.java

@@ -229,28 +229,20 @@ private List<List<String>> verifySubtotalsSpec(
       List<DimensionSpec> dimensions
   )
   {
-    // if subtotalsSpec exists then validate that all are subsets of dimensions spec and are in same order.
-    // For example if we had {D1, D2, D3} in dimensions spec then
-    // {D2}, {D1, D2}, {D1, D3}, {D2, D3} etc are valid in subtotalsSpec while
-    // {D2, D1} is not as it is not in same order.
-    // {D4} is not as its not a subset.
-    // This restriction as enforced because implementation does sort merge on the results of top-level query
-    // results and expects that ordering of events does not change when dimension columns are removed from
-    // results of top level query.
+    // if subtotalsSpec exists then validate that all are subsets of dimensions spec.
     if (subtotalsSpec != null) {
       for (List<String> subtotalSpec : subtotalsSpec) {
-        int i = 0;
         for (String s : subtotalSpec) {
           boolean found = false;
-          for (; i < dimensions.size(); i++) {
-            if (s.equals(dimensions.get(i).getOutputName())) {
+          for (DimensionSpec ds : dimensions) {
+            if (s.equals(ds.getOutputName())) {
               found = true;
               break;
             }
           }
           if (!found) {
             throw new IAE(
-                "Subtotal spec %s is either not a subset or items are in different order than in dimensions.",
+                "Subtotal spec %s is either not a subset of top level dimensions.",
                 subtotalSpec
             );
           }

processing/src/main/java/org/apache/druid/query/groupby/orderby/DefaultLimitSpec.java

@@ -53,6 +53,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  *
@@ -211,6 +212,15 @@ private ValueType getOrderByType(final OrderByColumnSpec columnSpec, final List<
     throw new ISE("Unknown column in order clause[%s]", columnSpec);
   }
 
+  @Override
+  public LimitSpec filterColumns(Set<String> names)
+  {
+    return new DefaultLimitSpec(
+        columns.stream().filter(c -> names.contains(c.getDimension())).collect(Collectors.toList()),
+        limit
+    );
+  }
+
   private Ordering<ResultRow> makeComparator(
       Object2IntMap<String> rowOrderLookup,
       boolean hasTimestamp,

processing/src/main/java/org/apache/druid/query/groupby/orderby/LimitSpec.java

@@ -28,6 +28,7 @@
 import org.apache.druid.query.groupby.ResultRow;
 
 import javax.annotation.Nullable;
+import java.util.Set;
 
 /**
  *
@@ -53,4 +54,13 @@ static LimitSpec nullToNoopLimitSpec(@Nullable LimitSpec limitSpec)
   Function<Sequence<ResultRow>, Sequence<ResultRow>> build(GroupByQuery query);
 
   LimitSpec merge(LimitSpec other);
+
+  /**
+   * Discard sorting columns not contained in given set. This is used when generating new queries, e.g. to process
+   * subtotal spec in GroupBy query.
+   *
+   * @param names columns names to keep
+   * @return new LimitSpec that works with fitlered set of columns
+   */
+  LimitSpec filterColumns(Set<String> names);
 }

processing/src/main/java/org/apache/druid/query/groupby/orderby/NoopLimitSpec.java

@@ -26,6 +26,8 @@
 import org.apache.druid.query.groupby.GroupByQuery;
 import org.apache.druid.query.groupby.ResultRow;
 
+import java.util.Set;
+
 /**
  *
  */
@@ -57,6 +59,12 @@ public LimitSpec merge(LimitSpec other)
     return other;
   }
 
+  @Override
+  public LimitSpec filterColumns(Set<String> names)
+  {
+    return this;
+  }
+
   @Override
   public String toString()
   {