Groupby Metrics

benchmarks/src/test/java/org/apache/druid/benchmark/GroupByDeserializationBenchmark.java

@@ -36,7 +36,6 @@
 import org.apache.druid.query.groupby.GroupByQuery;
 import org.apache.druid.query.groupby.GroupByQueryConfig;
 import org.apache.druid.query.groupby.GroupByQueryQueryToolChest;
-import org.apache.druid.query.groupby.GroupByStatsProvider;
 import org.apache.druid.query.groupby.ResultRow;
 import org.apache.druid.segment.TestHelper;
 import org.apache.druid.segment.column.ColumnType;
@@ -114,7 +113,7 @@ public boolean isIntermediateResultAsMapCompat()
         },
         null,
         null,
-        new GroupByStatsProvider()
+        null
     );
 
     decoratedMapper = groupByQueryQueryToolChest.decorateObjectMapper(undecoratedMapper, sqlQuery);

benchmarks/src/test/java/org/apache/druid/benchmark/GroupByTypeInterfaceBenchmark.java

@@ -54,7 +54,6 @@
 import org.apache.druid.query.groupby.GroupByQueryQueryToolChest;
 import org.apache.druid.query.groupby.GroupByQueryRunnerFactory;
 import org.apache.druid.query.groupby.GroupByResourcesReservationPool;
-import org.apache.druid.query.groupby.GroupByStatsProvider;
 import org.apache.druid.query.groupby.GroupingEngine;
 import org.apache.druid.query.groupby.ResultRow;
 import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
@@ -369,7 +368,6 @@ public String getFormatString()
     };
 
     final Supplier<GroupByQueryConfig> configSupplier = Suppliers.ofInstance(config);
