feat: add query/segments/count metric to data nodes (#19624)

jtuglu1 · web-flow · commit 4e517615b827 · 2026-06-29T17:27:17.000-07:00
Allows for quick queries to understand how many segments a given query issued scans for against a particular data node. Better than counting the unique query/segment/time or similar, especially if you are sampling higher volume metrics like query/segment/time, but still want to have quick way to know stats about # of segments queried per node.
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
@@ -64,7 +64,7 @@ Most metric values reset each emission period, as specified in `druid.monitoring
 |`query/failed/count`|Number of failed queries.|This metric is only available if the `QueryCountStatsMonitor` module is included.| |
 |`query/interrupted/count`|Number of queries interrupted due to cancellation.|This metric is only available if the `QueryCountStatsMonitor` module is included.| |
 |`query/timeout/count`|Number of timed out queries.|This metric is only available if the `QueryCountStatsMonitor` module is included.| |
-|`query/segments/count`|This metric is not enabled by default. See the `QueryMetrics` Interface for reference regarding enabling this metric. Number of segments that will be touched by the query. In the broker, it makes a plan to distribute the query to realtime tasks and historicals based on a snapshot of segment distribution state. If there are some segments moved after this snapshot is created, certain historicals and realtime tasks can report those segments as missing to the broker. The broker will resend the query to the new servers that serve those segments after move. In this case, those segments can be counted more than once in this metric.||Varies|
+|`query/segments/count`|Number of segments that will be touched by the query. The Broker makes a plan to distribute the query to realtime tasks and historicals based on a snapshot of segment distribution state. If there are some segments moved after this snapshot is created, certain historicals and realtime tasks can report those segments as missing to the Broker. The Broker will resend the query to the new servers that serve those segments after move. In this case, those segments can be counted more than once in this metric.||Varies|
 |`query/priority`|Assigned lane and priority, only if Laning strategy is enabled. Refer to [Laning strategies](../configuration/index.md#laning-strategies)|`lane`, `dataSource`, `type`|0|
 |`sqlQuery/time`|Milliseconds taken to complete a SQL query.|`id`, `nativeQueryIds`, `dataSource`, `remoteAddress`, `success`, `engine`, `statusCode`|< 1s|
 |`sqlQuery/planningTimeMs`|Milliseconds taken to plan a SQL to native query.|`id`, `nativeQueryIds`, `dataSource`, `remoteAddress`, `success`, `engine`| |
@@ -107,6 +107,7 @@ Most metric values reset each emission period, as specified in `druid.monitoring
 |Metric|Description|Dimensions|Normal value|
 |------|-----------|----------|------------|
 |`query/time`|Milliseconds taken to complete a query.|<p>Common: `dataSource`, `type`, `interval`, `hasFilters`, `duration`, `context`, `remoteAddress`, `id`, `statusCode`.</p><p> Aggregation Queries: `numMetrics`, `numComplexMetrics`.</p><p> GroupBy: `numDimensions`.</p><p> TopN: `threshold`, `dimension`.</p>|< 1s|
+|`query/segments/count`|Number of segments this Historical scans for a query.|<p>Common: `dataSource`, `type`, `interval`, `hasFilters`, `duration`, `context`, `remoteAddress`, `id`.</p><p> Aggregation Queries: `numMetrics`, `numComplexMetrics`.</p><p> GroupBy: `numDimensions`.</p><p> TopN: `threshold`, `dimension`.</p>|Varies|
 |`query/segment/time`|Milliseconds taken to query individual segment. Includes time to page in the segment from disk.|`id`, `status`, `segment`, `vectorized`.|several hundred milliseconds|
 |`query/wait/time`|Milliseconds spent waiting for a segment to be scanned.|`id`, `segment`|< several hundred milliseconds|
 |`segment/scan/pending`|Number of segments in queue waiting to be scanned.||Close to 0|
@@ -141,6 +142,7 @@ to represent the task ID are deprecated and will be removed in a future release.
 |Metric|Description|Dimensions|Normal value|
 |------|-----------|----------|------------|
 |`query/time`|Milliseconds taken to complete a query.|<p>Common: `dataSource`, `type`, `interval`, `hasFilters`, `duration`, `context`, `remoteAddress`, `id`, `statusCode`.</p><p> Aggregation Queries: `numMetrics`, `numComplexMetrics`.</p><p> GroupBy: `numDimensions`.</p><p> TopN: `threshold`, `dimension`.</p>|< 1s|
+|`query/segments/count`|Number of segments this Peon scans for a query.|<p>Common: `dataSource`, `type`, `interval`, `hasFilters`, `duration`, `context`, `remoteAddress`, `id`.</p><p> Aggregation Queries: `numMetrics`, `numComplexMetrics`.</p><p> GroupBy: `numDimensions`. </p><p>TopN: `threshold`, `dimension`.</p>|Varies|
 |`query/wait/time`|Milliseconds spent waiting for a segment to be scanned.|`id`, `segment`|several hundred milliseconds|
 |`segment/scan/pending`|Number of segments in queue waiting to be scanned.||Close to 0|
 |`segment/scan/active`|Number of segments currently scanned. This metric also indicates how many threads from `druid.processing.numThreads` are currently being used.||Close to `druid.processing.numThreads`|
diff --git a/extensions-contrib/ambari-metrics-emitter/src/main/resources/defaultWhiteListMap.json b/extensions-contrib/ambari-metrics-emitter/src/main/resources/defaultWhiteListMap.json
@@ -34,6 +34,10 @@
     "dataSource",
     "type"
   ],
