interval chunk query runner now processes individual chunk in a threadpool

docs/content/Broker-Config.md

@@ -41,7 +41,8 @@ Druid uses Jetty to serve HTTP requests.
 
 #### Processing
 
-The broker only uses processing configs for nested groupBy queries.
+The broker uses processing configs for nested groupBy queries. And, optionally, Long-interval queries (of any type) can be broken into shorter interval queries and processed in parallel inside this thread pool. For more details, see "chunkPeriod" in [Querying](Querying.html) doc.
+
 
 |Property|Description|Default|
 |--------|-----------|-------|
@@ -52,10 +53,6 @@ The broker only uses processing configs for nested groupBy queries.
 
 #### General Query Configuration
 
-|Property|Description|Default|
-|--------|-----------|-------|
-|`druid.query.chunkPeriod`|Long-interval queries (of any type) may be broken into shorter interval queries, reducing the impact on resources. Use ISO 8601 periods. For example, if this property is set to `P1M` (one month), then a query covering a year would be broken into 12 smaller queries. |0 (off)|
-
 ##### GroupBy Query Config
 
 |Property|Description|Default|

docs/content/Querying.md

@@ -145,6 +145,8 @@ Properties shared by all query types
 |populateCache | `true`              | Flag indicating whether to save the results of the query to the query cache. Primarily used for debugging. This may be overriden in the broker or historical node configuration |
 |bySegment     | `false`             | Return "by segment" results. Pimarily used for debugging, setting it to `true` returns results associated with the data segment they came from |
 |finalize      | `true`              | Flag indicating whether to "finalize" aggregation results. Primarily used for debugging. For instance, the `hyperUnique` aggregator will return the full HyperLogLog sketch instead of the estimated cardinality when this flag is set to `false` |
+|chunkPeriod   | `0` (off)           | At broker, Long-interval queries (of any type) may be broken into shorter interval queries, reducing the impact on resources. Use ISO 8601 periods. For example, if this property is set to `P1M` (one month), then a query covering a year would be broken into 12 smaller queries. All the query chunks will be processed asynchronously inside query processing executor service. Make sure "druid.processing.numThreads" is configured appropriately. |
+
 
 Query Cancellation
 ------------------

extensions/histogram/src/test/java/io/druid/query/aggregation/histogram/ApproximateHistogramGroupByQueryTest.java

@@ -84,7 +84,8 @@ public ByteBuffer get()
         engine,
         QueryRunnerTestHelper.NOOP_QUERYWATCHER,
         configSupplier,
-        new GroupByQueryQueryToolChest(configSupplier, mapper, engine, pool),
+        new GroupByQueryQueryToolChest(configSupplier, mapper, engine, pool,
+                QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()),
         pool
     );
 
@@ -105,7 +106,8 @@ public boolean isSingleThreaded()
         singleThreadEngine,
         QueryRunnerTestHelper.NOOP_QUERYWATCHER,
         singleThreadedConfigSupplier,
-        new GroupByQueryQueryToolChest(singleThreadedConfigSupplier, mapper, singleThreadEngine, pool),
+        new GroupByQueryQueryToolChest(singleThreadedConfigSupplier, mapper, singleThreadEngine, pool,
+                QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()),
         pool
     );
 

extensions/histogram/src/test/java/io/druid/query/aggregation/histogram/ApproximateHistogramTopNQueryTest.java

@@ -61,7 +61,7 @@ public static Collection<?> constructorFeeder() throws IOException
         QueryRunnerTestHelper.makeQueryRunners(
             new TopNQueryRunnerFactory(
                 TestQueryRunners.getPool(),
-                new TopNQueryQueryToolChest(new TopNQueryConfig()),
+                new TopNQueryQueryToolChest(new TopNQueryConfig(), QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()),
                 QueryRunnerTestHelper.NOOP_QUERYWATCHER
             )
         )
@@ -79,7 +79,7 @@ public ByteBuffer get()
                       }
                     }
                 ),
-                new TopNQueryQueryToolChest(new TopNQueryConfig()),
+                new TopNQueryQueryToolChest(new TopNQueryConfig(), QueryRunnerTestHelper.NoopIntervalChunkingQueryRunnerDecorator()),
                 QueryRunnerTestHelper.NOOP_QUERYWATCHER
             )
         )