-    final GroupByStatsProvider groupByStatsProvider = new GroupByStatsProvider();
     final GroupByResourcesReservationPool groupByResourcesReservationPool =
         new GroupByResourcesReservationPool(mergePool, config);
     final GroupingEngine groupingEngine = new GroupingEngine(
@@ -378,8 +376,7 @@ public String getFormatString()
         groupByResourcesReservationPool,
         TestHelper.makeJsonMapper(),
         new ObjectMapper(new SmileFactory()),
-        QueryBenchmarkUtil.NOOP_QUERYWATCHER,
-        groupByStatsProvider
+        QueryBenchmarkUtil.NOOP_QUERYWATCHER
     );
 
     factory = new GroupByQueryRunnerFactory(

benchmarks/src/test/java/org/apache/druid/benchmark/query/CachingClusteredClientBenchmark.java

@@ -81,7 +81,6 @@
 import org.apache.druid.query.groupby.GroupByQueryRunnerFactory;
 import org.apache.druid.query.groupby.GroupByQueryRunnerTest;
 import org.apache.druid.query.groupby.GroupByResourcesReservationPool;
-import org.apache.druid.query.groupby.GroupByStatsProvider;
 import org.apache.druid.query.groupby.GroupingEngine;
 import org.apache.druid.query.groupby.ResultRow;
 import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
@@ -346,7 +345,6 @@ private static GroupByQueryRunnerFactory makeGroupByQueryRunnerFactory(
         bufferSupplier,
         processingConfig.getNumMergeBuffers()
     );
-    final GroupByStatsProvider groupByStatsProvider = new GroupByStatsProvider();
     final GroupByResourcesReservationPool groupByResourcesReservationPool =
         new GroupByResourcesReservationPool(mergeBufferPool, config);
     final GroupingEngine groupingEngine = new GroupingEngine(
@@ -355,8 +353,7 @@ private static GroupByQueryRunnerFactory makeGroupByQueryRunnerFactory(
         groupByResourcesReservationPool,
         mapper,
         mapper,
-        QueryRunnerTestHelper.NOOP_QUERYWATCHER,
-        groupByStatsProvider
+        QueryRunnerTestHelper.NOOP_QUERYWATCHER
     );
     final GroupByQueryQueryToolChest toolChest = new GroupByQueryQueryToolChest(groupingEngine, groupByResourcesReservationPool);
     return new GroupByQueryRunnerFactory(groupingEngine, toolChest, bufferPool);

benchmarks/src/test/java/org/apache/druid/benchmark/query/GroupByBenchmark.java

@@ -65,7 +65,6 @@
 import org.apache.druid.query.groupby.GroupByQueryQueryToolChest;
 import org.apache.druid.query.groupby.GroupByQueryRunnerFactory;
 import org.apache.druid.query.groupby.GroupByResourcesReservationPool;
-import org.apache.druid.query.groupby.GroupByStatsProvider;
 import org.apache.druid.query.groupby.GroupingEngine;
 import org.apache.druid.query.groupby.ResultRow;
 import org.apache.druid.query.groupby.orderby.DefaultLimitSpec;
@@ -486,7 +485,6 @@ public String getFormatString()
     };
 
     final Supplier<GroupByQueryConfig> configSupplier = Suppliers.ofInstance(config);
-    final GroupByStatsProvider groupByStatsProvider = new GroupByStatsProvider();
     final GroupByResourcesReservationPool groupByResourcesReservationPool =
         new GroupByResourcesReservationPool(mergePool, config);
     final GroupingEngine groupingEngine = new GroupingEngine(
@@ -495,8 +493,7 @@ public String getFormatString()
         groupByResourcesReservationPool,
         TestHelper.makeJsonMapper(),
         new ObjectMapper(new SmileFactory()),
-        QueryBenchmarkUtil.NOOP_QUERYWATCHER,
-        groupByStatsProvider
+        QueryBenchmarkUtil.NOOP_QUERYWATCHER
     );
 
     factory = new GroupByQueryRunnerFactory(

extensions-contrib/virtual-columns/src/test/java/org/apache/druid/segment/MapVirtualColumnGroupByTest.java

@@ -44,7 +44,6 @@
 import org.apache.druid.query.groupby.GroupByQueryQueryToolChest;
 import org.apache.druid.query.groupby.GroupByQueryRunnerFactory;
 import org.apache.druid.query.groupby.GroupByResourcesReservationPool;
-import org.apache.druid.query.groupby.GroupByStatsProvider;
 import org.apache.druid.query.groupby.GroupingEngine;
 import org.apache.druid.query.groupby.ResultRow;
 import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
@@ -74,7 +73,6 @@ public void setup() throws IOException
 
     final BlockingPool<ByteBuffer> mergePool =
         new DefaultBlockingPool<>(() -> ByteBuffer.allocate(1024), 1);
-    final GroupByStatsProvider groupByStatsProvider = new GroupByStatsProvider();
 
     final GroupByResourcesReservationPool groupByResourcesReservationPool =
         new GroupByResourcesReservationPool(mergePool, config);
@@ -110,8 +108,7 @@ public int getNumThreads()
         groupByResourcesReservationPool,
         TestHelper.makeJsonMapper(),
         new DefaultObjectMapper(),
-        QueryRunnerTestHelper.NOOP_QUERYWATCHER,
-        groupByStatsProvider
+        QueryRunnerTestHelper.NOOP_QUERYWATCHER
     );
 
     final GroupByQueryRunnerFactory factory = new GroupByQueryRunnerFactory(

extensions-core/datasketches/src/test/java/org/apache/druid/segment/DatasketchesProjectionTest.java

@@ -62,7 +62,6 @@
 import org.apache.druid.query.groupby.GroupByQuery;
 import org.apache.druid.query.groupby.GroupByQueryConfig;
 import org.apache.druid.query.groupby.GroupByResourcesReservationPool;
-import org.apache.druid.query.groupby.GroupByStatsProvider;
 import org.apache.druid.query.groupby.GroupingEngine;
 import org.apache.druid.query.groupby.ResultRow;
 import org.apache.druid.segment.column.ColumnType;
@@ -279,8 +278,7 @@ public DatasketchesProjectionTest(
         TestHelper.makeJsonMapper(),
         TestHelper.makeSmileMapper(),
         (query, future) -> {
-        },
-        new GroupByStatsProvider()
+        }
     );
   }
 

