This is an automated email from the ASF dual-hosted git repository.
jtuglu1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 4e517615b82 feat: add query/segments/count metric to data nodes
(#19624)
4e517615b82 is described below
commit 4e517615b827eec7f7f2c4cce67e1f3c78685a3b
Author: jtuglu1 <[email protected]>
AuthorDate: Mon Jun 29 17:27:17 2026 -0700
feat: add query/segments/count metric to data nodes (#19624)
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.
---
docs/operations/metrics.md | 4 +-
.../src/main/resources/defaultWhiteListMap.json | 4 +
.../main/resources/defaultMetricDimensions.json | 9 +-
.../src/main/resources/defaultWhiteListMap.json | 4 +
.../src/main/resources/defaultMetrics.json | 1 +
.../main/resources/defaultMetricDimensions.json | 1 +
.../apache/druid/query/DefaultQueryMetrics.java | 1 +
.../java/org/apache/druid/query/QueryMetrics.java | 11 +-
.../resources/loggingEmitterAllowedMetrics.json | 3 +-
.../resources/loggingEmitterAllowedMetrics.json | 1 +
.../appenderator/SinkQuerySegmentWalker.java | 56 ++++++++-
.../org/apache/druid/server/ServerManager.java | 1 +
.../apache/druid/query/EmittingQueryMetrics.java | 134 +++++++++++++++++++++
.../timeseries/EmittingTimeseriesQueryMetrics.java | 79 ++++++++++++
.../appenderator/StreamAppenderatorTest.java | 76 +++++++++++-
.../appenderator/StreamAppenderatorTester.java | 11 +-
.../server/coordination/ServerManagerTest.java | 27 ++++-
17 files changed, 409 insertions(+), 14 deletions(-)
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index 8fa59a1a55f..a3390f86a2a 100644
--- 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. [...]
+|`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 [...]
|`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
index 8ac86ed46fc..88dee4c11cc 100644
---
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
index bc0c451870f..3a681d25cd9 100644
---
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"
}
-}
\ No newline at end of file
+}
diff --git
a/extensions-contrib/graphite-emitter/src/main/resources/defaultWhiteListMap.json
b/extensions-contrib/graphite-emitter/src/main/resources/defaultWhiteListMap.json
index 44bd5ef8db9..e0713e3663c 100644
---
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
index bbb4d7b11a5..e582958e768 100644
---
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
index 828031cfb11..9eae1dd5a62 100644
---
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
index f3a4539f780..cc994f72a18 100644
--- 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
index ebe99476062..a6a14846be5 100644
--- 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 @@ public interface QueryMetrics<QueryType extends Query<?>>
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
index 6349c04b5dd..9bcdb712952 100644
--- 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": []
-}
\ No newline at end of file
+}
diff --git a/processing/src/test/resources/loggingEmitterAllowedMetrics.json
b/processing/src/test/resources/loggingEmitterAllowedMetrics.json
index 06ff68822b3..3d3200362f0 100644
--- 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
index 2b5a6210b2c..b50939c85cd 100644
---
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 javax.annotation.Nullable;
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 class SinkQuerySegmentWalker implements
QuerySegmentWalker
// 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 class SinkQuerySegmentWalker implements
QuerySegmentWalker
allRunners.put(descriptor, Collections.singletonList(new
NoopQueryRunner<>()));
} else {
allSegmentReferences.addAll(sinkSegmentReferences);
+ queriedSinks.add(sinkSegmentId);
allRunners.put(
descriptor,
@@ -354,6 +359,8 @@ public class SinkQuerySegmentWalker implements
QuerySegmentWalker
);
}
+ 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 class SinkQuerySegmentWalker implements
QuerySegmentWalker
),
segmentMetricsAccumulator,
Collections.emptySet(),
- null
+ null,
+ segmentCount
),
toolChest,
emitter,
@@ -452,12 +460,15 @@ public class SinkQuerySegmentWalker implements
QuerySegmentWalker
}
/**
- * 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 @@ public class SinkQuerySegmentWalker implements
QuerySegmentWalker
private final Set<String> metricsToCompute;
@Nullable
private final String segmentId;
+ @Nullable
+ private final Integer segmentCount;
private final long creationTimeNs;
private SinkMetricsEmittingQueryRunner(
@@ -480,6 +493,27 @@ public class SinkQuerySegmentWalker implements
QuerySegmentWalker
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;
@@ -487,6 +521,7 @@ public class SinkQuerySegmentWalker implements
QuerySegmentWalker
this.segmentMetricsAccumulator = segmentMetricsAccumulator;
this.metricsToCompute = metricsToCompute;
this.segmentId = segmentId;
+ this.segmentCount = segmentCount;
this.creationTimeNs = System.nanoTime();
}
@@ -544,6 +579,21 @@ public class SinkQuerySegmentWalker implements
QuerySegmentWalker
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
index 60e854f0f6c..1e085be7854 100644
--- 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 class ServerManager implements QuerySegmentWalker
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
new file mode 100644
index 00000000000..a0996b77723
--- /dev/null
+++ b/server/src/test/java/org/apache/druid/query/EmittingQueryMetrics.java
@@ -0,0 +1,134 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Test-only {@link QueryMetrics} that emits every metric {@link
DefaultQueryMetrics} disables by default.
+ * Production keeps those methods as no-ops, but tests that need to assert on
any of them can use this implementation
+ * (or {@link Factory}) instead.
+ */
+public class EmittingQueryMetrics<QueryType extends Query<?>> extends
DefaultQueryMetrics<QueryType>
+{
+ @Override
+ public QueryMetrics<QueryType> reportBackPressureTime(long timeNs)
+ {
+ return reportMetric("query/node/backpressure",
TimeUnit.NANOSECONDS.toMillis(timeNs));
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportBitmapConstructionTime(long timeNs)
+ {
+ return reportMetric("query/segment/bitmap/constructionTime",
TimeUnit.NANOSECONDS.toMillis(timeNs));
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportSegmentRows(long numRows)
+ {
+ return reportMetric("query/segment/rows", numRows);
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportPreFilteredRows(long numRows)
+ {
+ return reportMetric("query/segment/preFilteredRows", numRows);
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportParallelMergeParallelism(int
parallelism)
+ {
+ return reportMetric("query/parallelMerge/parallelism", parallelism);
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportParallelMergeInputSequences(long
numSequences)
+ {
+ return reportMetric("query/parallelMerge/inputSequences", numSequences);
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportParallelMergeInputRows(long numRows)
+ {
+ return reportMetric("query/parallelMerge/inputRows", numRows);
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportParallelMergeOutputRows(long numRows)
+ {
+ return reportMetric("query/parallelMerge/outputRows", numRows);
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportParallelMergeTaskCount(long numTasks)
+ {
+ return reportMetric("query/parallelMerge/taskCount", numTasks);
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportParallelMergeTotalCpuTime(long timeNs)
+ {
+ return reportMetric("query/parallelMerge/cpuTime",
TimeUnit.NANOSECONDS.toMicros(timeNs));
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportParallelMergeTotalTime(long timeNs)
+ {
+ return reportMetric("query/parallelMerge/time",
TimeUnit.NANOSECONDS.toMillis(timeNs));
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportParallelMergeFastestPartitionTime(long
timeNs)
+ {
+ return reportMetric("query/parallelMerge/fastestPartitionTime",
TimeUnit.NANOSECONDS.toMillis(timeNs));
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportParallelMergeSlowestPartitionTime(long
timeNs)
+ {
+ return reportMetric("query/parallelMerge/slowestPartitionTime",
TimeUnit.NANOSECONDS.toMillis(timeNs));
+ }
+
+ @Override
+ public QueryMetrics<QueryType> reportQueriedSegmentCount(long segmentCount)
+ {
+ return reportMetric(QUERY_SEGMENTS_COUNT, segmentCount);
+ }
+
+ /**
+ * {@link GenericQueryMetricsFactory} producing {@link EmittingQueryMetrics}.
+ */
+ public static class Factory implements GenericQueryMetricsFactory
+ {
+ @Override
+ public QueryMetrics<Query<?>> makeMetrics(Query<?> query)
+ {
+ final EmittingQueryMetrics<Query<?>> queryMetrics = new
EmittingQueryMetrics<>();
+ queryMetrics.query(query);
+ return queryMetrics;
+ }
+
+ @Override
+ public QueryMetrics<Query<?>> makeMetrics()
+ {
+ return new EmittingQueryMetrics<>();
+ }
+ }
+}
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
new file mode 100644
index 00000000000..e2d7b861233
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/query/timeseries/EmittingTimeseriesQueryMetrics.java
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.query.timeseries;
+
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.query.EmittingQueryMetrics;
+
+/**
+ * {@link TimeseriesQueryMetrics} counterpart of {@link EmittingQueryMetrics}:
it inherits the emitting overrides for
+ * every metric {@code DefaultQueryMetrics} disables by default, and adds the
Timeseries-specific dimensions (mirroring
+ * {@link DefaultTimeseriesQueryMetrics}) so it can be used by {@link
TimeseriesQueryQueryToolChest} in tests.
+ */
+public class EmittingTimeseriesQueryMetrics extends
EmittingQueryMetrics<TimeseriesQuery>
+ implements TimeseriesQueryMetrics
+{
+ @Override
+ public void query(TimeseriesQuery query)
+ {
+ super.query(query);
+ limit(query);
+ numMetrics(query);
+ numComplexMetrics(query);
+ granularity(query);
+ }
+
+ @Override
+ public void limit(TimeseriesQuery query)
+ {
+ setDimension("limit", String.valueOf(query.getLimit()));
+ }
+
+ @Override
+ public void numMetrics(TimeseriesQuery query)
+ {
+ setDimension("numMetrics",
String.valueOf(query.getAggregatorSpecs().size()));
+ }
+
+ @Override
+ public void numComplexMetrics(TimeseriesQuery query)
+ {
+ int numComplexAggs =
DruidMetrics.findNumComplexAggs(query.getAggregatorSpecs());
+ setDimension("numComplexMetrics", String.valueOf(numComplexAggs));
+ }
+
+ @Override
+ public void granularity(TimeseriesQuery query)
+ {
+ // Don't emit by default
+ }
+
+ /**
+ * {@link TimeseriesQueryMetricsFactory} producing {@link
EmittingTimeseriesQueryMetrics}.
+ */
+ public static class Factory implements TimeseriesQueryMetricsFactory
+ {
+ @Override
+ public TimeseriesQueryMetrics makeMetrics()
+ {
+ return new EmittingTimeseriesQueryMetrics();
+ }
+ }
+}
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
index 904d99b7cde..bc5b3c6de36 100644
---
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
@@ -2294,10 +2294,84 @@ public class StreamAppenderatorTest extends
InitializedNullHandlingTest
}
}
+ @Test
+ public void testQueryByIntervalsEmitsQueriedSegmentCountMetric() throws
Exception
+ {
+ try (
+ StubServiceEmitter serviceEmitter = new StubServiceEmitter();
+ final StreamAppenderatorTester tester =
+ new StreamAppenderatorTester.Builder().maxRowsInMemory(100)
+
.basePersistDirectory(temporaryFolder.newFolder())
+
.withServiceEmitter(serviceEmitter)
+ .build()) {
+ final Appenderator appenderator = tester.getAppenderator();
+
+ appenderator.startJob();
+ appenderator.add(IDENTIFIERS.get(0), ir("2000", "foo", 1),
Suppliers.ofInstance(Committers.nil()));
+ appenderator.add(IDENTIFIERS.get(1), ir("2000", "foo", 4),
Suppliers.ofInstance(Committers.nil()));
+
+ final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
+
.dataSource(StreamAppenderatorTester.DATASOURCE)
+
.intervals(ImmutableList.of(Intervals.of("2000/2001")))
+ .aggregators(
+ Collections.singletonList(
+ new
LongSumAggregatorFactory("count", "count")
+ )
+ )
+ .granularity(Granularities.DAY)
+ .build();
+
+ QueryPlus.wrap(query).run(appenderator,
ResponseContext.createEmpty()).toList();
+
+ serviceEmitter.verifyEmitted(DefaultQueryMetrics.QUERY_SEGMENTS_COUNT,
1);
+ serviceEmitter.verifyValue(DefaultQueryMetrics.QUERY_SEGMENTS_COUNT, 2L);
+ }
+ }
+
+ @Test
+ public void testQueryEmitsQueriedSegmentCountCountsSinksNotHydrants() throws
Exception
+ {
+ try (
+ StubServiceEmitter serviceEmitter = new StubServiceEmitter();
+ final StreamAppenderatorTester tester =
+ new StreamAppenderatorTester.Builder().maxRowsInMemory(100)
+
.basePersistDirectory(temporaryFolder.newFolder())
+
.withServiceEmitter(serviceEmitter)
+ .build()) {
+ final Appenderator appenderator = tester.getAppenderator();
+
+ appenderator.startJob();
+ // Persisting between adds to the same identifier swaps the current
FireHydrant for a new one, so this single
+ // Sink ends up with multiple hydrants. query/segments/count must report
1 (the Sink/segment), not the hydrant count.
+ appenderator.add(IDENTIFIERS.get(0), ir("2000", "foo", 1),
Suppliers.ofInstance(Committers.nil()));
+ appenderator.persistAll(Committers.nil()).get();
+ appenderator.add(IDENTIFIERS.get(0), ir("2000", "bar", 2),
Suppliers.ofInstance(Committers.nil()));
+ appenderator.persistAll(Committers.nil()).get();
+ appenderator.add(IDENTIFIERS.get(0), ir("2000", "baz", 3),
Suppliers.ofInstance(Committers.nil()));
+
+ final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
+
.dataSource(StreamAppenderatorTester.DATASOURCE)
+
.intervals(ImmutableList.of(Intervals.of("2000/2001")))
+ .aggregators(
+ Collections.singletonList(
+ new
LongSumAggregatorFactory("count", "count")
+ )
+ )
+ .granularity(Granularities.DAY)
+ .build();
+
+ QueryPlus.wrap(query).run(appenderator,
ResponseContext.createEmpty()).toList();
+
+ serviceEmitter.verifyEmitted(DefaultQueryMetrics.QUERY_SEGMENTS_COUNT,
1);
+ serviceEmitter.verifyValue(DefaultQueryMetrics.QUERY_SEGMENTS_COUNT, 1L);
+ }
+ }
+
private void verifySinkMetrics(StubServiceEmitter emitter, Set<String>
segmentIds)
{
- int segments = segmentIds.size();
+ final int segments = segmentIds.size();
emitter.verifyEmitted(DefaultQueryMetrics.QUERY_CPU_TIME, 1);
+ emitter.verifyEmitted(DefaultQueryMetrics.QUERY_SEGMENTS_COUNT, 1);
Assert.assertEquals(segments,
emitter.getMetricEvents(DefaultQueryMetrics.QUERY_SEGMENT_TIME).size());
Assert.assertEquals(segments,
emitter.getMetricEvents(DefaultQueryMetrics.QUERY_SEGMENT_AND_CACHE_TIME).size());
Assert.assertEquals(segments,
emitter.getMetricEvents(DefaultQueryMetrics.QUERY_WAIT_TIME).size());
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
index 5d659199108..93571eba5ba 100644
---
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
@@ -39,8 +39,8 @@ import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.java.util.emitter.core.NoopEmitter;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.math.expr.ExprMacroTable;
-import org.apache.druid.query.DefaultGenericQueryMetricsFactory;
import org.apache.druid.query.DefaultQueryRunnerFactoryConglomerate;
+import org.apache.druid.query.EmittingQueryMetrics;
import org.apache.druid.query.ForwardingQueryProcessingPool;
import org.apache.druid.query.QueryRunnerTestHelper;
import org.apache.druid.query.aggregation.CountAggregatorFactory;
@@ -53,6 +53,7 @@ import org.apache.druid.query.scan.ScanQueryConfig;
import org.apache.druid.query.scan.ScanQueryEngine;
import org.apache.druid.query.scan.ScanQueryQueryToolChest;
import org.apache.druid.query.scan.ScanQueryRunnerFactory;
+import org.apache.druid.query.timeseries.EmittingTimeseriesQueryMetrics;
import org.apache.druid.query.timeseries.TimeseriesQuery;
import org.apache.druid.query.timeseries.TimeseriesQueryEngine;
import org.apache.druid.query.timeseries.TimeseriesQueryQueryToolChest;
@@ -209,12 +210,12 @@ public class StreamAppenderatorTester implements
AutoCloseable
indexMerger,
DefaultQueryRunnerFactoryConglomerate.buildFromQueryRunnerFactories(ImmutableMap.of(
TimeseriesQuery.class, new TimeseriesQueryRunnerFactory(
- new TimeseriesQueryQueryToolChest(),
+ new TimeseriesQueryQueryToolChest(new
EmittingTimeseriesQueryMetrics.Factory()),
new TimeseriesQueryEngine(),
QueryRunnerTestHelper.NOOP_QUERYWATCHER
),
ScanQuery.class, new ScanQueryRunnerFactory(
- new
ScanQueryQueryToolChest(DefaultGenericQueryMetricsFactory.instance()),
+ new ScanQueryQueryToolChest(new
EmittingQueryMetrics.Factory()),
new ScanQueryEngine(),
new ScanQueryConfig()
)
@@ -252,12 +253,12 @@ public class StreamAppenderatorTester implements
AutoCloseable
indexMerger,
DefaultQueryRunnerFactoryConglomerate.buildFromQueryRunnerFactories(ImmutableMap.of(
TimeseriesQuery.class, new TimeseriesQueryRunnerFactory(
- new TimeseriesQueryQueryToolChest(),
+ new TimeseriesQueryQueryToolChest(new
EmittingTimeseriesQueryMetrics.Factory()),
new TimeseriesQueryEngine(),
QueryRunnerTestHelper.NOOP_QUERYWATCHER
),
ScanQuery.class, new ScanQueryRunnerFactory(
- new
ScanQueryQueryToolChest(DefaultGenericQueryMetricsFactory.instance()),
+ new ScanQueryQueryToolChest(new
EmittingQueryMetrics.Factory()),
new ScanQueryEngine(),
new ScanQueryConfig()
)
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
index 92b2dcfc8ab..f4dfa43e8e2 100644
---
a/server/src/test/java/org/apache/druid/server/coordination/ServerManagerTest.java
+++
b/server/src/test/java/org/apache/druid/server/coordination/ServerManagerTest.java
@@ -52,10 +52,12 @@ import
org.apache.druid.java.util.common.guava.YieldingAccumulator;
import org.apache.druid.java.util.common.guava.YieldingSequenceBase;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
import org.apache.druid.query.DataSource;
import org.apache.druid.query.DefaultQueryMetrics;
import org.apache.druid.query.DefaultQueryRunnerFactoryConglomerate;
import org.apache.druid.query.Druids;
+import org.apache.druid.query.EmittingQueryMetrics;
import org.apache.druid.query.ForwardingQueryProcessingPool;
import org.apache.druid.query.NoopQueryRunner;
import org.apache.druid.query.Query;
@@ -482,6 +484,29 @@ public class ServerManagerTest
Assert.assertSame(NoopQueryRunner.class, queryRunner.getClass());
}
+ @Test
+ public void testGetQueryRunnerForIntervalsEmitsQueriedSegmentCountMetric()
+ {
+ final StubServiceEmitter stubServiceEmitter =
StubServiceEmitter.createStarted();
+ serviceEmitter = stubServiceEmitter;
+ factory = new MyQueryRunnerFactory(new CountDownLatch(0), new
CountDownLatch(0), new CountDownLatch(0));
+ conglomerate =
DefaultQueryRunnerFactoryConglomerate.buildFromQueryRunnerFactories(ImmutableMap.of(
+ SearchQuery.class,
+ factory
+ ));
+ serverManager =
Guice.createInjector(BoundFieldModule.of(this)).getInstance(ServerManager.class);
+
+ final Interval interval = Intervals.of("P2d/2011-04-02");
+ final SearchQuery query = searchQuery("test", interval, Granularities.DAY);
+
+ serverManager.getQueryRunnerForIntervals(query, ImmutableList.of(interval))
+ .run(QueryPlus.wrap(query))
+ .toList();
+
+ stubServiceEmitter.verifyEmitted(DefaultQueryMetrics.QUERY_SEGMENTS_COUNT,
1);
+ stubServiceEmitter.verifyValue(DefaultQueryMetrics.QUERY_SEGMENTS_COUNT,
2L);
+ }
+
@Test
public void
testGetQueryRunnerForSegments_whenTimelineIsMissingReportingMissingSegmentsOnQueryDataSource()
{
@@ -830,7 +855,7 @@ public class ServerManagerTest
@Override
public QueryMetrics<Query<?>> makeMetrics(QueryType query)
{
- return new DefaultQueryMetrics<>();
+ return new EmittingQueryMetrics<>();
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
