apache · kfaraz · Apr 28, 2025 · Apr 10, 2025 · Apr 10, 2025 · Apr 10, 2025
diff --git a/...marks/src/test/java/org/apache/druid/benchmark/query/CachingClusteredClientBenchmark.java b/...marks/src/test/java/org/apache/druid/benchmark/query/CachingClusteredClientBenchmark.java
@@ -27,6 +27,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Ordering;
+import org.apache.druid.client.BrokerViewOfCoordinatorConfig;
 import org.apache.druid.client.CachingClusteredClient;
 import org.apache.druid.client.DruidServer;
 import org.apache.druid.client.ImmutableDruidServer;
@@ -53,6 +54,7 @@
 import org.apache.druid.java.util.common.io.Closer;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.query.BaseQuery;
 import org.apache.druid.query.BrokerParallelMergeConfig;
 import org.apache.druid.query.BySegmentQueryRunner;
 import org.apache.druid.query.DefaultQueryRunnerFactoryConglomerate;
@@ -100,8 +102,11 @@
 import org.apache.druid.segment.generator.GeneratorBasicSchemas;
 import org.apache.druid.segment.generator.GeneratorSchemaInfo;
 import org.apache.druid.segment.generator.SegmentGenerator;
+import org.apache.druid.server.ClientQuerySegmentWalker;
 import org.apache.druid.server.QueryStackTests;
+import org.apache.druid.server.ResourceIdPopulatingQueryRunner;
 import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordination.TestCoordinatorClient;
 import org.apache.druid.server.metrics.NoopServiceEmitter;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.DataSegment.PruneSpecsHolder;
@@ -176,6 +181,7 @@ public class CachingClusteredClientBenchmark
   private final QuerySegmentSpec basicSchemaIntervalSpec = new MultipleIntervalSegmentSpec(
       Collections.singletonList(basicSchema.getDataInterval())
   );
+  private final BrokerViewOfCoordinatorConfig filter = new BrokerViewOfCoordinatorConfig(new TestCoordinatorClient());
 
   private final int numProcessingThreads = 4;
 
@@ -222,6 +228,7 @@ public void setup()
           rowsPerSegment
       );
       queryableIndexes.put(dataSegment, index);
+      filter.start();
     }
 
     final DruidProcessingConfig processingConfig = new DruidProcessingConfig()