processing/src/main/java/org/apache/druid/query/MetricsEmittingQueryRunner.java

@@ -34,6 +34,8 @@
  */
 public class MetricsEmittingQueryRunner<T> implements QueryRunner<T>
 {
+  public static final ObjLongConsumer<QueryMetrics<?>> NOOP_METRIC_REPORTER = (metrics, value) -> { };
+
   private final ServiceEmitter emitter;
   private final QueryToolChest<T, ? extends Query<T>> queryToolChest;
   private final QueryRunner<T> queryRunner;

processing/src/main/java/org/apache/druid/query/QueryDataSource.java

@@ -56,7 +56,7 @@ public Set<String> getTableNames()
   }
 
   @JsonProperty
-  public Query getQuery()
+  public Query<?> getQuery()
   {
     return query;
   }

processing/src/main/java/org/apache/druid/query/context/ResponseContext.java

@@ -244,7 +244,7 @@ public Object mergeValues(Object oldValue, Object newValue)
    */
   public static class LongKey extends AbstractKey
   {
-    LongKey(String name, boolean inHeader)
+    public LongKey(String name, boolean inHeader)
     {
       super(name, inHeader, false, Long.class);
     }

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

@@ -22,8 +22,14 @@
 import org.apache.druid.query.DefaultQueryMetrics;
 import org.apache.druid.query.DruidMetrics;
 
+import java.util.concurrent.atomic.LongAdder;
+
 public class DefaultGroupByQueryMetrics extends DefaultQueryMetrics<GroupByQuery> implements GroupByQueryMetrics
 {
+  private final LongAdder mergeBufferAcquisitonTime = new LongAdder();
+  private final LongAdder bytesSpilledToStorage = new LongAdder();
+  private final LongAdder mergeDictionarySize = new LongAdder();
+
   @Override
   public void query(GroupByQuery query)
   {
@@ -58,4 +64,50 @@ public void granularity(GroupByQuery query)
   {
     //Don't emit by default
   }
+
+  @Override
+  public void mergeBufferAcquisitionTime(long mergeBufferAcquisitionTime)
+  {
+    reportMetric("mergeBufferAcquisitonTime", mergeBufferAcquisitionTime);
+  }
+
+  @Override
+  public void bytesSpilledToStorage(long bytesSpilledToStorage)
+  {
+    reportMetric("bytesSpilledToStorage", bytesSpilledToStorage);
+  }
+
+  @Override
+  public void mergeDictionarySize(long mergeDictionarySize)
+  {
+    reportMetric("mergeDictionarySize", mergeDictionarySize);
+  }
+
+  @Override
+  public long getSpilledBytes()
+  {
+    return bytesSpilledToStorage.longValue();
+  }
+
+  @Override
+  public long getMergeDictionarySize()
+  {
+    return mergeDictionarySize.longValue();
+  }
+
+  @Override
+  public long getMergeBufferAcquisitionTime()
+  {
+    return mergeBufferAcquisitonTime.longValue();
+  }
+
+  // TODO: Delete after debugging...
+  @Override
+  public String toString()
+  {
+    return "bytesSpilledToStorage[" + bytesSpilledToStorage.longValue()
+           + "], mergeDictionarySize[" + mergeDictionarySize.longValue()
+           + "], mergeBufferAcquisitonTime[" + mergeBufferAcquisitonTime.longValue()
+           + ']';
+  }
 }

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

@@ -54,4 +54,22 @@ public interface GroupByQueryMetrics extends QueryMetrics<GroupByQuery>
    */
   @PublicApi
   void granularity(GroupByQuery query);
+
+  @PublicApi
+  void mergeBufferAcquisitionTime(long mergeBufferAcquisitionTime);
+
+  @PublicApi
+  void bytesSpilledToStorage(long bytesSpilledToStorage);
+
+  @PublicApi
+  void mergeDictionarySize(long mergeDictionarySize);
+
+  @PublicApi
+  long getSpilledBytes();
+
+  @PublicApi
+  long getMergeDictionarySize();
+
+  @PublicApi
+  long getMergeBufferAcquisitionTime();
 }