+  "query/segments/count": [
+    "dataSource",
+    "type"
+  ],
   "query/segment/time": [
     "dataSource",
     "type"
diff --git a/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json b/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json
@@ -21,6 +21,13 @@
     "type": "timer",
     "timeUnit": "MILLISECONDS"
   },
+  "query/segments/count": {
+    "dimensions": [
+      "dataSource",
+      "type"
+    ],
+    "type": "counter"
+  },
   "query/segment/time": {
     "dimensions": [],
     "type": "timer",
@@ -515,4 +522,4 @@
     ],
     "type": "gauge"
   }
-}
+}
diff --git a/extensions-contrib/graphite-emitter/src/main/resources/defaultWhiteListMap.json b/extensions-contrib/graphite-emitter/src/main/resources/defaultWhiteListMap.json
@@ -21,6 +21,10 @@
     "dataSource",
     "type"
   ],
+  "query/segments/count": [
+    "dataSource",
+    "type"
+  ],
   "query/segment/time": [
     "dataSource",
     "type"
diff --git a/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json b/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
@@ -28,6 +28,7 @@
   "schemacache/inTransitSMQPublishedResults/count" : { "dimensions" : [], "type" : "count", "help": "Number of segments for which schema is cached after back filling in the database."},
   "serverview/sync/healthy" : { "dimensions" : ["server"], "type" : "gauge", "help": "Sync status of the Broker with a segment-loading server such as a Historical or Peon."},
   "serverview/sync/unstableTime" : { "dimensions" : ["server"], "type" : "timer", "conversionFactor": 1000.0, "help": "Time in seconds for which the Broker has been failing to sync with a segment-loading server."},
+  "query/segments/count" : { "dimensions" : ["dataSource", "type"], "type" : "count", "help": "Number of segments this data node scans for a query."},
   "query/segment/time" : { "dimensions" : [], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to query individual segment. Includes time to page in the segment from disk."},
   "query/wait/time" : { "dimensions" : [], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds spent waiting for a segment to be scanned."},
   "segment/scan/pending" : { "dimensions" : [], "type" : "gauge", "help": "Number of segments in queue waiting to be scanned."},
diff --git a/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json b/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json
@@ -5,6 +5,7 @@
   "query/node/ttfb" : { "dimensions" : ["server"], "type" : "timer"},
   "query/node/bytes" : { "dimensions" : ["server"], "type" : "count"},
 
+  "query/segments/count" : { "dimensions" : ["dataSource", "type"], "type" : "count"},
   "query/segment/time" : { "dimensions" : [], "type" : "timer"},
   "query/wait/time" : { "dimensions" : [], "type" : "timer"},
   "segment/scan/pending" : { "dimensions" : [], "type" : "gauge"},
diff --git a/processing/src/main/java/org/apache/druid/query/DefaultQueryMetrics.java b/processing/src/main/java/org/apache/druid/query/DefaultQueryMetrics.java
@@ -46,6 +46,7 @@ public class DefaultQueryMetrics<QueryType extends Query<?>> implements QueryMet
   public static final String QUERY_BYTES = "query/bytes";
   public static final String QUERY_CPU_TIME = "query/cpu/time";
   public static final String QUERY_WAIT_TIME = "query/wait/time";
+  public static final String QUERY_SEGMENTS_COUNT = "query/segments/count";
   public static final String QUERY_SEGMENT_TIME = "query/segment/time";
   public static final String QUERY_SEGMENT_AND_CACHE_TIME = "query/segmentAndCache/time";
   public static final String QUERY_RESULT_CACHE_HIT = "query/resultCache/hit";
diff --git a/processing/src/main/java/org/apache/druid/query/QueryMetrics.java b/processing/src/main/java/org/apache/druid/query/QueryMetrics.java
@@ -329,7 +329,16 @@ default void filterBundle(FilterBundle.BundleInfo bundleInfo)
   QueryMetrics<QueryType> reportQueryBytes(long byteCount);
 
   /**
-   * Registers "segments queried count" metric.
+   * Registers the {@code query/segments/count} metric, the number of segments touched by the query.
+   *
+   * Emitted once per query. The meaning of the metric depends on the emitting process:
+   * <ul>
+   *   <li>On the Broker, it is the number of segments in the planned query distribution (a snapshot-based
+   *       count that may double-count segments that moved and were re-fetched).</li>
+   *   <li>On a data node (Historical, Peon/realtime), it is the number of segments that node actually scanned
+   *       for the query.</li>
+   * </ul>
+   * The two are disambiguated by the emitting service/host, not by the metric name.
    */
   QueryMetrics<QueryType> reportQueriedSegmentCount(long segmentCount);
 
diff --git a/processing/src/main/resources/loggingEmitterAllowedMetrics.json b/processing/src/main/resources/loggingEmitterAllowedMetrics.json
@@ -134,6 +134,7 @@
     "query/node/bytes": [],
     "query/node/time": [],
     "query/node/ttfb": [],
+    "query/segments/count": [],
     "query/segment/time": [],
     "query/segmentAndCache/time": [],
     "query/success/count": [],
@@ -249,4 +250,4 @@
     "tier/total/capacity": [],
     "zk/connected": [],
     "zk/reconnect/time": []
-}
+}
diff --git a/processing/src/test/resources/loggingEmitterAllowedMetrics.json b/processing/src/test/resources/loggingEmitterAllowedMetrics.json
@@ -134,6 +134,7 @@
   "query/node/bytes": [],
   "query/node/time": [],
   "query/node/ttfb": [],
+  "query/segments/count": [],
   "query/segment/time": [],
   "query/segmentAndCache/time": [],
   "query/success/count": [],
diff --git a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/SinkQuerySegmentWalker.java b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/SinkQuerySegmentWalker.java
@@ -82,6 +82,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -210,6 +211,9 @@ public <T> QueryRunner<T> getQueryRunnerForSegments(final Query<T> query, final
     // QueryRunner returned by this method is closed. (We can't do the acquisition and releasing at the level of
     // each FireHydrant's runner, since then it wouldn't be properly all-or-nothing on a per-Sink basis.)
     final List<SinkSegmentReference> allSegmentReferences = new ArrayList<>();
+    // Distinct Sinks (Druid segments) actually queried by this node. Tracked separately from
+    // allSegmentReferences, which holds one reference per FireHydrant (subsegment) and would overcount.
+    final Set<SegmentId> queriedSinks = new HashSet<>();
     final Map<SegmentDescriptor, SegmentId> segmentIdMap = new HashMap<>();
     final LinkedHashMap<SegmentDescriptor, List<QueryRunner<T>>> allRunners = new LinkedHashMap<>();
     final ConcurrentHashMap<String, SinkMetricsEmittingQueryRunner.SegmentMetrics> segmentMetricsAccumulator = new ConcurrentHashMap<>();
@@ -241,6 +245,7 @@ public <T> QueryRunner<T> getQueryRunnerForSegments(final Query<T> query, final
           allRunners.put(descriptor, Collections.singletonList(new NoopQueryRunner<>()));
         } else {
           allSegmentReferences.addAll(sinkSegmentReferences);
+          queriedSinks.add(sinkSegmentId);
 
           allRunners.put(
               descriptor,
@@ -354,6 +359,8 @@ public <T> QueryRunner<T> getQueryRunnerForSegments(final Query<T> query, final
         );
       }
 
+      final int segmentCount = queriedSinks.size();
+
       // 1) Populate resource id to the query
       // 2) Merge results using the toolChest, finalize if necessary.
       // 3) Measure CPU time of that operation.
@@ -370,7 +377,8 @@ public <T> QueryRunner<T> getQueryRunnerForSegments(final Query<T> query, final
                       ),
                       segmentMetricsAccumulator,
                       Collections.emptySet(),
-                      null
+                      null,
+                      segmentCount
                   ),
                   toolChest,
                   emitter,
@@ -452,12 +460,15 @@ public static String makeHydrantCacheIdentifier(final SegmentId segmentId, final
   }
 
   /**
-   * This class is responsible for emitting query/segment/time, query/wait/time and query/segmentAndCache/Time metrics for a Sink.
+   * This class is responsible for emitting query/segment/time, query/wait/time, query/segmentAndCache/Time and
+   * query/segments/count metrics for a Sink.
    * It accumulates query/segment/time and query/segmentAndCache/time metric for each FireHydrant at the level of Sink.
    * query/wait/time metric is the time taken to process the first FireHydrant for the Sink.
    * <p>
    * This class operates in two distinct modes based on whether {@link SinkMetricsEmittingQueryRunner#segmentId} is null or non-null.
-   * When segmentId is non-null, it accumulates the metrics. When segmentId is null, it emits the accumulated metrics.
+   * When segmentId is non-null, it accumulates the per-segment metrics. When segmentId is null, it emits the accumulated
+   * per-segment metrics. Unlike those, query/segments/count is a single per-query total (the number of Sinks scanned by
+   * this node), so it is emitted exactly once by the segmentId-null runner that receives a non-null {@code segmentCount}.
    * <p>
    * This class is derived from {@link org.apache.druid.query.MetricsEmittingQueryRunner}.
    */
@@ -470,6 +481,8 @@ private static class SinkMetricsEmittingQueryRunner<T> implements QueryRunner<T>
     private final Set<String> metricsToCompute;
     @Nullable
     private final String segmentId;
+    @Nullable
+    private final Integer segmentCount;
     private final long creationTimeNs;
 
     private SinkMetricsEmittingQueryRunner(
@@ -480,13 +493,35 @@ private SinkMetricsEmittingQueryRunner(
         Set<String> metricsToCompute,
         @Nullable String segmentId
     )
+    {
+      this(
+          emitter,
+          queryToolChest,
+          queryRunner,
+          segmentMetricsAccumulator,
+          metricsToCompute,
+          segmentId,
+          null
+      );
+    }
+
+    private SinkMetricsEmittingQueryRunner(
+        ServiceEmitter emitter,
+        QueryToolChest<T, ? extends Query<T>> queryToolChest,
+        QueryRunner<T> queryRunner,
+        ConcurrentHashMap<String, SegmentMetrics> segmentMetricsAccumulator,
+        Set<String> metricsToCompute,
+        @Nullable String segmentId,
+        @Nullable Integer segmentCount
+    )
     {
       this.emitter = emitter;
       this.queryToolChest = queryToolChest;
       this.queryRunner = queryRunner;
       this.segmentMetricsAccumulator = segmentMetricsAccumulator;
       this.metricsToCompute = metricsToCompute;
       this.segmentId = segmentId;
+      this.segmentCount = segmentCount;
       this.creationTimeNs = System.nanoTime();
     }
 
@@ -544,6 +579,21 @@ public void after(boolean isDone, Throwable thrown)
                     log.error(e, "Failed to emit metrics for segment[%s]", segmentAndMetrics.getKey());
                   }
                 }
+
+                // Emit total segmentCount once per query
+                if (segmentCount != null) {
+                  final QueryMetrics<?> segmentCountMetrics =
+                      queryPlus.withoutThreadUnsafeState().withQueryMetrics(queryToolChest).getQueryMetrics();
+                  segmentCountMetrics.reportQueriedSegmentCount(segmentCount);
+
+                  try {
+                    segmentCountMetrics.emit(emitter);
+                  }
+                  catch (Exception e) {
+                    // Query should not fail, because of emitter failure. Swallowing the exception.
+                    log.error(e, "Failed to emit queried segment count metric.");
+                  }
+                }
               }
             }
           }
diff --git a/server/src/main/java/org/apache/druid/server/ServerManager.java b/server/src/main/java/org/apache/druid/server/ServerManager.java
@@ -644,6 +644,7 @@ public Sequence<T> run(QueryPlus<T> queryPlus, ResponseContext responseContext)
             segmentMapFn
         );
         closer.registerAll(segmentReferences);
+        queryPlus.getQueryMetrics().reportQueriedSegmentCount(segmentReferences.size()).emit(emitter);
 
         final FunctionalIterable<QueryRunner<T>> queryRunners = getQueryRunnersForSegments(
             query,
diff --git a/server/src/test/java/org/apache/druid/query/EmittingQueryMetrics.java b/server/src/test/java/org/apache/druid/query/EmittingQueryMetrics.java
diff --git a/server/src/test/java/org/apache/druid/query/timeseries/EmittingTimeseriesQueryMetrics.java b/server/src/test/java/org/apache/druid/query/timeseries/EmittingTimeseriesQueryMetrics.java
diff --git a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java
diff --git a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTester.java
diff --git a/server/src/test/java/org/apache/druid/server/coordination/ServerManagerTest.java b/server/src/test/java/org/apache/druid/server/coordination/ServerManagerTest.java