@@ -368,18 +375,21 @@ public void tearDown() throws IOException
   @OutputTimeUnit(TimeUnit.MICROSECONDS)
   public void timeseriesQuery(Blackhole blackhole)
   {
-    query = Druids.newTimeseriesQueryBuilder()
-                  .dataSource(DATA_SOURCE)
-                  .intervals(basicSchemaIntervalSpec)
-                  .aggregators(new LongSumAggregatorFactory("sumLongSequential", "sumLongSequential"))
-                  .granularity(Granularity.fromString(queryGranularity))
-                  .context(
-                      ImmutableMap.of(
-                          QueryContexts.BROKER_PARALLEL_MERGE_KEY, parallelCombine,
-                          QueryContexts.BROKER_PARALLELISM, parallelism
-                      )
-                  )
-                  .build();
+    Query<?> q = Druids.newTimeseriesQueryBuilder()
+                       .dataSource(DATA_SOURCE)
+                       .intervals(basicSchemaIntervalSpec)
+                       .aggregators(new LongSumAggregatorFactory("sumLongSequential", "sumLongSequential"))
+                       .granularity(Granularity.fromString(queryGranularity))
+                       .context(
+                           ImmutableMap.of(
+                               BaseQuery.QUERY_ID, "BenchmarkQuery",
+                               QueryContexts.BROKER_PARALLEL_MERGE_KEY, parallelCombine,
+                               QueryContexts.BROKER_PARALLELISM, parallelism
+                           )
+                       )
+                       .build();
+
+    query = prepareQuery(q);
 
     final List<Result<TimeseriesResultValue>> results = runQuery();
 
@@ -393,7 +403,7 @@ public void timeseriesQuery(Blackhole blackhole)
   @OutputTimeUnit(TimeUnit.MICROSECONDS)
   public void topNQuery(Blackhole blackhole)
   {
-    query = new TopNQueryBuilder()
+    Query<?> q = new TopNQueryBuilder()
         .dataSource(DATA_SOURCE)
         .intervals(basicSchemaIntervalSpec)
         .dimension(new DefaultDimensionSpec("dimZipf", null))
@@ -403,12 +413,15 @@ public void topNQuery(Blackhole blackhole)
         .threshold(10_000) // we are primarily measuring 'broker' merge time, so collect a significant number of results
         .context(
             ImmutableMap.of(
+                BaseQuery.QUERY_ID, "BenchmarkQuery",
                 QueryContexts.BROKER_PARALLEL_MERGE_KEY, parallelCombine,
                 QueryContexts.BROKER_PARALLELISM, parallelism
             )
         )
         .build();
 
+    query = prepareQuery(q);
+
     final List<Result<TopNResultValue>> results = runQuery();
 
     for (Result<TopNResultValue> result : results) {
@@ -421,7 +434,7 @@ public void topNQuery(Blackhole blackhole)
   @OutputTimeUnit(TimeUnit.MICROSECONDS)
   public void groupByQuery(Blackhole blackhole)
   {
-    query = GroupByQuery
+    Query<?> q = GroupByQuery
         .builder()
         .setDataSource(DATA_SOURCE)
         .setQuerySegmentSpec(basicSchemaIntervalSpec)
@@ -433,19 +446,33 @@ public void groupByQuery(Blackhole blackhole)
         .setGranularity(Granularity.fromString(queryGranularity))
         .setContext(
             ImmutableMap.of(
+                BaseQuery.QUERY_ID, "BenchmarkQuery",
                 QueryContexts.BROKER_PARALLEL_MERGE_KEY, parallelCombine,
                 QueryContexts.BROKER_PARALLELISM, parallelism
             )
         )
         .build();
 
+    query = prepareQuery(q);
+
     final List<ResultRow> results = runQuery();
 
     for (ResultRow result : results) {
       blackhole.consume(result);
     }
   }
 
+  private <T> Query<T> prepareQuery(Query<T> query)
+  {
+    return ResourceIdPopulatingQueryRunner.populateResourceId(query)
+                                          .withDataSource(ClientQuerySegmentWalker.generateSubqueryIds(
+                                              query.getDataSource(),
+                                              query.getId(),
+                                              query.getSqlQueryId(),
+                                              query.context().getString(QueryContexts.QUERY_RESOURCE_ID)
+                                          ));
+  }
+
   private <T> List<T> runQuery()
   {
     //noinspection unchecked
@@ -496,7 +523,7 @@ void addSegmentToServer(DruidServer server, DataSegment segment)
     {
       final ServerSelector selector = selectors.computeIfAbsent(
           segment.getId().toString(),
-          k -> new ServerSelector(segment, tierSelectorStrategy)
+          k -> new ServerSelector(segment, tierSelectorStrategy, filter)
       );
       selector.addServerAndUpdateSegment(servers.get(server), segment);
       timelines.computeIfAbsent(segment.getDataSource(), k -> new VersionedIntervalTimeline<>(Ordering.natural()))

diff --git a/docs/api-reference/service-status-api.md b/docs/api-reference/service-status-api.md
@@ -658,6 +658,138 @@ Host: http://COORDINATOR_IP:COORDINATOR_PORT
 
 </details>
 
+
+### Get Historical Cloning Status
+
+Retrieves the current status of Historical cloning from the Coordinator.
+
+#### URL
+
+`GET` `/druid/coordinator/v1/config/cloneStatus`
+
+#### Responses
+
+<Tabs>
+
+<TabItem value="56" label="200 SUCCESS">
+
+
+<br/>
+
+*Successfully retrieved cloning status*
+
+</TabItem>
+</Tabs>
+
+#### Sample request
+
+<Tabs>
+
+<TabItem value="58" label="cURL">
+
+
+```shell
+curl "http://COORDINATOR_IP:COORDINATOR_PORT/druid/coordinator/v1/config/cloneStatus"
+```
+
+</TabItem>
+<TabItem value="59" label="HTTP">
+
+
+```http
+GET /druid/coordinator/v1/config/cloneStatus HTTP/1.1
+Host: http://COORDINATOR_IP:COORDINATOR_PORT
+```
+
+</TabItem>
+</Tabs>
+
+#### Sample response
+
+<details>
+  <summary>View the response</summary>
+
+```json
+{
+  "cloneStatus": [
+    {
+      "sourceServer": "localhost:8089",
+      "targetServer": "localhost:8083",
+      "state": "IN_PROGRESS",
+      "segmentLoadsRemaining": 0,
+      "segmentDropsRemaining": 0,
+      "bytesToLoad": 0
+    }
+  ]
+}
+```
+
+</details>
+
+### Get Broker dynamic configuration view
+
+Retrieves the list of Brokers which have an up-to-date view of Coordinator dynamic configuration.
+
+#### URL
+
+`GET` `/druid/coordinator/v1/config/syncedBrokers`
+
+#### Responses
+
+<Tabs>
+
+<TabItem value="56" label="200 SUCCESS">
+
+
+<br/>
+
+*Successfully retrieved Broker Configuration view*
+
+</TabItem>
+</Tabs>
+
+#### Sample request
+
+<Tabs>
+
+<TabItem value="58" label="cURL">
+
+
+```shell
+curl "http://COORDINATOR_IP:COORDINATOR_PORT/druid/coordinator/v1/config/syncedBrokers"
+```
+
+</TabItem>
+<TabItem value="59" label="HTTP">
+
+
+```http
+GET /druid/coordinator/v1/config/syncedBrokers HTTP/1.1
+Host: http://COORDINATOR_IP:COORDINATOR_PORT
+```
+
+</TabItem>
+</Tabs>
+
+#### Sample response
+
+<details>
+  <summary>View the response</summary>
+
+```json
+{
+  "syncedBrokers": [
+    {
+      "host": "localhost",
+      "port": 8082,
+      "syncTimeInMs": 1745756337472
+    }
+  ]
+}
+```
+
+</details>
+
 ## Overlord
 
 ### Get Overlord leader address
@@ -1334,4 +1466,4 @@ Host: http://BROKER_IP:BROKER_PORT
 
 #### Sample response
 
-A successful response to this endpoint results in an empty response body.
+A successful response to this endpoint results in an empty response body.
diff --git a/docs/querying/query-context.md b/docs/querying/query-context.md
@@ -66,6 +66,7 @@ See [SQL query context](sql-query-context.md) for other query context parameters
 |`debug`| `false` | Flag indicating whether to enable debugging outputs for the query. When set to false, no additional logs will be produced (logs produced will be entirely dependent on your logging level). When set to true, the following addition logs will be produced:<br />- Log the stack trace of the exception (if any) produced by the query |
 |`setProcessingThreadNames`|`true`| Whether processing thread names will be set to `queryType_dataSource_intervals` while processing a query. This aids in interpreting thread dumps, and is on by default. Query overhead can be reduced slightly by setting this to `false`. This has a tiny effect in most scenarios, but can be meaningful in high-QPS, low-per-segment-processing-time scenarios. |
 |`sqlPlannerBloat`|`1000`|Calcite parameter which controls whether to merge two Project operators when inlining expressions causes complexity to increase. Implemented as a workaround to exception `There are not enough rules to produce a node with desired properties: convention=DRUID, sort=[]` thrown after rejecting the merge of two projects.|
+|`cloneQueryMode`|`excludeClones`| Indicates whether clone Historicals should be queried by brokers. Clone servers are created by the `cloneServers` Coordinator dynamic configuration. Possible values are `excludeClones`, `includeClones` and `clonesPreferred`. `excludeClones` means that clone Historicals are not queried by the broker. `clonesPreferred` indicates that when given a choice between the clone Historical and the original Historical which is being cloned, the broker chooses the clones; Historicals which are not involved in the cloning process will still be queried. `includeClones` means that broker queries any Historical without regarding clone status. This parameter only affects native queries; MSQ does not query Historicals directly, and Dart will not respect this context parameter.|
 
 ## Parameters by query type
 

diff --git a/...ection/src/test/java/org/apache/druid/query/materializedview/DatasourceOptimizerTest.java b/...ection/src/test/java/org/apache/druid/query/materializedview/DatasourceOptimizerTest.java
@@ -31,6 +31,7 @@
 import org.apache.druid.client.BatchServerInventoryView;
 import org.apache.druid.client.BrokerSegmentWatcherConfig;
 import org.apache.druid.client.BrokerServerView;
+import org.apache.druid.client.BrokerViewOfCoordinatorConfig;
 import org.apache.druid.client.DirectDruidClientFactory;
 import org.apache.druid.client.DruidServer;
 import org.apache.druid.client.selector.HighestPriorityTierSelectorStrategy;
@@ -58,6 +59,7 @@
 import org.apache.druid.segment.realtime.appenderator.SegmentSchemas;
 import org.apache.druid.server.coordination.DruidServerMetadata;
 import org.apache.druid.server.coordination.ServerType;
+import org.apache.druid.server.coordination.TestCoordinatorClient;
 import org.apache.druid.server.coordinator.simulate.TestDruidLeaderSelector;
 import org.apache.druid.server.initialization.ZkPathsConfig;
 import org.apache.druid.server.metrics.NoopServiceEmitter;
@@ -319,12 +321,15 @@ public CallbackAction segmentSchemasAnnounced(SegmentSchemas segmentSchemas)
         EasyMock.createMock(HttpClient.class)
     );
 
+    BrokerViewOfCoordinatorConfig filter = new BrokerViewOfCoordinatorConfig(new TestCoordinatorClient());
+    filter.start();
     brokerServerView = new BrokerServerView(
         druidClientFactory,
         baseView,
         new HighestPriorityTierSelectorStrategy(new RandomServerSelectorStrategy()),
         new NoopServiceEmitter(),
-        new BrokerSegmentWatcherConfig()
+        new BrokerSegmentWatcherConfig(),
+        filter
     );
     baseView.start();
   }

diff --git a/...rage-query/src/test/java/org/apache/druid/query/movingaverage/MovingAverageQueryTest.java b/...rage-query/src/test/java/org/apache/druid/query/movingaverage/MovingAverageQueryTest.java
@@ -316,7 +316,6 @@ public long getMaxQueuedBytes()
         return 0L;
       }
     };
-
     CachingClusteredClient baseClient = new CachingClusteredClient(
         conglomerate,
         new TimelineServerView()

diff --git a/...ge-query/src/main/java/org/apache/druid/msq/dart/controller/DartTableInputSpecSlicer.java b/...ge-query/src/main/java/org/apache/druid/msq/dart/controller/DartTableInputSpecSlicer.java
@@ -39,6 +39,7 @@
 import org.apache.druid.msq.input.table.RichSegmentDescriptor;
 import org.apache.druid.msq.input.table.SegmentsInputSlice;
 import org.apache.druid.msq.input.table.TableInputSpec;
+import org.apache.druid.query.CloneQueryMode;
 import org.apache.druid.query.TableDataSource;
 import org.apache.druid.query.filter.DimFilterUtils;
 import org.apache.druid.server.coordination.DruidServerMetadata;
@@ -162,7 +163,8 @@ public List<InputSlice> sliceDynamic(
    */
   int findWorkerForServerSelector(final ServerSelector serverSelector, final int maxNumSlices)
   {
-    final QueryableDruidServer server = serverSelector.pick(null);
+    // Currently, Dart does not support clone query modes, all servers can be queried.
+    final QueryableDruidServer server = serverSelector.pick(null, CloneQueryMode.INCLUDECLONES);
 
     if (server == null) {
       return UNKNOWN;
@@ -279,7 +281,8 @@ static boolean shouldIncludeSegment(final ServerSelector serverSelector)
     int numRealtimeServers = 0;
     int numOtherServers = 0;
 
-    for (final DruidServerMetadata server : serverSelector.getAllServers()) {
+    // Currently, Dart does not support clone query modes, all servers can be queried.
+    for (final DruidServerMetadata server : serverSelector.getAllServers(CloneQueryMode.INCLUDECLONES)) {
       if (SegmentSource.REALTIME.getUsedServerTypes().contains(server.getType())) {
         numRealtimeServers++;
       } else {

diff --git a/...uery/src/test/java/org/apache/druid/msq/dart/controller/DartTableInputSpecSlicerTest.java b/...uery/src/test/java/org/apache/druid/msq/dart/controller/DartTableInputSpecSlicerTest.java
@@ -28,6 +28,7 @@
 import org.apache.druid.client.QueryableDruidServer;
 import org.apache.druid.client.TimelineServerView;
 import org.apache.druid.client.selector.HighestPriorityTierSelectorStrategy;
+import org.apache.druid.client.selector.HistoricalFilter;
 import org.apache.druid.client.selector.RandomServerSelectorStrategy;
 import org.apache.druid.client.selector.ServerSelector;
 import org.apache.druid.data.input.StringTuple;
@@ -214,7 +215,8 @@ void setUp()
       final IntList segmentServers = entry.getValue();
       final ServerSelector serverSelector = new ServerSelector(
           dataSegment,
-          new HighestPriorityTierSelectorStrategy(new RandomServerSelectorStrategy())
+          new HighestPriorityTierSelectorStrategy(new RandomServerSelectorStrategy()),
+          HistoricalFilter.IDENTITY_FILTER
       );
       for (int serverNumber : segmentServers) {
         final DruidServerMetadata serverMetadata = SERVERS.get(serverNumber);