processing/src/main/java/io/druid/query/AsyncQueryRunner.java

@@ -0,0 +1,83 @@
+/*
+* Licensed to Metamarkets Group Inc. (Metamarkets) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. Metamarkets 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 io.druid.query;
+
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import com.google.common.base.Supplier;
+import com.google.common.base.Throwables;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.metamx.common.guava.LazySequence;
+import com.metamx.common.guava.Sequence;
+
+public class AsyncQueryRunner<T> implements QueryRunner<T>
+{
+
+  private final QueryRunner<T> baseRunner;
+  private final ListeningExecutorService executor;
+  private final QueryWatcher queryWatcher;
+
+  public AsyncQueryRunner(QueryRunner<T> baseRunner, ExecutorService executor, QueryWatcher queryWatcher) {
+    this.baseRunner = baseRunner;
+    this.executor = MoreExecutors.listeningDecorator(executor);
+    this.queryWatcher = queryWatcher;
+  }
+
+  @Override
+  public Sequence<T> run(final Query<T> query, final Map<String, Object> responseContext)
+  {
+    final int priority = query.getContextPriority(0);
+    final ListenableFuture<Sequence<T>> future = executor.submit(new AbstractPrioritizedCallable<Sequence<T>>(priority)
+        {
+          @Override
+          public Sequence<T> call() throws Exception
+          {
+            //Note: this is assumed that baseRunner does most of the work eagerly on call to the
+            //run() method and resulting sequence accumulate/yield is fast.
+            return baseRunner.run(query, responseContext);
+          }
+        });
+    queryWatcher.registerQuery(query, future);
+
+    return new LazySequence<>(new Supplier<Sequence<T>>()
+    {
+      @Override
+      public Sequence<T> get()
+      {
+        try {
+          Number timeout = query.getContextValue(QueryContextKeys.TIMEOUT);
+          if (timeout == null) {
+            return future.get();
+          } else {
+            return future.get(timeout.longValue(), TimeUnit.MILLISECONDS);
+          }
+        } catch (ExecutionException | InterruptedException | TimeoutException ex) {
+          throw Throwables.propagate(ex);
+        }
+      }
+    });
+  }
+}

processing/src/main/java/io/druid/query/IntervalChunkingQueryRunner.java

@@ -22,6 +22,9 @@
 import com.metamx.common.guava.FunctionalIterable;
 import com.metamx.common.guava.Sequence;
 import com.metamx.common.guava.Sequences;
+import com.metamx.emitter.service.ServiceEmitter;
+import com.metamx.emitter.service.ServiceMetricEvent;
+
 import io.druid.granularity.PeriodGranularity;
 import io.druid.query.spec.MultipleIntervalSegmentSpec;
 import org.joda.time.Interval;
@@ -31,57 +34,90 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 
 /**
  */
 public class IntervalChunkingQueryRunner<T> implements QueryRunner<T>
 {
   private final QueryRunner<T> baseRunner;
-  private final Period period;
 
-  public IntervalChunkingQueryRunner(QueryRunner<T> baseRunner, Period period)
+  private final QueryToolChest<T, Query<T>> toolChest;
+  private final ExecutorService executor;
+  private final QueryWatcher queryWatcher;
+  private final ServiceEmitter emitter;
+
+  public IntervalChunkingQueryRunner(QueryRunner<T> baseRunner, QueryToolChest<T, Query<T>> toolChest,
+      ExecutorService executor, QueryWatcher queryWatcher, ServiceEmitter emitter)
   {
     this.baseRunner = baseRunner;
-    this.period = period;
+    this.toolChest = toolChest;
+    this.executor = executor;
+    this.queryWatcher = queryWatcher;
+    this.emitter = emitter;
   }
 
   @Override
   public Sequence<T> run(final Query<T> query, final Map<String, Object> responseContext)
   {
-    if (period.getMillis() == 0) {
+    final Period chunkPeriod = getChunkPeriod(query);
+    if (chunkPeriod.toStandardDuration().getMillis() == 0) {
       return baseRunner.run(query, responseContext);
     }
 
-    return Sequences.concat(
-        FunctionalIterable
-            .create(query.getIntervals())
-            .transformCat(
-                new Function<Interval, Iterable<Interval>>()
+    List<Interval> chunkIntervals = Lists.newArrayList(FunctionalIterable
+        .create(query.getIntervals())
+        .transformCat(
+            new Function<Interval, Iterable<Interval>>()
+            {
+              @Override
+              public Iterable<Interval> apply(Interval input)
+              {
+                return splitInterval(input, chunkPeriod);
+              }
+            }
+        ));
+
+    if(chunkIntervals.size() <= 1) {
+      return baseRunner.run(query, responseContext);
+    }
+
+    final QueryRunner<T> finalQueryRunner = new AsyncQueryRunner<T>(
+        //Note: it is assumed that toolChest.mergeResults(..) gives a query runner that is
+        //not lazy i.e. it does most of its work on call to run() method 
+        toolChest.mergeResults(
+            new MetricsEmittingQueryRunner<T>(
+                emitter,
+                new Function<Query<T>, ServiceMetricEvent.Builder>()
                 {
                   @Override
-                  public Iterable<Interval> apply(Interval input)
+                  public ServiceMetricEvent.Builder apply(Query<T> input)
                   {
-                    return splitInterval(input);
+                    return toolChest.makeMetricBuilder(input);
                   }
-                }
-            )
-            .transform(
-                new Function<Interval, Sequence<T>>()
-                {
-                  @Override
-                  public Sequence<T> apply(Interval singleInterval)
-                  {
-                    return baseRunner.run(
-                        query.withQuerySegmentSpec(new MultipleIntervalSegmentSpec(Arrays.asList(singleInterval))),
-                        responseContext
+                },
+                baseRunner
+            ).withWaitMeasuredFromNow()),
+        executor, queryWatcher);
+
+    return Sequences.concat(
+        Lists.newArrayList(FunctionalIterable.create(chunkIntervals).transform(
+            new Function<Interval, Sequence<T>>()
+            {
+              @Override
+              public Sequence<T> apply(Interval singleInterval)
+              {
+                return finalQueryRunner.run(
+                    query.withQuerySegmentSpec(new MultipleIntervalSegmentSpec(Arrays.asList(singleInterval))),
+                    responseContext
                     );
-                  }
-                }
-            )
-    );
+              }
+            }
+            ))
+        );
   }
 
-  private Iterable<Interval> splitInterval(Interval interval)
+  private Iterable<Interval> splitInterval(Interval interval, Period period)
   {
     if (interval.getEndMillis() == interval.getStartMillis()) {
       return Lists.newArrayList(interval);
@@ -106,4 +142,9 @@ private Iterable<Interval> splitInterval(Interval interval)
 
     return intervals;
   }
+
+  private Period getChunkPeriod(Query<T> query) {
+    String p = query.getContextValue(QueryContextKeys.CHUNK_PERIOD, "P0D");
+    return Period.parse(p);
+  }
 }

processing/src/main/java/io/druid/query/IntervalChunkingQueryRunnerDecorator.java

@@ -0,0 +1,49 @@
+/*
+* Licensed to Metamarkets Group Inc. (Metamarkets) under one
+* or more contributor license agreements. See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership. Metamarkets 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 io.druid.query;
+
+import io.druid.guice.annotations.Processing;
+
+import java.util.concurrent.ExecutorService;
+
+import com.google.inject.Inject;
+import com.metamx.emitter.service.ServiceEmitter;
+
+public class IntervalChunkingQueryRunnerDecorator
+{
+  private final ExecutorService executor;
+  private final QueryWatcher queryWatcher;
+  private final ServiceEmitter emitter;
+
+  @Inject
+  public IntervalChunkingQueryRunnerDecorator(@Processing ExecutorService executor, QueryWatcher queryWatcher,
+      ServiceEmitter emitter)
+  {
+    this.executor = executor;
+    this.queryWatcher = queryWatcher;
+    this.emitter = emitter;
+  }
+
+  public <T> QueryRunner<T> decorate(QueryRunner<T> delegate, QueryToolChest<T, ? extends Query<T>> toolChest)
+  {
+    return new IntervalChunkingQueryRunner<T>(delegate, (QueryToolChest<T, Query<T>>)toolChest,
+        executor, queryWatcher, emitter);
+  }
+}