You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by bs...@apache.org on 2019/01/11 18:42:24 UTC
[incubator-druid] branch master updated: Use Guava Compatible
immediate executor service (#6815)
This is an automated email from the ASF dual-hosted git repository.
bslim pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-druid.git
The following commit(s) were added to refs/heads/master by this push:
new 5d2947c Use Guava Compatible immediate executor service (#6815)
5d2947c is described below
commit 5d2947cd5229d683494b904a4146249f5dfac05f
Author: Charles Allen <ch...@allen-net.com>
AuthorDate: Fri Jan 11 10:42:19 2019 -0800
Use Guava Compatible immediate executor service (#6815)
* Use multi-guava version friendly direct executor implementation
* Don't use a singleton
* Fix strict compliation complaints
* Copy Guava's DirectExecutor
* Fix javadoc
* Imports are the devil
---
codestyle/druid-forbidden-apis.txt | 3 +
.../common/concurrent/DirectExecutorService.java | 170 ++
.../druid/java/util/common/concurrent/Execs.java | 8 +
.../org/apache/druid/concurrent/ExecsTest.java | 6 +
.../util/common/guava/WithEffectSequenceTest.java | 8 +-
.../query/lookup/KafkaLookupExtractorFactory.java | 2 +-
.../druid/indexing/kafka/KafkaIndexTaskTest.java | 2 +-
.../indexing/kinesis/KinesisIndexTaskTest.java | 2 +-
.../variance/VarianceGroupByQueryTest.java | 5 +-
.../apache/druid/indexer/IndexGeneratorJob.java | 37 +-
.../supervisor/SeekableStreamSupervisor.java | 2 +-
.../indexing/worker/WorkerCuratorCoordinator.java | 4 +-
.../druid/indexing/worker/WorkerTaskManager.java | 17 +-
.../AppenderatorDriverRealtimeIndexTaskTest.java | 2 +-
.../common/task/RealtimeIndexTaskTest.java | 2 +-
.../druid/indexing/overlord/TaskLifecycleTest.java | 4 +-
.../overlord/hrtr/HttpRemoteTaskRunnerTest.java | 70 +-
.../apache/druid/query/QueryRunnerTestHelper.java | 8 +-
.../apache/druid/query/SchemaEvolutionTest.java | 4 +-
.../query/aggregation/AggregationTestHelper.java | 4 +-
.../query/groupby/GroupByQueryMergeBufferTest.java | 4 +-
.../groupby/GroupByQueryRunnerFailureTest.java | 4 +-
.../query/groupby/GroupByQueryRunnerTest.java | 2897 ++++++++++++++++----
.../groupby/GroupByTimeseriesQueryRunnerTest.java | 5 +-
.../query/metadata/SegmentMetadataQueryTest.java | 20 +-
.../query/scan/MultiSegmentScanQueryTest.java | 16 +-
.../query/spec/SpecificSegmentQueryRunnerTest.java | 5 +-
.../client/cache/BackgroundCachePopulator.java | 3 +-
.../druid/segment/realtime/RealtimeManager.java | 6 +-
.../appenderator/SinkQuerySegmentWalker.java | 4 +-
.../appenderator/StreamAppenderatorDriver.java | 8 +-
.../java/org/apache/druid/server/QueryManager.java | 4 +-
.../druid/client/CachingClusteredClientTest.java | 97 +-
.../druid/client/HttpServerInventoryViewTest.java | 18 +-
.../client/BatchServerInventoryViewTest.java | 15 +-
.../CuratorDruidNodeAnnouncerAndDiscoveryTest.java | 5 +-
.../segment/realtime/RealtimeManagerTest.java | 8 +-
...CoordinatorBasedSegmentHandoffNotifierTest.java | 6 +-
.../plumber/RealtimePlumberSchoolTest.java | 10 +-
.../BatchDataSegmentAnnouncerTest.java | 5 +-
.../java/org/apache/druid/cli/DumpSegment.java | 4 +-
.../druid/sql/calcite/schema/DruidSchema.java | 4 +-
.../util/SpecificSegmentsQuerySegmentWalker.java | 4 +-
43 files changed, 2751 insertions(+), 761 deletions(-)
diff --git a/codestyle/druid-forbidden-apis.txt b/codestyle/druid-forbidden-apis.txt
index be5c40f..25735d0 100644
--- a/codestyle/druid-forbidden-apis.txt
+++ b/codestyle/druid-forbidden-apis.txt
@@ -20,6 +20,9 @@ com.google.common.collect.Sets#newHashSet() @ Create java.util.HashSet directly
com.google.common.collect.Sets#newLinkedHashSet() @ Create java.util.LinkedHashSet directly
com.google.common.collect.Sets#newTreeSet() @ Create java.util.TreeSet directly
com.google.common.collect.Sets#newTreeSet(java.util.Comparator) @ Create java.util.TreeSet directly
+com.google.common.util.concurrent.MoreExecutors#sameThreadExecutor() @ Use org.apache.druid.java.util.common.concurrent.Execs#directExecutor()
+com.google.common.util.concurrent.MoreExecutors#newDirectExecutorService() @ Use org.apache.druid.java.util.common.concurrent.Execs#directExecutor()
+com.google.common.util.concurrent.MoreExecutors#directExecutor() @ Use org.apache.druid.java.util.common.concurrent.Execs#directExecutor()
com.google.common.util.concurrent.Futures#transform(com.google.common.util.concurrent.ListenableFuture, com.google.common.util.concurrent.AsyncFunction) @ Use org.apache.druid.java.util.common.concurrent.ListenableFutures#transformAsync
java.io.File#toURL() @ Use java.io.File#toURI() and java.net.URI#toURL() instead
java.lang.String#matches(java.lang.String) @ Use startsWith(), endsWith(), contains(), or compile and cache a Pattern explicitly
diff --git a/core/src/main/java/org/apache/druid/java/util/common/concurrent/DirectExecutorService.java b/core/src/main/java/org/apache/druid/java/util/common/concurrent/DirectExecutorService.java
new file mode 100644
index 0000000..c692445
--- /dev/null
+++ b/core/src/main/java/org/apache/druid/java/util/common/concurrent/DirectExecutorService.java
@@ -0,0 +1,170 @@
+/*
+ * 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.java.util.common.concurrent;
+
+
+import com.google.common.util.concurrent.AbstractListeningExecutorService;
+
+import javax.annotation.concurrent.GuardedBy;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+// Copy of Guava's Apache 2.0 licensed https://github.com/google/guava/blob/a5cafa67da64a12444037bd4f4c30c39a0c184aa/guava/src/com/google/common/util/concurrent/MoreExecutors.java#L240-L339
+
+/**
+ * Creates an executor service that runs each task in the thread that invokes {@code
+ * execute/submit}, as in {@link java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy} This applies both to individually submitted
+ * tasks and to collections of tasks submitted via {@code invokeAll} or {@code invokeAny}. In the
+ * latter case, tasks will run serially on the calling thread. Tasks are run to completion before
+ * a {@code Future} is returned to the caller (unless the executor has been shutdown).
+ *
+ * <p>Although all tasks are immediately executed in the thread that submitted the task, this
+ * {@code ExecutorService} imposes a small locking overhead on each task submission in order to
+ * implement shutdown and termination behavior.
+ *
+ * <p>The implementation deviates from the {@code ExecutorService} specification with regards to
+ * the {@code shutdownNow} method. First, "best-effort" with regards to canceling running tasks is
+ * implemented as "no-effort". No interrupts or other attempts are made to stop threads executing
+ * tasks. Second, the returned list will always be empty, as any submitted task is considered to
+ * have started execution. This applies also to tasks given to {@code invokeAll} or {@code
+ * invokeAny} which are pending serial execution, even the subset of the tasks that have not yet
+ * started execution. It is unclear from the {@code ExecutorService} specification if these should
+ * be included, and it's much easier to implement the interpretation that they not be. Finally, a
+ * call to {@code shutdown} or {@code shutdownNow} may result in concurrent calls to {@code
+ * invokeAll/invokeAny} throwing RejectedExecutionException, although a subset of the tasks may
+ * already have been executed.
+ */
+public class DirectExecutorService extends AbstractListeningExecutorService
+{
+
+ /**
+ * Lock used whenever accessing the state variables (runningTasks, shutdown) of the executor
+ */
+ private final Object lock = new Object();
+
+ /*
+ * Conceptually, these two variables describe the executor being in
+ * one of three states:
+ * - Active: shutdown == false
+ * - Shutdown: runningTasks > 0 and shutdown == true
+ * - Terminated: runningTasks == 0 and shutdown == true
+ */
+ @GuardedBy("lock")
+ private int runningTasks = 0;
+
+ @GuardedBy("lock")
+ private boolean shutdown = false;
+
+ @Override
+ public void execute(Runnable command)
+ {
+ startTask();
+ try {
+ command.run();
+ }
+ finally {
+ endTask();
+ }
+ }
+
+ @Override
+ public boolean isShutdown()
+ {
+ synchronized (lock) {
+ return shutdown;
+ }
+ }
+
+ @Override
+ public void shutdown()
+ {
+ synchronized (lock) {
+ shutdown = true;
+ if (runningTasks == 0) {
+ lock.notifyAll();
+ }
+ }
+ }
+
+ // See newDirectExecutorService javadoc for unusual behavior of this method.
+ @Override
+ public List<Runnable> shutdownNow()
+ {
+ shutdown();
+ return Collections.emptyList();
+ }
+
+ @Override
+ public boolean isTerminated()
+ {
+ synchronized (lock) {
+ return shutdown && runningTasks == 0;
+ }
+ }
+
+ @Override
+ public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException
+ {
+ long nanos = unit.toNanos(timeout);
+ synchronized (lock) {
+ while (true) {
+ if (shutdown && runningTasks == 0) {
+ return true;
+ } else if (nanos <= 0) {
+ return false;
+ } else {
+ long now = System.nanoTime();
+ TimeUnit.NANOSECONDS.timedWait(lock, nanos);
+ nanos -= System.nanoTime() - now; // subtract the actual time we waited
+ }
+ }
+ }
+ }
+
+ /**
+ * Checks if the executor has been shut down and increments the running task count.
+ *
+ * @throws RejectedExecutionException if the executor has been previously shutdown
+ */
+ private void startTask()
+ {
+ synchronized (lock) {
+ if (shutdown) {
+ throw new RejectedExecutionException("Executor already shutdown");
+ }
+ runningTasks++;
+ }
+ }
+
+ /**
+ * Decrements the running task count.
+ */
+ private void endTask()
+ {
+ synchronized (lock) {
+ int numRunning = --runningTasks;
+ if (numRunning == 0) {
+ lock.notifyAll();
+ }
+ }
+ }
+}
diff --git a/core/src/main/java/org/apache/druid/java/util/common/concurrent/Execs.java b/core/src/main/java/org/apache/druid/java/util/common/concurrent/Execs.java
index 58cffb3..a310e56 100644
--- a/core/src/main/java/org/apache/druid/java/util/common/concurrent/Execs.java
+++ b/core/src/main/java/org/apache/druid/java/util/common/concurrent/Execs.java
@@ -21,6 +21,7 @@ package org.apache.druid.java.util.common.concurrent;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
+import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import javax.annotation.Nullable;
@@ -38,9 +39,11 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
+ *
*/
public class Execs
{
+
/**
* Returns an ExecutorService which is terminated and shutdown from the beginning and not able to accept any tasks.
*/
@@ -152,4 +155,9 @@ public class Execs
}
);
}
+
+ public static ListeningExecutorService directExecutor()
+ {
+ return new DirectExecutorService();
+ }
}
diff --git a/core/src/test/java/org/apache/druid/concurrent/ExecsTest.java b/core/src/test/java/org/apache/druid/concurrent/ExecsTest.java
index cedd99d..8634411 100644
--- a/core/src/test/java/org/apache/druid/concurrent/ExecsTest.java
+++ b/core/src/test/java/org/apache/druid/concurrent/ExecsTest.java
@@ -118,4 +118,10 @@ public class ExecsTest
blockingExecutor.shutdown();
producer.shutdown();
}
+
+ @Test
+ public void testDirectExecutorFactory()
+ {
+ Assert.assertNotNull(Execs.directExecutor());
+ }
}
diff --git a/core/src/test/java/org/apache/druid/java/util/common/guava/WithEffectSequenceTest.java b/core/src/test/java/org/apache/druid/java/util/common/guava/WithEffectSequenceTest.java
index d90e6a1..72182f5 100644
--- a/core/src/test/java/org/apache/druid/java/util/common/guava/WithEffectSequenceTest.java
+++ b/core/src/test/java/org/apache/druid/java/util/common/guava/WithEffectSequenceTest.java
@@ -19,7 +19,7 @@
package org.apache.druid.java.util.common.guava;
-import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.junit.Assert;
import org.junit.Test;
@@ -40,11 +40,11 @@ public class WithEffectSequenceTest
.simple(Arrays.asList(1, 2, 3))
.withEffect(
() -> effect1.set(counter.incrementAndGet()),
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
)
.withEffect(
() -> effect2.set(counter.incrementAndGet()),
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
// Run sequence via accumulate
sequence.toList();
@@ -70,7 +70,7 @@ public class WithEffectSequenceTest
});
final AtomicBoolean effectExecuted = new AtomicBoolean();
Sequence<Integer> seqWithEffect =
- throwingSeq.withEffect(() -> effectExecuted.set(true), MoreExecutors.sameThreadExecutor());
+ throwingSeq.withEffect(() -> effectExecuted.set(true), Execs.directExecutor());
try {
seqWithEffect.toList();
Assert.fail("expected RuntimeException");
diff --git a/extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java b/extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java
index 4fb1a24..e65c7a3 100644
--- a/extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java
+++ b/extensions-core/kafka-extraction-namespace/src/main/java/org/apache/druid/query/lookup/KafkaLookupExtractorFactory.java
@@ -261,7 +261,7 @@ public class KafkaLookupExtractorFactory implements LookupExtractorFactory
}
}
},
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
this.future = future;
final Stopwatch stopwatch = Stopwatch.createStarted();
diff --git a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
index f448777..0b6e7c9 100644
--- a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
+++ b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
@@ -2452,7 +2452,7 @@ public class KafkaIndexTaskTest
EasyMock.createNiceMock(DataSegmentServerAnnouncer.class),
handoffNotifierFactory,
this::makeTimeseriesOnlyConglomerate,
- MoreExecutors.sameThreadExecutor(), // queryExecutorService
+ Execs.directExecutor(), // queryExecutorService
EasyMock.createMock(MonitorScheduler.class),
new SegmentLoaderFactory(
new SegmentLoaderLocalCacheManager(null, segmentLoaderConfig, testUtils.getTestObjectMapper())
diff --git a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java
index cef9795..31bedd9 100644
--- a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java
+++ b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java
@@ -2765,7 +2765,7 @@ public class KinesisIndexTaskTest extends EasyMockSupport
EasyMock.createNiceMock(DataSegmentServerAnnouncer.class),
handoffNotifierFactory,
this::makeTimeseriesOnlyConglomerate,
- MoreExecutors.sameThreadExecutor(), // queryExecutorService
+ Execs.directExecutor(), // queryExecutorService
EasyMock.createMock(MonitorScheduler.class),
new SegmentLoaderFactory(
new SegmentLoaderLocalCacheManager(null, segmentLoaderConfig, testUtils.getTestObjectMapper())
diff --git a/extensions-core/stats/src/test/java/org/apache/druid/query/aggregation/variance/VarianceGroupByQueryTest.java b/extensions-core/stats/src/test/java/org/apache/druid/query/aggregation/variance/VarianceGroupByQueryTest.java
index bd19c74..1b0ef25 100644
--- a/extensions-core/stats/src/test/java/org/apache/druid/query/aggregation/variance/VarianceGroupByQueryTest.java
+++ b/extensions-core/stats/src/test/java/org/apache/druid/query/aggregation/variance/VarianceGroupByQueryTest.java
@@ -20,8 +20,8 @@
package org.apache.druid.query.aggregation.variance;
import com.google.common.collect.ImmutableList;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.data.input.Row;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.PeriodGranularity;
import org.apache.druid.query.QueryRunner;
import org.apache.druid.query.QueryRunnerTestHelper;
@@ -47,6 +47,7 @@ import java.util.Collections;
import java.util.List;
/**
+ *
*/
@RunWith(Parameterized.class)
public class VarianceGroupByQueryTest
@@ -72,7 +73,7 @@ public class VarianceGroupByQueryTest
this.testName = testName;
this.config = config;
this.factory = factory;
- this.runner = factory.mergeRunners(MoreExecutors.sameThreadExecutor(), ImmutableList.of(runner));
+ this.runner = factory.mergeRunners(Execs.directExecutor(), ImmutableList.of(runner));
}
@Test
diff --git a/indexing-hadoop/src/main/java/org/apache/druid/indexer/IndexGeneratorJob.java b/indexing-hadoop/src/main/java/org/apache/druid/indexer/IndexGeneratorJob.java
index 9448206..02c5c50 100644
--- a/indexing-hadoop/src/main/java/org/apache/druid/indexer/IndexGeneratorJob.java
+++ b/indexing-hadoop/src/main/java/org/apache/druid/indexer/IndexGeneratorJob.java
@@ -97,6 +97,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/**
+ *
*/
public class IndexGeneratorJob implements Jobby
{
@@ -240,7 +241,8 @@ public class IndexGeneratorJob implements Jobby
Map<String, Object> metrics = TaskMetricsUtils.makeIngestionRowMetrics(
jobCounters.findCounter(HadoopDruidIndexerConfig.IndexJobCounters.ROWS_PROCESSED_COUNTER).getValue(),
- jobCounters.findCounter(HadoopDruidIndexerConfig.IndexJobCounters.ROWS_PROCESSED_WITH_ERRORS_COUNTER).getValue(),
+ jobCounters.findCounter(HadoopDruidIndexerConfig.IndexJobCounters.ROWS_PROCESSED_WITH_ERRORS_COUNTER)
+ .getValue(),
jobCounters.findCounter(HadoopDruidIndexerConfig.IndexJobCounters.ROWS_UNPARSEABLE_COUNTER).getValue(),
jobCounters.findCounter(HadoopDruidIndexerConfig.IndexJobCounters.ROWS_THROWN_AWAY_COUNTER).getValue()
);
@@ -345,7 +347,9 @@ public class IndexGeneratorJob implements Jobby
throw new ISE("WTF?! No bucket found for row: %s", inputRow);
}
- final long truncatedTimestamp = granularitySpec.getQueryGranularity().bucketStart(inputRow.getTimestamp()).getMillis();
+ final long truncatedTimestamp = granularitySpec.getQueryGranularity()
+ .bucketStart(inputRow.getTimestamp())
+ .getMillis();
final byte[] hashedDimensions = hashFunction.hashBytes(
HadoopDruidIndexerConfig.JSON_MAPPER.writeValueAsBytes(
Rows.toGroupKey(
@@ -359,17 +363,17 @@ public class IndexGeneratorJob implements Jobby
// and they contain the columns as they show up in the segment after ingestion, not what you would see in raw
// data
InputRowSerde.SerializeResult serializeResult = inputRow instanceof SegmentInputRow ?
- InputRowSerde.toBytes(
- typeHelperMap,
- inputRow,
- aggsForSerializingSegmentInputRow
- )
- :
- InputRowSerde.toBytes(
- typeHelperMap,
- inputRow,
- aggregators
- );
+ InputRowSerde.toBytes(
+ typeHelperMap,
+ inputRow,
+ aggsForSerializingSegmentInputRow
+ )
+ :
+ InputRowSerde.toBytes(
+ typeHelperMap,
+ inputRow,
+ aggregators
+ );
context.write(
new SortableBytes(
@@ -678,7 +682,7 @@ public class IndexGeneratorJob implements Jobby
);
persistExecutor = MoreExecutors.listeningDecorator(executorService);
} else {
- persistExecutor = MoreExecutors.sameThreadExecutor();
+ persistExecutor = Execs.directExecutor();
}
for (final BytesWritable bw : values) {
@@ -786,7 +790,10 @@ public class IndexGeneratorJob implements Jobby
// ShardSpec to be published.
final ShardSpec shardSpecForPublishing;
if (config.isForceExtendableShardSpecs()) {
- shardSpecForPublishing = new NumberedShardSpec(shardSpecForPartitioning.getPartitionNum(), config.getShardSpecCount(bucket));
+ shardSpecForPublishing = new NumberedShardSpec(
+ shardSpecForPartitioning.getPartitionNum(),
+ config.getShardSpecCount(bucket)
+ );
} else {
shardSpecForPublishing = shardSpecForPartitioning;
}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
index ddd854d..25250ac 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
@@ -1051,7 +1051,7 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
{
notices.add(new RunNotice());
}
- }, MoreExecutors.sameThreadExecutor()
+ }, Execs.directExecutor()
);
listenerRegistered = true;
}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerCuratorCoordinator.java b/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerCuratorCoordinator.java
index 53d1495..7e7c098 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerCuratorCoordinator.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerCuratorCoordinator.java
@@ -22,7 +22,6 @@ package org.apache.druid.indexing.worker;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Joiner;
import com.google.common.collect.ImmutableMap;
-import com.google.common.util.concurrent.MoreExecutors;
import com.google.inject.Inject;
import org.apache.curator.framework.CuratorFramework;
import org.apache.druid.curator.CuratorUtils;
@@ -30,6 +29,7 @@ import org.apache.druid.curator.announcement.Announcer;
import org.apache.druid.indexing.overlord.config.RemoteTaskRunnerConfig;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.lifecycle.LifecycleStart;
import org.apache.druid.java.util.common.lifecycle.LifecycleStop;
import org.apache.druid.java.util.common.logger.Logger;
@@ -77,7 +77,7 @@ public class WorkerCuratorCoordinator
this.curatorFramework = curatorFramework;
this.worker = worker;
- this.announcer = new Announcer(curatorFramework, MoreExecutors.sameThreadExecutor());
+ this.announcer = new Announcer(curatorFramework, Execs.directExecutor());
this.baseAnnouncementsPath = getPath(Arrays.asList(indexerZkConfig.getAnnouncementsPath(), worker.getHost()));
this.baseTaskPath = getPath(Arrays.asList(indexerZkConfig.getTasksPath(), worker.getHost()));
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java b/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java
index 561a6e1..0ec7b41 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/worker/WorkerTaskManager.java
@@ -28,7 +28,6 @@ import com.google.common.collect.Lists;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.inject.Inject;
import org.apache.druid.client.indexing.IndexingService;
@@ -69,7 +68,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
/**
* This class manages the list of tasks assigned to this worker.
- *
+ * <p>
* It persists the list of assigned and completed tasks on disk. assigned task from disk is deleted as soon as it
* starts running and completed task on disk is deleted based on a periodic schedule where overlord is asked for
* active tasks to see which completed tasks are safe to delete.
@@ -226,7 +225,7 @@ public abstract class WorkerTaskManager
// do nothing
}
},
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
}
@@ -456,9 +455,12 @@ public abstract class WorkerTaskManager
);
if (fullResponseHolder.getStatus().getCode() == 200) {
String responseContent = fullResponseHolder.getContent();
- taskStatusesFromOverlord = jsonMapper.readValue(responseContent, new TypeReference<Map<String, TaskStatus>>()
- {
- });
+ taskStatusesFromOverlord = jsonMapper.readValue(
+ responseContent,
+ new TypeReference<Map<String, TaskStatus>>()
+ {
+ }
+ );
log.debug("Received completed task status response [%s].", responseContent);
} else if (fullResponseHolder.getStatus().getCode() == 404) {
// NOTE: this is to support backward compatibility, when overlord doesn't have "activeTasks" endpoint.
@@ -516,7 +518,7 @@ public abstract class WorkerTaskManager
TimeUnit.MINUTES
);
}
-
+
public void workerEnabled()
{
Preconditions.checkState(lifecycleLock.awaitStarted(1, TimeUnit.SECONDS), "not started");
@@ -717,5 +719,6 @@ public abstract class WorkerTaskManager
//in Overlord as well as MiddleManagers then WorkerTaskMonitor should be deleted, this class should no longer be abstract
//and the methods below should be removed.
protected abstract void taskStarted(String taskId);
+
protected abstract void taskAnnouncementChanged(TaskAnnouncement announcement);
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTaskTest.java
index 5c28242..59899fe 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTaskTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/AppenderatorDriverRealtimeIndexTaskTest.java
@@ -1607,7 +1607,7 @@ public class AppenderatorDriverRealtimeIndexTaskTest
EasyMock.createNiceMock(DataSegmentServerAnnouncer.class),
handoffNotifierFactory,
() -> conglomerate,
- MoreExecutors.sameThreadExecutor(), // queryExecutorService
+ Execs.directExecutor(), // queryExecutorService
EasyMock.createMock(MonitorScheduler.class),
new SegmentLoaderFactory(
new SegmentLoaderLocalCacheManager(null, segmentLoaderConfig, testUtils.getTestObjectMapper())
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/RealtimeIndexTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/RealtimeIndexTaskTest.java
index 73b5a2c..5317b66 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/RealtimeIndexTaskTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/RealtimeIndexTaskTest.java
@@ -1076,7 +1076,7 @@ public class RealtimeIndexTaskTest
EasyMock.createNiceMock(DataSegmentServerAnnouncer.class),
handoffNotifierFactory,
() -> conglomerate,
- MoreExecutors.sameThreadExecutor(), // queryExecutorService
+ Execs.directExecutor(), // queryExecutorService
EasyMock.createMock(MonitorScheduler.class),
new SegmentLoaderFactory(
new SegmentLoaderLocalCacheManager(null, segmentLoaderConfig, testUtils.getTestObjectMapper())
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
index 66f0ec9..31cbf2c 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/TaskLifecycleTest.java
@@ -33,7 +33,6 @@ import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Ordering;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.client.cache.CachePopulatorStats;
import org.apache.druid.client.cache.MapCache;
import org.apache.druid.data.input.Firehose;
@@ -81,6 +80,7 @@ import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.RE;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.guava.Comparators;
import org.apache.druid.java.util.emitter.EmittingLogger;
@@ -611,7 +611,7 @@ public class TaskLifecycleTest
EasyMock.createNiceMock(DataSegmentServerAnnouncer.class),
handoffNotifierFactory,
() -> queryRunnerFactoryConglomerate, // query runner factory conglomerate corporation unionized collective
- MoreExecutors.sameThreadExecutor(), // query executor service
+ Execs.directExecutor(), // query executor service
monitorScheduler, // monitor scheduler
new SegmentLoaderFactory(
new SegmentLoaderLocalCacheManager(null, segmentLoaderConfig, new DefaultObjectMapper())
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunnerTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunnerTest.java
index 25e172a..814a20f 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunnerTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/hrtr/HttpRemoteTaskRunnerTest.java
@@ -27,7 +27,6 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.curator.framework.CuratorFramework;
import org.apache.druid.common.guava.DSuppliers;
import org.apache.druid.discovery.DiscoveryDruidNode;
@@ -49,6 +48,7 @@ import org.apache.druid.indexing.overlord.setup.DefaultWorkerBehaviorConfig;
import org.apache.druid.indexing.worker.TaskAnnouncement;
import org.apache.druid.indexing.worker.Worker;
import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.http.client.HttpClient;
import org.apache.druid.segment.TestHelper;
import org.apache.druid.server.DruidNode;
@@ -73,6 +73,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
/**
+ *
*/
public class HttpRemoteTaskRunnerTest
{
@@ -91,7 +92,8 @@ public class HttpRemoteTaskRunnerTest
HttpRemoteTaskRunner taskRunner = new HttpRemoteTaskRunner(
TestHelper.makeJsonMapper(),
- new HttpRemoteTaskRunnerConfig() {
+ new HttpRemoteTaskRunnerConfig()
+ {
@Override
public int getPendingTasksRunnerNumThreads()
{
@@ -105,7 +107,8 @@ public class HttpRemoteTaskRunnerTest
EasyMock.createNiceMock(TaskStorage.class),
EasyMock.createNiceMock(CuratorFramework.class),
new IndexerZkConfig(new ZkPathsConfig(), null, null, null, null)
- ) {
+ )
+ {
@Override
protected WorkerHolder createWorkerHolder(
ObjectMapper smileMapper,
@@ -184,7 +187,8 @@ public class HttpRemoteTaskRunnerTest
HttpRemoteTaskRunner taskRunner = new HttpRemoteTaskRunner(
TestHelper.makeJsonMapper(),
- new HttpRemoteTaskRunnerConfig() {
+ new HttpRemoteTaskRunnerConfig()
+ {
@Override
public int getPendingTasksRunnerNumThreads()
{
@@ -198,7 +202,8 @@ public class HttpRemoteTaskRunnerTest
EasyMock.createNiceMock(TaskStorage.class),
EasyMock.createNiceMock(CuratorFramework.class),
new IndexerZkConfig(new ZkPathsConfig(), null, null, null, null)
- ) {
+ )
+ {
@Override
protected WorkerHolder createWorkerHolder(
ObjectMapper smileMapper,
@@ -284,7 +289,8 @@ public class HttpRemoteTaskRunnerTest
HttpRemoteTaskRunner taskRunner = new HttpRemoteTaskRunner(
TestHelper.makeJsonMapper(),
- new HttpRemoteTaskRunnerConfig() {
+ new HttpRemoteTaskRunnerConfig()
+ {
@Override
public int getPendingTasksRunnerNumThreads()
{
@@ -298,7 +304,8 @@ public class HttpRemoteTaskRunnerTest
taskStorageMock,
EasyMock.createNiceMock(CuratorFramework.class),
new IndexerZkConfig(new ZkPathsConfig(), null, null, null, null)
- ) {
+ )
+ {
@Override
protected WorkerHolder createWorkerHolder(
ObjectMapper smileMapper,
@@ -316,7 +323,8 @@ public class HttpRemoteTaskRunnerTest
config,
workersSyncExec,
listener,
- worker);
+ worker
+ );
} else {
throw new ISE("No WorkerHolder for [%s].", worker.getHost());
}
@@ -420,7 +428,8 @@ public class HttpRemoteTaskRunnerTest
HttpRemoteTaskRunner taskRunner = new HttpRemoteTaskRunner(
TestHelper.makeJsonMapper(),
- new HttpRemoteTaskRunnerConfig() {
+ new HttpRemoteTaskRunnerConfig()
+ {
@Override
public int getPendingTasksRunnerNumThreads()
{
@@ -434,7 +443,8 @@ public class HttpRemoteTaskRunnerTest
EasyMock.createNiceMock(TaskStorage.class),
EasyMock.createNiceMock(CuratorFramework.class),
new IndexerZkConfig(new ZkPathsConfig(), null, null, null, null)
- ) {
+ )
+ {
@Override
protected WorkerHolder createWorkerHolder(
ObjectMapper smileMapper,
@@ -452,7 +462,8 @@ public class HttpRemoteTaskRunnerTest
config,
workersSyncExec,
listener,
- worker);
+ worker
+ );
} else {
throw new ISE("No WorkerHolder for [%s].", worker.getHost());
}
@@ -593,7 +604,8 @@ public class HttpRemoteTaskRunnerTest
HttpRemoteTaskRunner taskRunner = new HttpRemoteTaskRunner(
TestHelper.makeJsonMapper(),
- new HttpRemoteTaskRunnerConfig() {
+ new HttpRemoteTaskRunnerConfig()
+ {
@Override
public Period getTaskCleanupTimeout()
{
@@ -607,7 +619,8 @@ public class HttpRemoteTaskRunnerTest
EasyMock.createNiceMock(TaskStorage.class),
EasyMock.createNiceMock(CuratorFramework.class),
new IndexerZkConfig(new ZkPathsConfig(), null, null, null, null)
- ) {
+ )
+ {
@Override
protected WorkerHolder createWorkerHolder(
ObjectMapper smileMapper,
@@ -625,7 +638,8 @@ public class HttpRemoteTaskRunnerTest
config,
workersSyncExec,
listener,
- worker);
+ worker
+ );
} else {
throw new ISE("No WorkerHolder for [%s].", worker.getHost());
}
@@ -801,7 +815,8 @@ public class HttpRemoteTaskRunnerTest
config,
workersSyncExec,
listener,
- worker);
+ worker
+ );
} else {
throw new ISE("No WorkerHolder for [%s].", worker.getHost());
}
@@ -914,9 +929,10 @@ public class HttpRemoteTaskRunnerTest
Assert.assertEquals(task1.getId(), Iterables.getOnlyElement(taskRunner.getRunningTasks()).getTaskId());
Assert.assertEquals(task2.getId(), Iterables.getOnlyElement(taskRunner.getPendingTasks()).getTaskId());
- Assert.assertEquals("host3:8080",
- Iterables.getOnlyElement(taskRunner.markWorkersLazy(Predicates.alwaysTrue(), Integer.MAX_VALUE))
- .getHost()
+ Assert.assertEquals(
+ "host3:8080",
+ Iterables.getOnlyElement(taskRunner.markWorkersLazy(Predicates.alwaysTrue(), Integer.MAX_VALUE))
+ .getHost()
);
}
@@ -969,7 +985,9 @@ public class HttpRemoteTaskRunnerTest
// Another "rogue-worker" reports running it, and gets asked to shutdown the task
WorkerHolder rogueWorkerHolder = EasyMock.createMock(WorkerHolder.class);
- EasyMock.expect(rogueWorkerHolder.getWorker()).andReturn(new Worker("http", "rogue-worker", "127.0.0.1", 5, "v1")).anyTimes();
+ EasyMock.expect(rogueWorkerHolder.getWorker())
+ .andReturn(new Worker("http", "rogue-worker", "127.0.0.1", 5, "v1"))
+ .anyTimes();
rogueWorkerHolder.shutdownTask(task.getId());
EasyMock.replay(rogueWorkerHolder);
taskRunner.taskAddedOrUpdated(TaskAnnouncement.create(
@@ -982,7 +1000,9 @@ public class HttpRemoteTaskRunnerTest
// "rogue-worker" reports FAILURE for the task, ignored
rogueWorkerHolder = EasyMock.createMock(WorkerHolder.class);
- EasyMock.expect(rogueWorkerHolder.getWorker()).andReturn(new Worker("http", "rogue-worker", "127.0.0.1", 5, "v1")).anyTimes();
+ EasyMock.expect(rogueWorkerHolder.getWorker())
+ .andReturn(new Worker("http", "rogue-worker", "127.0.0.1", 5, "v1"))
+ .anyTimes();
EasyMock.replay(rogueWorkerHolder);
taskRunner.taskAddedOrUpdated(TaskAnnouncement.create(
task,
@@ -1003,7 +1023,9 @@ public class HttpRemoteTaskRunnerTest
// "rogue-worker" reports running it, and gets asked to shutdown the task
rogueWorkerHolder = EasyMock.createMock(WorkerHolder.class);
- EasyMock.expect(rogueWorkerHolder.getWorker()).andReturn(new Worker("http", "rogue-worker", "127.0.0.1", 5, "v1")).anyTimes();
+ EasyMock.expect(rogueWorkerHolder.getWorker())
+ .andReturn(new Worker("http", "rogue-worker", "127.0.0.1", 5, "v1"))
+ .anyTimes();
rogueWorkerHolder.shutdownTask(task.getId());
EasyMock.replay(rogueWorkerHolder);
taskRunner.taskAddedOrUpdated(TaskAnnouncement.create(
@@ -1016,7 +1038,9 @@ public class HttpRemoteTaskRunnerTest
// "rogue-worker" reports FAILURE for the tasks, ignored
rogueWorkerHolder = EasyMock.createMock(WorkerHolder.class);
- EasyMock.expect(rogueWorkerHolder.getWorker()).andReturn(new Worker("http", "rogue-worker", "127.0.0.1", 5, "v1")).anyTimes();
+ EasyMock.expect(rogueWorkerHolder.getWorker())
+ .andReturn(new Worker("http", "rogue-worker", "127.0.0.1", 5, "v1"))
+ .anyTimes();
EasyMock.replay(rogueWorkerHolder);
taskRunner.taskAddedOrUpdated(TaskAnnouncement.create(
task,
@@ -1228,7 +1252,7 @@ public class HttpRemoteTaskRunnerTest
listenerNotificationsAccumulator.add(ImmutableList.of(taskId, status));
}
},
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
}
diff --git a/processing/src/test/java/org/apache/druid/query/QueryRunnerTestHelper.java b/processing/src/test/java/org/apache/druid/query/QueryRunnerTestHelper.java
index cfc18e8..68a91d6 100644
--- a/processing/src/test/java/org/apache/druid/query/QueryRunnerTestHelper.java
+++ b/processing/src/test/java/org/apache/druid/query/QueryRunnerTestHelper.java
@@ -24,9 +24,9 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.granularity.Granularity;
import org.apache.druid.java.util.common.guava.MergeSequence;
@@ -76,11 +76,13 @@ import java.util.List;
import java.util.Map;
/**
+ *
*/
public class QueryRunnerTestHelper
{
- public static final QueryWatcher NOOP_QUERYWATCHER = (query, future) -> {};
+ public static final QueryWatcher NOOP_QUERYWATCHER = (query, future) -> {
+ };
public static final String segmentId = "testSegment";
public static final String dataSource = "testing";
@@ -506,7 +508,7 @@ public class QueryRunnerTestHelper
public static IntervalChunkingQueryRunnerDecorator sameThreadIntervalChunkingQueryRunnerDecorator()
{
return new IntervalChunkingQueryRunnerDecorator(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
QueryRunnerTestHelper.NOOP_QUERYWATCHER,
new ServiceEmitter("dummy", "dummy", new NoopEmitter())
);
diff --git a/processing/src/test/java/org/apache/druid/query/SchemaEvolutionTest.java b/processing/src/test/java/org/apache/druid/query/SchemaEvolutionTest.java
index 2a5fc3d..7f8ea4c 100644
--- a/processing/src/test/java/org/apache/druid/query/SchemaEvolutionTest.java
+++ b/processing/src/test/java/org/apache/druid/query/SchemaEvolutionTest.java
@@ -23,7 +23,6 @@ import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.io.Closeables;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.common.config.NullHandling;
import org.apache.druid.data.input.InputRow;
import org.apache.druid.data.input.impl.DimensionsSpec;
@@ -32,6 +31,7 @@ import org.apache.druid.data.input.impl.TimeAndDimsParseSpec;
import org.apache.druid.data.input.impl.TimestampSpec;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.guava.FunctionalIterable;
import org.apache.druid.java.util.common.guava.Sequence;
import org.apache.druid.query.aggregation.CountAggregatorFactory;
@@ -106,7 +106,7 @@ public class SchemaEvolutionTest
final Sequence<T> results = new FinalizeResultsQueryRunner<>(
factory.getToolchest().mergeResults(
factory.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
FunctionalIterable
.create(indexes)
.transform(
diff --git a/processing/src/test/java/org/apache/druid/query/aggregation/AggregationTestHelper.java b/processing/src/test/java/org/apache/druid/query/aggregation/AggregationTestHelper.java
index 27137b9..bd2c13f 100644
--- a/processing/src/test/java/org/apache/druid/query/aggregation/AggregationTestHelper.java
+++ b/processing/src/test/java/org/apache/druid/query/aggregation/AggregationTestHelper.java
@@ -32,7 +32,6 @@ import com.google.common.base.Suppliers;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.io.Closeables;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.io.IOUtils;
import org.apache.commons.io.LineIterator;
import org.apache.druid.collections.CloseableStupidPool;
@@ -42,6 +41,7 @@ import org.apache.druid.data.input.impl.InputRowParser;
import org.apache.druid.data.input.impl.StringInputRowParser;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularity;
import org.apache.druid.java.util.common.guava.CloseQuietly;
import org.apache.druid.java.util.common.guava.Sequence;
@@ -607,7 +607,7 @@ public class AggregationTestHelper implements Closeable
toolChest.mergeResults(
toolChest.preMergeQueryDecoration(
factory.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Lists.transform(
segments,
new Function<Segment, QueryRunner>()
diff --git a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryMergeBufferTest.java b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryMergeBufferTest.java
index e65aaa2..19f3a58 100644
--- a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryMergeBufferTest.java
+++ b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryMergeBufferTest.java
@@ -25,11 +25,11 @@ import com.google.common.base.Suppliers;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.collections.CloseableDefaultBlockingPool;
import org.apache.druid.collections.CloseableStupidPool;
import org.apache.druid.collections.ReferenceCountingResourceHolder;
import org.apache.druid.data.input.Row;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.query.DruidProcessingConfig;
import org.apache.druid.query.QueryContexts;
@@ -222,7 +222,7 @@ public class GroupByQueryMergeBufferTest
public GroupByQueryMergeBufferTest(QueryRunner<Row> runner)
{
- this.runner = factory.mergeRunners(MoreExecutors.sameThreadExecutor(), ImmutableList.of(runner));
+ this.runner = factory.mergeRunners(Execs.directExecutor(), ImmutableList.of(runner));
}
@Before
diff --git a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java
index a54a210..9222ba5 100644
--- a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java
+++ b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java
@@ -24,11 +24,11 @@ import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.collections.CloseableDefaultBlockingPool;
import org.apache.druid.collections.CloseableStupidPool;
import org.apache.druid.collections.ReferenceCountingResourceHolder;
import org.apache.druid.data.input.Row;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.query.DruidProcessingConfig;
import org.apache.druid.query.InsufficientResourcesException;
@@ -183,7 +183,7 @@ public class GroupByQueryRunnerFailureTest
public GroupByQueryRunnerFailureTest(QueryRunner<Row> runner)
{
- this.runner = factory.mergeRunners(MoreExecutors.sameThreadExecutor(), ImmutableList.of(runner));
+ this.runner = factory.mergeRunners(Execs.directExecutor(), ImmutableList.of(runner));
}
@Test(timeout = 60_000L)
diff --git a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java
index 646e2d3..9142e4d 100644
--- a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java
+++ b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerTest.java
@@ -29,7 +29,6 @@ import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
import com.google.common.collect.Ordering;
import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.collections.CloseableDefaultBlockingPool;
import org.apache.druid.collections.CloseableStupidPool;
import org.apache.druid.common.config.NullHandling;
@@ -40,6 +39,7 @@ import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.DurationGranularity;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.granularity.PeriodGranularity;
@@ -441,7 +441,7 @@ public class GroupByQueryRunnerTest
{
this.config = config;
this.factory = factory;
- this.runner = factory.mergeRunners(MoreExecutors.sameThreadExecutor(), ImmutableList.of(runner));
+ this.runner = factory.mergeRunners(Execs.directExecutor(), ImmutableList.of(runner));
}
@Test
@@ -452,33 +452,251 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("quality", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new FloatSumAggregatorFactory("idxFloat", "indexFloat"),
- new DoubleSumAggregatorFactory("idxDouble", "index"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new FloatSumAggregatorFactory("idxFloat", "indexFloat"),
+ new DoubleSumAggregatorFactory("idxDouble", "index")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "rows", 1L, "idx", 135L, "idxFloat", 135.88510131835938f, "idxDouble", 135.88510131835938d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 1L, "idx", 118L, "idxFloat", 118.57034, "idxDouble", 118.57034),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "entertainment", "rows", 1L, "idx", 158L, "idxFloat", 158.747224, "idxDouble", 158.747224),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "rows", 1L, "idx", 120L, "idxFloat", 120.134704, "idxDouble", 120.134704),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "rows", 3L, "idx", 2870L, "idxFloat", 2871.8866900000003f, "idxDouble", 2871.8866900000003d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "rows", 1L, "idx", 121L, "idxFloat", 121.58358f, "idxDouble", 121.58358d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "premium", "rows", 3L, "idx", 2900L, "idxFloat", 2900.798647f, "idxDouble", 2900.798647d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technology", "rows", 1L, "idx", 78L, "idxFloat", 78.622547f, "idxDouble", 78.622547d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "travel", "rows", 1L, "idx", 119L, "idxFloat", 119.922742f, "idxDouble", 119.922742d),
-
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotive", "rows", 1L, "idx", 147L, "idxFloat", 147.42593f, "idxDouble", 147.42593d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "business", "rows", 1L, "idx", 112L, "idxFloat", 112.987027f, "idxDouble", 112.987027d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "entertainment", "rows", 1L, "idx", 166L, "idxFloat", 166.016049f, "idxDouble", 166.016049d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "health", "rows", 1L, "idx", 113L, "idxFloat", 113.446008f, "idxDouble", 113.446008d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzanine", "rows", 3L, "idx", 2447L, "idxFloat", 2448.830613f, "idxDouble", 2448.830613d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "news", "rows", 1L, "idx", 114L, "idxFloat", 114.290141f, "idxDouble", 114.290141d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "rows", 3L, "idx", 2505L, "idxFloat", 2506.415148f, "idxDouble", 2506.415148d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technology", "rows", 1L, "idx", 97L, "idxFloat", 97.387433f, "idxDouble", 97.387433d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "travel", "rows", 1L, "idx", 126L, "idxFloat", 126.411364f, "idxDouble", 126.411364d)
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "automotive",
+ "rows",
+ 1L,
+ "idx",
+ 135L,
+ "idxFloat",
+ 135.88510131835938f,
+ "idxDouble",
+ 135.88510131835938d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "business",
+ "rows",
+ 1L,
+ "idx",
+ 118L,
+ "idxFloat",
+ 118.57034,
+ "idxDouble",
+ 118.57034
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 158L,
+ "idxFloat",
+ 158.747224,
+ "idxDouble",
+ 158.747224
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "health",
+ "rows",
+ 1L,
+ "idx",
+ 120L,
+ "idxFloat",
+ 120.134704,
+ "idxDouble",
+ 120.134704
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "mezzanine",
+ "rows",
+ 3L,
+ "idx",
+ 2870L,
+ "idxFloat",
+ 2871.8866900000003f,
+ "idxDouble",
+ 2871.8866900000003d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "news",
+ "rows",
+ 1L,
+ "idx",
+ 121L,
+ "idxFloat",
+ 121.58358f,
+ "idxDouble",
+ 121.58358d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "premium",
+ "rows",
+ 3L,
+ "idx",
+ 2900L,
+ "idxFloat",
+ 2900.798647f,
+ "idxDouble",
+ 2900.798647d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "technology",
+ "rows",
+ 1L,
+ "idx",
+ 78L,
+ "idxFloat",
+ 78.622547f,
+ "idxDouble",
+ 78.622547d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "travel",
+ "rows",
+ 1L,
+ "idx",
+ 119L,
+ "idxFloat",
+ 119.922742f,
+ "idxDouble",
+ 119.922742d
+ ),
+
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "automotive",
+ "rows",
+ 1L,
+ "idx",
+ 147L,
+ "idxFloat",
+ 147.42593f,
+ "idxDouble",
+ 147.42593d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "business",
+ "rows",
+ 1L,
+ "idx",
+ 112L,
+ "idxFloat",
+ 112.987027f,
+ "idxDouble",
+ 112.987027d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 166L,
+ "idxFloat",
+ 166.016049f,
+ "idxDouble",
+ 166.016049d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "health",
+ "rows",
+ 1L,
+ "idx",
+ 113L,
+ "idxFloat",
+ 113.446008f,
+ "idxDouble",
+ 113.446008d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "mezzanine",
+ "rows",
+ 3L,
+ "idx",
+ 2447L,
+ "idxFloat",
+ 2448.830613f,
+ "idxDouble",
+ 2448.830613d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "news",
+ "rows",
+ 1L,
+ "idx",
+ 114L,
+ "idxFloat",
+ 114.290141f,
+ "idxDouble",
+ 114.290141d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "premium",
+ "rows",
+ 3L,
+ "idx",
+ 2505L,
+ "idxFloat",
+ 2506.415148f,
+ "idxDouble",
+ 2506.415148d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "technology",
+ "rows",
+ 1L,
+ "idx",
+ 97L,
+ "idxFloat",
+ 97.387433f,
+ "idxDouble",
+ 97.387433d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "travel",
+ "rows",
+ 1L,
+ "idx",
+ 126L,
+ "idxFloat",
+ 126.411364f,
+ "idxDouble",
+ 126.411364d
+ )
);
Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
@@ -538,115 +756,295 @@ public class GroupByQueryRunnerTest
.build();
List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "travel", "post", "travelx", "rows", 1L, "idx", 119L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technology", "post", "technologyx", "rows", 1L, "idx", 78L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "premium", "post", "premiumx", "rows", 3L, "idx", 2900L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "post", "newsx", "rows", 1L, "idx", 121L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "post", "mezzaninex", "rows", 3L, "idx", 2870L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "post", "healthx", "rows", 1L, "idx", 120L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "entertainment", "post", "entertainmentx", "rows", 1L, "idx", 158L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "post", "businessx", "rows", 1L, "idx", 118L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "post", "automotivex", "rows", 1L, "idx", 135L),
-
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "travel", "post", "travelx", "rows", 1L, "idx", 126L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technology", "post", "technologyx", "rows", 1L, "idx", 97L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "post", "premiumx", "rows", 3L, "idx", 2505L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "news", "post", "newsx", "rows", 1L, "idx", 114L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzanine", "post", "mezzaninex", "rows", 3L, "idx", 2447L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "health", "post", "healthx", "rows", 1L, "idx", 113L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "entertainment", "post", "entertainmentx", "rows", 1L, "idx", 166L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "business", "post", "businessx", "rows", 1L, "idx", 112L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotive", "post", "automotivex", "rows", 1L, "idx", 147L)
- );
-
- Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
- TestHelper.assertExpectedObjects(expectedResults, results, "string-postAgg");
- }
-
- @Test
- public void testGroupByWithStringVirtualColumn()
- {
- GroupByQuery query = GroupByQuery
- .builder()
- .setDataSource(QueryRunnerTestHelper.dataSource)
- .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setVirtualColumns(
- new ExpressionVirtualColumn(
- "vc",
- "quality + 'x'",
- ValueType.STRING,
- TestExprMacroTable.INSTANCE
- )
- )
- .setDimensions(new DefaultDimensionSpec("vc", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount, new LongSumAggregatorFactory("idx", "index"))
- .setGranularity(QueryRunnerTestHelper.dayGran)
- .build();
-
- List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotivex", "rows", 1L, "idx", 135L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "businessx", "rows", 1L, "idx", 118L),
GroupByQueryRunnerTestHelper.createExpectedRow(
"2011-04-01",
"alias",
- "entertainmentx",
+ "travel",
+ "post",
+ "travelx",
"rows",
1L,
"idx",
- 158L
+ 119L
),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "healthx", "rows", 1L, "idx", 120L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzaninex", "rows", 3L, "idx", 2870L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "newsx", "rows", 1L, "idx", 121L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "premiumx", "rows", 3L, "idx", 2900L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technologyx", "rows", 1L, "idx", 78L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "travelx", "rows", 1L, "idx", 119L),
-
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotivex", "rows", 1L, "idx", 147L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "businessx", "rows", 1L, "idx", 112L),
GroupByQueryRunnerTestHelper.createExpectedRow(
- "2011-04-02",
+ "2011-04-01",
"alias",
- "entertainmentx",
+ "technology",
+ "post",
+ "technologyx",
"rows",
1L,
"idx",
- 166L
+ 78L
),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "healthx", "rows", 1L, "idx", 113L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzaninex", "rows", 3L, "idx", 2447L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "newsx", "rows", 1L, "idx", 114L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premiumx", "rows", 3L, "idx", 2505L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technologyx", "rows", 1L, "idx", 97L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "travelx", "rows", 1L, "idx", 126L)
- );
-
- Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
- TestHelper.assertExpectedObjects(expectedResults, results, "virtual-column");
- }
-
- @Test
- public void testGroupByWithDurationGranularity()
- {
- GroupByQuery query = GroupByQuery
- .builder()
- .setDataSource(QueryRunnerTestHelper.dataSource)
- .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setDimensions(new DefaultDimensionSpec("quality", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount, new LongSumAggregatorFactory("idx", "index"))
- .setGranularity(new DurationGranularity(86400L, 0L))
- .build();
-
- List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "rows", 1L, "idx", 135L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 1L, "idx", 118L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "entertainment", "rows", 1L, "idx", 158L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "rows", 1L, "idx", 120L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "rows", 3L, "idx", 2870L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "rows", 1L, "idx", 121L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "premium", "rows", 3L, "idx", 2900L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technology", "rows", 1L, "idx", 78L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "travel", "rows", 1L, "idx", 119L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "premium",
+ "post",
+ "premiumx",
+ "rows",
+ 3L,
+ "idx",
+ 2900L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "news",
+ "post",
+ "newsx",
+ "rows",
+ 1L,
+ "idx",
+ 121L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "mezzanine",
+ "post",
+ "mezzaninex",
+ "rows",
+ 3L,
+ "idx",
+ 2870L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "health",
+ "post",
+ "healthx",
+ "rows",
+ 1L,
+ "idx",
+ 120L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "entertainment",
+ "post",
+ "entertainmentx",
+ "rows",
+ 1L,
+ "idx",
+ 158L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "business",
+ "post",
+ "businessx",
+ "rows",
+ 1L,
+ "idx",
+ 118L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "automotive",
+ "post",
+ "automotivex",
+ "rows",
+ 1L,
+ "idx",
+ 135L
+ ),
+
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "travel",
+ "post",
+ "travelx",
+ "rows",
+ 1L,
+ "idx",
+ 126L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "technology",
+ "post",
+ "technologyx",
+ "rows",
+ 1L,
+ "idx",
+ 97L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "premium",
+ "post",
+ "premiumx",
+ "rows",
+ 3L,
+ "idx",
+ 2505L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "news",
+ "post",
+ "newsx",
+ "rows",
+ 1L,
+ "idx",
+ 114L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "mezzanine",
+ "post",
+ "mezzaninex",
+ "rows",
+ 3L,
+ "idx",
+ 2447L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "health",
+ "post",
+ "healthx",
+ "rows",
+ 1L,
+ "idx",
+ 113L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "entertainment",
+ "post",
+ "entertainmentx",
+ "rows",
+ 1L,
+ "idx",
+ 166L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "business",
+ "post",
+ "businessx",
+ "rows",
+ 1L,
+ "idx",
+ 112L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "automotive",
+ "post",
+ "automotivex",
+ "rows",
+ 1L,
+ "idx",
+ 147L
+ )
+ );
+
+ Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
+ TestHelper.assertExpectedObjects(expectedResults, results, "string-postAgg");
+ }
+
+ @Test
+ public void testGroupByWithStringVirtualColumn()
+ {
+ GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.dataSource)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
+ .setVirtualColumns(
+ new ExpressionVirtualColumn(
+ "vc",
+ "quality + 'x'",
+ ValueType.STRING,
+ TestExprMacroTable.INSTANCE
+ )
+ )
+ .setDimensions(new DefaultDimensionSpec("vc", "alias"))
+ .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount, new LongSumAggregatorFactory("idx", "index"))
+ .setGranularity(QueryRunnerTestHelper.dayGran)
+ .build();
+
+ List<Row> expectedResults = Arrays.asList(
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotivex", "rows", 1L, "idx", 135L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "businessx", "rows", 1L, "idx", 118L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "entertainmentx",
+ "rows",
+ 1L,
+ "idx",
+ 158L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "healthx", "rows", 1L, "idx", 120L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzaninex", "rows", 3L, "idx", 2870L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "newsx", "rows", 1L, "idx", 121L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "premiumx", "rows", 3L, "idx", 2900L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technologyx", "rows", 1L, "idx", 78L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "travelx", "rows", 1L, "idx", 119L),
+
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotivex", "rows", 1L, "idx", 147L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "businessx", "rows", 1L, "idx", 112L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "entertainmentx",
+ "rows",
+ 1L,
+ "idx",
+ 166L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "healthx", "rows", 1L, "idx", 113L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzaninex", "rows", 3L, "idx", 2447L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "newsx", "rows", 1L, "idx", 114L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premiumx", "rows", 3L, "idx", 2505L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technologyx", "rows", 1L, "idx", 97L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "travelx", "rows", 1L, "idx", 126L)
+ );
+
+ Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
+ TestHelper.assertExpectedObjects(expectedResults, results, "virtual-column");
+ }
+
+ @Test
+ public void testGroupByWithDurationGranularity()
+ {
+ GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.dataSource)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
+ .setDimensions(new DefaultDimensionSpec("quality", "alias"))
+ .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount, new LongSumAggregatorFactory("idx", "index"))
+ .setGranularity(new DurationGranularity(86400L, 0L))
+ .build();
+
+ List<Row> expectedResults = Arrays.asList(
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "rows", 1L, "idx", 135L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 1L, "idx", 118L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "entertainment", "rows", 1L, "idx", 158L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "rows", 1L, "idx", 120L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "rows", 3L, "idx", 2870L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "rows", 1L, "idx", 121L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "premium", "rows", 3L, "idx", 2900L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technology", "rows", 1L, "idx", 78L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "travel", "rows", 1L, "idx", 119L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotive", "rows", 1L, "idx", 147L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "business", "rows", 1L, "idx", 112L),
@@ -838,29 +1236,69 @@ public class GroupByQueryRunnerTest
.build();
List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "a", "alias2", "a", "rows", 2L, "idx", 282L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "a", "alias2", "preferred", "rows", 2L, "idx", 282L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "preferred", "alias2", "a", "rows", 2L, "idx", 282L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "preferred", "alias2", "preferred", "rows", 2L, "idx", 282L)
- );
-
- Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
- TestHelper.assertExpectedObjects(expectedResults, results, "two-multi-value-dims");
- }
-
- @Test
- public void testMultipleDimensionsOneOfWhichIsMultiValue1()
- {
- GroupByQuery query = GroupByQuery
- .builder()
- .setDataSource(QueryRunnerTestHelper.dataSource)
- .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setDimensions(
- new DefaultDimensionSpec("placementish", "alias"),
- new DefaultDimensionSpec("quality", "quality")
- ).setAggregatorSpecs(QueryRunnerTestHelper.rowsCount, new LongSumAggregatorFactory("idx", "index"))
- .setGranularity(QueryRunnerTestHelper.allGran)
- .build();
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "a",
+ "alias2",
+ "a",
+ "rows",
+ 2L,
+ "idx",
+ 282L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "a",
+ "alias2",
+ "preferred",
+ "rows",
+ 2L,
+ "idx",
+ 282L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "preferred",
+ "alias2",
+ "a",
+ "rows",
+ 2L,
+ "idx",
+ 282L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "preferred",
+ "alias2",
+ "preferred",
+ "rows",
+ 2L,
+ "idx",
+ 282L
+ )
+ );
+
+ Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
+ TestHelper.assertExpectedObjects(expectedResults, results, "two-multi-value-dims");
+ }
+
+ @Test
+ public void testMultipleDimensionsOneOfWhichIsMultiValue1()
+ {
+ GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.dataSource)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
+ .setDimensions(
+ new DefaultDimensionSpec("placementish", "alias"),
+ new DefaultDimensionSpec("quality", "quality")
+ ).setAggregatorSpecs(QueryRunnerTestHelper.rowsCount, new LongSumAggregatorFactory("idx", "index"))
+ .setGranularity(QueryRunnerTestHelper.allGran)
+ .build();
List<Row> expectedResults = Arrays.asList(
GroupByQueryRunnerTestHelper.createExpectedRow(
@@ -1304,25 +1742,41 @@ public class GroupByQueryRunnerTest
expectedException.expect(ResourceLimitExceededException.class);
} else {
expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "rows", 1L, "idx", 135L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 1L, "idx", 118L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "entertainment", "rows", 1L, "idx", 158L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "rows", 1L, "idx", 120L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "rows", 3L, "idx", 2870L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "rows", 1L, "idx", 121L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "premium", "rows", 3L, "idx", 2900L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technology", "rows", 1L, "idx", 78L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "travel", "rows", 1L, "idx", 119L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "rows", 1L, "idx", 135L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 1L, "idx", 118L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 158L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "rows", 1L, "idx", 120L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "rows", 3L, "idx", 2870L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "rows", 1L, "idx", 121L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "premium", "rows", 3L, "idx", 2900L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technology", "rows", 1L, "idx", 78L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "travel", "rows", 1L, "idx", 119L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotive", "rows", 1L, "idx", 147L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "business", "rows", 1L, "idx", 112L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "entertainment", "rows", 1L, "idx", 166L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "health", "rows", 1L, "idx", 113L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzanine", "rows", 3L, "idx", 2447L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "news", "rows", 1L, "idx", 114L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "rows", 3L, "idx", 2505L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technology", "rows", 1L, "idx", 97L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "travel", "rows", 1L, "idx", 126L)
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotive", "rows", 1L, "idx", 147L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "business", "rows", 1L, "idx", 112L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 166L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "health", "rows", 1L, "idx", 113L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzanine", "rows", 3L, "idx", 2447L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "news", "rows", 1L, "idx", 114L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "rows", 3L, "idx", 2505L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technology", "rows", 1L, "idx", 97L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "travel", "rows", 1L, "idx", 126L)
);
}
@@ -1390,7 +1844,15 @@ public class GroupByQueryRunnerTest
expectedResults = Arrays.asList(
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "rows", 1L, "idx", 135L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 1L, "idx", 118L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "entertainment", "rows", 1L, "idx", 158L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 158L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "rows", 1L, "idx", 120L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "rows", 3L, "idx", 2870L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "rows", 1L, "idx", 121L),
@@ -1400,7 +1862,15 @@ public class GroupByQueryRunnerTest
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotive", "rows", 1L, "idx", 147L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "business", "rows", 1L, "idx", 112L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "entertainment", "rows", 1L, "idx", 166L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 166L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "health", "rows", 1L, "idx", 113L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzanine", "rows", 3L, "idx", 2447L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "news", "rows", 1L, "idx", 114L),
@@ -1435,7 +1905,15 @@ public class GroupByQueryRunnerTest
expectedResults = Arrays.asList(
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "rows", 1L, "idx", 135L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 1L, "idx", 118L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "entertainment", "rows", 1L, "idx", 158L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 158L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "rows", 1L, "idx", 120L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "rows", 3L, "idx", 2870L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "rows", 1L, "idx", 121L),
@@ -1445,7 +1923,15 @@ public class GroupByQueryRunnerTest
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotive", "rows", 1L, "idx", 147L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "business", "rows", 1L, "idx", 112L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "entertainment", "rows", 1L, "idx", 166L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 166L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "health", "rows", 1L, "idx", 113L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzanine", "rows", 3L, "idx", 2447L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "news", "rows", 1L, "idx", 114L),
@@ -1485,7 +1971,15 @@ public class GroupByQueryRunnerTest
expectedResults = Arrays.asList(
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "rows", 1L, "idx", 135L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 1L, "idx", 118L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "entertainment", "rows", 1L, "idx", 158L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 158L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "rows", 1L, "idx", 120L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "rows", 3L, "idx", 2870L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "rows", 1L, "idx", 121L),
@@ -1495,7 +1989,15 @@ public class GroupByQueryRunnerTest
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotive", "rows", 1L, "idx", 147L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "business", "rows", 1L, "idx", 112L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "entertainment", "rows", 1L, "idx", 166L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 166L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "health", "rows", 1L, "idx", 113L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzanine", "rows", 3L, "idx", 2447L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "news", "rows", 1L, "idx", 114L),
@@ -1981,23 +2483,73 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.fullOnInterval)
.setDimensions(new DefaultDimensionSpec("market", "market"))
- .setAggregatorSpecs(new LongFirstAggregatorFactory("first", "index"),
- new LongLastAggregatorFactory("last", "index"))
+ .setAggregatorSpecs(
+ new LongFirstAggregatorFactory("first", "index"),
+ new LongLastAggregatorFactory("last", "index")
+ )
.setGranularity(QueryRunnerTestHelper.monthGran)
.build();
List<Row> expectedResults = Arrays.asList(
GroupByQueryRunnerTestHelper.createExpectedRow("2011-01-01", "market", "spot", "first", 100L, "last", 155L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-01-01", "market", "total_market", "first", 1000L, "last", 1127L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-01-01",
+ "market",
+ "total_market",
+ "first",
+ 1000L,
+ "last",
+ 1127L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-01-01", "market", "upfront", "first", 800L, "last", 943L),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-02-01", "market", "spot", "first", 132L, "last", 114L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-02-01", "market", "total_market", "first", 1203L, "last", 1292L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-02-01", "market", "upfront", "first", 1667L, "last", 1101L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-02-01",
+ "market",
+ "total_market",
+ "first",
+ 1203L,
+ "last",
+ 1292L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-02-01",
+ "market",
+ "upfront",
+ "first",
+ 1667L,
+ "last",
+ 1101L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-03-01", "market", "spot", "first", 153L, "last", 125L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-03-01", "market", "total_market", "first", 1124L, "last", 1366L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-03-01", "market", "upfront", "first", 1166L, "last", 1063L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-03-01",
+ "market",
+ "total_market",
+ "first",
+ 1124L,
+ "last",
+ 1366L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-03-01",
+ "market",
+ "upfront",
+ "first",
+ 1166L,
+ "last",
+ 1063L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "market", "spot", "first", 135L, "last", 120L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "market", "total_market", "first", 1314L, "last", 1029L),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "market",
+ "total_market",
+ "first",
+ 1314L,
+ "last",
+ 1029L
+ ),
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "market", "upfront", "first", 1447L, "last", 780L)
);
@@ -2013,11 +2565,13 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.emptyInterval)
.setDimensions(new DefaultDimensionSpec("market", "market"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- QueryRunnerTestHelper.indexLongSum,
- QueryRunnerTestHelper.qualityCardinality,
- new LongFirstAggregatorFactory("first", "index"),
- new LongLastAggregatorFactory("last", "index"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ QueryRunnerTestHelper.indexLongSum,
+ QueryRunnerTestHelper.qualityCardinality,
+ new LongFirstAggregatorFactory("first", "index"),
+ new LongLastAggregatorFactory("last", "index")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
@@ -2716,7 +3270,8 @@ public class GroupByQueryRunnerTest
TestHelper.assertExpectedObjects(
expectedResults,
mergeRunner.run(QueryPlus.wrap(builder.build()), context),
- "no-limit");
+ "no-limit"
+ );
TestHelper.assertExpectedObjects(
Iterables.limit(expectedResults, 5),
mergeRunner.run(QueryPlus.wrap(builder.setLimit(5).build()), context),
@@ -2832,8 +3387,16 @@ public class GroupByQueryRunnerTest
.setInterval("2011-04-02/2011-04-04")
.setDimensions(new DefaultDimensionSpec("quality", "alias"))
.setAggregatorSpecs(QueryRunnerTestHelper.rowsCount, new LongSumAggregatorFactory("idx", "index"))
- .addOrderByColumn(new OrderByColumnSpec("rows", OrderByColumnSpec.Direction.DESCENDING, StringComparators.NUMERIC))
- .addOrderByColumn(new OrderByColumnSpec("alias", OrderByColumnSpec.Direction.ASCENDING, StringComparators.NUMERIC))
+ .addOrderByColumn(new OrderByColumnSpec(
+ "rows",
+ OrderByColumnSpec.Direction.DESCENDING,
+ StringComparators.NUMERIC
+ ))
+ .addOrderByColumn(new OrderByColumnSpec(
+ "alias",
+ OrderByColumnSpec.Direction.ASCENDING,
+ StringComparators.NUMERIC
+ ))
.setGranularity(new PeriodGranularity(new Period("P1M"), null, null));
final GroupByQuery query = builder.build();
@@ -3597,10 +4160,18 @@ public class GroupByQueryRunnerTest
public void testDimFilterHavingSpecWithExtractionFns()
{
String extractionJsFn = "function(str) { return 'super-' + str; }";
- ExtractionFn extractionFn = new JavaScriptExtractionFn(extractionJsFn, false, JavaScriptConfig.getEnabledInstance());
+ ExtractionFn extractionFn = new JavaScriptExtractionFn(
+ extractionJsFn,
+ false,
+ JavaScriptConfig.getEnabledInstance()
+ );
String extractionJsFn2 = "function(num) { return num + 10; }";
- ExtractionFn extractionFn2 = new JavaScriptExtractionFn(extractionJsFn2, false, JavaScriptConfig.getEnabledInstance());
+ ExtractionFn extractionFn2 = new JavaScriptExtractionFn(
+ extractionJsFn2,
+ false,
+ JavaScriptConfig.getEnabledInstance()
+ );
List<Row> expectedResults = Arrays.asList(
GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 2L, "idx", 217L),
@@ -4014,9 +4585,11 @@ public class GroupByQueryRunnerTest
null,
JavaScriptConfig.getEnabledInstance()
))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
@@ -4070,9 +4643,11 @@ public class GroupByQueryRunnerTest
null,
JavaScriptConfig.getEnabledInstance()
))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
@@ -4133,9 +4708,11 @@ public class GroupByQueryRunnerTest
null,
JavaScriptConfig.getEnabledInstance()
))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.setContext(ImmutableMap.of("chunkPeriod", "P1D"))
.build();
@@ -4199,9 +4776,11 @@ public class GroupByQueryRunnerTest
null,
JavaScriptConfig.getEnabledInstance()
))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
@@ -4238,9 +4817,11 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("quality", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
@@ -4248,9 +4829,11 @@ public class GroupByQueryRunnerTest
.builder()
.setDataSource(subquery)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new DoubleMaxAggregatorFactory("idx", "idx"),
- new DoubleMaxAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new DoubleMaxAggregatorFactory("idx", "idx"),
+ new DoubleMaxAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
@@ -4269,9 +4852,11 @@ public class GroupByQueryRunnerTest
.setVirtualColumns(
new ExpressionVirtualColumn("expr", "-index + 100", ValueType.FLOAT, TestExprMacroTable.INSTANCE)
)
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "expr"),
- new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "expr"),
+ new LongSumAggregatorFactory("indexMaxPlusTen", "indexMaxPlusTen")
+ )
.build();
query = (GroupByQuery) query.withDataSource(new QueryDataSource(subquery));
@@ -4315,10 +4900,12 @@ public class GroupByQueryRunnerTest
.builder()
.setDataSource(subquery)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setAggregatorSpecs(new DoubleMaxAggregatorFactory("idx1", "idx"),
- new DoubleMaxAggregatorFactory("idx2", "idx"),
- new DoubleMaxAggregatorFactory("idx3", "post_agg"),
- new DoubleMaxAggregatorFactory("idx4", "post_agg"))
+ .setAggregatorSpecs(
+ new DoubleMaxAggregatorFactory("idx1", "idx"),
+ new DoubleMaxAggregatorFactory("idx2", "idx"),
+ new DoubleMaxAggregatorFactory("idx3", "post_agg"),
+ new DoubleMaxAggregatorFactory("idx4", "post_agg")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
@@ -4532,8 +5119,10 @@ public class GroupByQueryRunnerTest
.setDataSource(subquery)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("alias", "alias"))
- .setAggregatorSpecs(new LongSumAggregatorFactory("rows", "rows"),
- new LongSumAggregatorFactory("idx", "idx_subpostagg"))
+ .setAggregatorSpecs(
+ new LongSumAggregatorFactory("rows", "rows"),
+ new LongSumAggregatorFactory("idx", "idx_subpostagg")
+ )
.setPostAggregatorSpecs(
Collections.singletonList(
new ArithmeticPostAggregator(
@@ -4801,8 +5390,10 @@ public class GroupByQueryRunnerTest
.setDataSource(subquery)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("alias", "alias"))
- .setAggregatorSpecs(new LongSumAggregatorFactory("rows", "rows"),
- new LongSumAggregatorFactory("idx", "idx_subpostagg"))
+ .setAggregatorSpecs(
+ new LongSumAggregatorFactory("rows", "rows"),
+ new LongSumAggregatorFactory("idx", "idx_subpostagg")
+ )
.setPostAggregatorSpecs(
Collections.singletonList(
new ArithmeticPostAggregator(
@@ -5016,16 +5607,18 @@ public class GroupByQueryRunnerTest
null,
JavaScriptConfig.getEnabledInstance()
))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new DoubleSumAggregatorFactory("idx_subagg", "index"),
- new JavaScriptAggregatorFactory(
- "js_agg",
- Arrays.asList("index", "market"),
- "function(current, index, dim){return current + index + dim.length;}",
- "function(){return 0;}",
- "function(a,b){return a + b;}",
- JavaScriptConfig.getEnabledInstance()
- ))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new DoubleSumAggregatorFactory("idx_subagg", "index"),
+ new JavaScriptAggregatorFactory(
+ "js_agg",
+ Arrays.asList("index", "market"),
+ "function(current, index, dim){return current + index + dim.length;}",
+ "function(){return 0;}",
+ "function(a,b){return a + b;}",
+ JavaScriptConfig.getEnabledInstance()
+ )
+ )
.setPostAggregatorSpecs(
Collections.singletonList(
new ArithmeticPostAggregator(
@@ -5058,9 +5651,11 @@ public class GroupByQueryRunnerTest
.setDataSource(subquery)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("alias", "alias"))
- .setAggregatorSpecs(new LongSumAggregatorFactory("rows", "rows"),
- new LongSumAggregatorFactory("idx", "idx_subpostagg"),
- new DoubleSumAggregatorFactory("js_outer_agg", "js_agg"))
+ .setAggregatorSpecs(
+ new LongSumAggregatorFactory("rows", "rows"),
+ new LongSumAggregatorFactory("idx", "idx_subpostagg"),
+ new DoubleSumAggregatorFactory("js_outer_agg", "js_agg")
+ )
.setPostAggregatorSpecs(
Collections.singletonList(
new ArithmeticPostAggregator(
@@ -5200,8 +5795,16 @@ public class GroupByQueryRunnerTest
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
- final DimFilter fridayFilter = new SelectorDimFilter(ColumnHolder.TIME_COLUMN_NAME, "Friday", new TimeFormatExtractionFn("EEEE", null, null, null, false));
- final DimFilter firstDaysFilter = new InDimFilter(ColumnHolder.TIME_COLUMN_NAME, ImmutableList.of("1", "2", "3"), new TimeFormatExtractionFn("d", null, null, null, false));
+ final DimFilter fridayFilter = new SelectorDimFilter(
+ ColumnHolder.TIME_COLUMN_NAME,
+ "Friday",
+ new TimeFormatExtractionFn("EEEE", null, null, null, false)
+ );
+ final DimFilter firstDaysFilter = new InDimFilter(
+ ColumnHolder.TIME_COLUMN_NAME,
+ ImmutableList.of("1", "2", "3"),
+ new TimeFormatExtractionFn("d", null, null, null, false)
+ );
final GroupByQuery query = GroupByQuery
.builder()
.setDataSource(subquery)
@@ -5472,9 +6075,11 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("quality", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new HyperUniquesAggregatorFactory("quality_uniques", "quality_uniques"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new HyperUniquesAggregatorFactory("quality_uniques", "quality_uniques")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
@@ -5483,9 +6088,11 @@ public class GroupByQueryRunnerTest
.setDataSource(subquery)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("alias", "alias"))
- .setAggregatorSpecs(new LongSumAggregatorFactory("rows", "rows"),
- new LongSumAggregatorFactory("idx", "idx"),
- new HyperUniquesAggregatorFactory("uniq", "quality_uniques"))
+ .setAggregatorSpecs(
+ new LongSumAggregatorFactory("rows", "rows"),
+ new LongSumAggregatorFactory("idx", "idx"),
+ new HyperUniquesAggregatorFactory("uniq", "quality_uniques")
+ )
.setGranularity(QueryRunnerTestHelper.allGran)
.build();
@@ -5517,313 +6124,1135 @@ public class GroupByQueryRunnerTest
"alias",
"entertainment",
"rows",
- 2L,
+ 2L,
+ "idx",
+ 324L,
+ "uniq",
+ 1.0002442201269182
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "health",
+ "rows",
+ 2L,
+ "idx",
+ 233L,
+ "uniq",
+ 1.0002442201269182
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "mezzanine",
+ "rows",
+ 6L,
+ "idx",
+ 5317L,
+ "uniq",
+ 1.0002442201269182
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "news",
+ "rows",
+ 2L,
+ "idx",
+ 235L,
+ "uniq",
+ 1.0002442201269182
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "premium",
+ "rows",
+ 6L,
+ "idx",
+ 5405L,
+ "uniq",
+ 1.0002442201269182
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "technology",
+ "rows",
+ 2L,
+ "idx",
+ 175L,
+ "uniq",
+ 1.0002442201269182
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "travel",
+ "rows",
+ 2L,
+ "idx",
+ 245L,
+ "uniq",
+ 1.0002442201269182
+ )
+ );
+
+ // Subqueries are handled by the ToolChest
+ Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
+ TestHelper.assertExpectedObjects(expectedResults, results, "subquery-hyperunique");
+ }
+
+ @Test
+ public void testSubqueryWithHyperUniquesPostAggregator()
+ {
+ GroupByQuery subquery = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.dataSource)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
+ .setDimensions(new ArrayList<>())
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new HyperUniquesAggregatorFactory("quality_uniques_inner", "quality_uniques")
+ )
+ .setPostAggregatorSpecs(
+ Collections.singletonList(
+ new FieldAccessPostAggregator("quality_uniques_inner_post", "quality_uniques_inner")
+ )
+ )
+ .setGranularity(QueryRunnerTestHelper.dayGran)
+ .build();
+
+ GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(subquery)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
+ .setDimensions(new ArrayList<>())
+ .setAggregatorSpecs(
+ new LongSumAggregatorFactory("rows", "rows"),
+ new LongSumAggregatorFactory("idx", "idx"),
+ new HyperUniquesAggregatorFactory("quality_uniques_outer", "quality_uniques_inner_post")
+ )
+ .setPostAggregatorSpecs(
+ Collections.singletonList(
+ new HyperUniqueFinalizingPostAggregator("quality_uniques_outer_post", "quality_uniques_outer")
+ )
+ )
+ .setGranularity(QueryRunnerTestHelper.allGran)
+ .build();
+
+ List<Row> expectedResults = Collections.singletonList(
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "rows",
+ 26L,
+ "idx",
+ 12446L,
+ "quality_uniques_outer",
+ 9.019833517963864,
+ "quality_uniques_outer_post",
+ 9.019833517963864
+ )
+ );
+
+ // Subqueries are handled by the ToolChest
+ Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
+ TestHelper.assertExpectedObjects(expectedResults, results, "subquery-hyperunique");
+ }
+
+ @Test
+ public void testSubqueryWithFirstLast()
+ {
+ GroupByQuery subquery = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.dataSource)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.fullOnInterval)
+ .setDimensions(new DefaultDimensionSpec("market", "market"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongFirstAggregatorFactory("innerfirst", "index"),
+ new LongLastAggregatorFactory("innerlast", "index")
+ )
+ .setGranularity(QueryRunnerTestHelper.dayGran)
+ .setContext(ImmutableMap.of("finalize", true))
+ .build();
+
+ GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(subquery)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.fullOnInterval)
+ .setDimensions(new ArrayList<>())
+ .setAggregatorSpecs(
+ new LongFirstAggregatorFactory("first", "innerfirst"),
+ new LongLastAggregatorFactory("last", "innerlast")
+ )
+ .setGranularity(QueryRunnerTestHelper.monthGran)
+ .build();
+
+ List<Row> expectedResults = Arrays.asList(
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-01-01", "first", 100L, "last", 943L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-02-01", "first", 132L, "last", 1101L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-03-01", "first", 153L, "last", 1063L),
+ GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "first", 135L, "last", 780L)
+ );
+
+ Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
+ TestHelper.assertExpectedObjects(expectedResults, results, "subquery-firstlast");
+ }
+
+ @Test
+ public void testGroupByWithSubtotalsSpec()
+ {
+ if (!config.getDefaultStrategy().equals(GroupByStrategySelector.STRATEGY_V2)) {
+ return;
+ }
+
+ GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.dataSource)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
+ .setDimensions(Lists.newArrayList(
+ new DefaultDimensionSpec("quality", "alias"),
+ new DefaultDimensionSpec("market", "market")
+ ))
+ .setAggregatorSpecs(
+ Arrays.asList(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new FloatSumAggregatorFactory("idxFloat", "indexFloat"),
+ new DoubleSumAggregatorFactory("idxDouble", "index")
+ )
+ )
+ .setGranularity(QueryRunnerTestHelper.dayGran)
+ .setSubtotalsSpec(ImmutableList.of(
+ ImmutableList.of("alias"),
+ ImmutableList.of("market"),
+ ImmutableList.of()
+ ))
+ .build();
+
+ List<Row> expectedResults = Arrays.asList(
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "automotive",
+ "rows",
+ 1L,
+ "idx",
+ 135L,
+ "idxFloat",
+ 135.88510131835938f,
+ "idxDouble",
+ 135.88510131835938d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "business",
+ "rows",
+ 1L,
+ "idx",
+ 118L,
+ "idxFloat",
+ 118.57034,
+ "idxDouble",
+ 118.57034
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 158L,
+ "idxFloat",
+ 158.747224,
+ "idxDouble",
+ 158.747224
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "health",
+ "rows",
+ 1L,
+ "idx",
+ 120L,
+ "idxFloat",
+ 120.134704,
+ "idxDouble",
+ 120.134704
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "mezzanine",
+ "rows",
+ 3L,
+ "idx",
+ 2870L,
+ "idxFloat",
+ 2871.8866900000003f,
+ "idxDouble",
+ 2871.8866900000003d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "news",
+ "rows",
+ 1L,
+ "idx",
+ 121L,
+ "idxFloat",
+ 121.58358f,
+ "idxDouble",
+ 121.58358d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "premium",
+ "rows",
+ 3L,
+ "idx",
+ 2900L,
+ "idxFloat",
+ 2900.798647f,
+ "idxDouble",
+ 2900.798647d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "technology",
+ "rows",
+ 1L,
+ "idx",
+ 78L,
+ "idxFloat",
+ 78.622547f,
+ "idxDouble",
+ 78.622547d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "travel",
+ "rows",
+ 1L,
+ "idx",
+ 119L,
+ "idxFloat",
+ 119.922742f,
+ "idxDouble",
+ 119.922742d
+ ),
+
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "automotive",
+ "rows",
+ 1L,
+ "idx",
+ 147L,
+ "idxFloat",
+ 147.42593f,
+ "idxDouble",
+ 147.42593d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "business",
+ "rows",
+ 1L,
+ "idx",
+ 112L,
+ "idxFloat",
+ 112.987027f,
+ "idxDouble",
+ 112.987027d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "entertainment",
+ "rows",
+ 1L,
+ "idx",
+ 166L,
+ "idxFloat",
+ 166.016049f,
+ "idxDouble",
+ 166.016049d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "health",
+ "rows",
+ 1L,
+ "idx",
+ 113L,
+ "idxFloat",
+ 113.446008f,
+ "idxDouble",
+ 113.446008d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "mezzanine",
+ "rows",
+ 3L,
+ "idx",
+ 2447L,
+ "idxFloat",
+ 2448.830613f,
+ "idxDouble",
+ 2448.830613d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "news",
+ "rows",
+ 1L,
+ "idx",
+ 114L,
+ "idxFloat",
+ 114.290141f,
+ "idxDouble",
+ 114.290141d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "premium",
+ "rows",
+ 3L,
+ "idx",
+ 2505L,
+ "idxFloat",
+ 2506.415148f,
+ "idxDouble",
+ 2506.415148d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "technology",
+ "rows",
+ 1L,
+ "idx",
+ 97L,
+ "idxFloat",
+ 97.387433f,
+ "idxDouble",
+ 97.387433d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "travel",
+ "rows",
+ 1L,
+ "idx",
+ 126L,
+ "idxFloat",
+ 126.411364f,
+ "idxDouble",
+ 126.411364d
+ ),
+
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 643.043177,
+ "idxFloat",
+ 643.043212890625,
+ "rows",
+ 5L,
+ "idx",
+ 640L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "total_market",
+ "idxDouble",
+ 1314.839715,
+ "idxFloat",
+ 1314.8397,
+ "rows",
+ 1L,
+ "idx",
+ 1314L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "upfront",
+ "idxDouble",
+ 1447.34116,
+ "idxFloat",
+ 1447.3412,
+ "rows",
+ 1L,
+ "idx",
+ 1447L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 266.090949,
+ "idxFloat",
+ 266.0909423828125,
+ "rows",
+ 2L,
+ "idx",
+ 265L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "total_market",
+ "idxDouble",
+ 1522.043733,
+ "idxFloat",
+ 1522.0437,
+ "rows",
+ 1L,
+ "idx",
+ 1522L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "upfront",
+ "idxDouble",
+ 1234.247546,
+ "idxFloat",
+ 1234.2476,
+ "rows",
+ 1L,
+ "idx",
+ 1234L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 198.545289,
+ "idxFloat",
+ 198.5452880859375,
+ "rows",
+ 2L,
+ "idx",
+ 197L
+ ),
+
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 650.806953,
+ "idxFloat",
+ 650.8069458007812,
+ "rows",
+ 5L,
+ "idx",
+ 648L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "total_market",
+ "idxDouble",
+ 1193.556278,
+ "idxFloat",
+ 1193.5563,
+ "rows",
+ 1L,
+ "idx",
+ 1193L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "upfront",
+ "idxDouble",
+ 1144.342401,
+ "idxFloat",
+ 1144.3424,
+ "rows",
+ 1L,
+ "idx",
+ 1144L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 249.591647,
+ "idxFloat",
+ 249.59164428710938,
+ "rows",
+ 2L,
+ "idx",
+ 249L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "total_market",
+ "idxDouble",
+ 1321.375057,
+ "idxFloat",
+ 1321.375,
+ "rows",
+ 1L,
+ "idx",
+ 1321L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "upfront",
+ "idxDouble",
+ 1049.738585,
+ "idxFloat",
+ 1049.7385,
+ "rows",
+ 1L,
+ "idx",
+ 1049L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 223.798797,
+ "idxFloat",
+ 223.79879760742188,
+ "rows",
+ 2L,
+ "idx",
+ 223L
+ ),
+
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 6626.151575318359,
+ "idxFloat",
+ 6626.152f,
+ "rows",
+ 13L,
+ "idx",
+ 6619L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 5833.209713,
+ "idxFloat",
+ 5833.209f,
+ "rows",
+ 13L,
+ "idx",
+ 5827L
+ )
+ );
+
+ Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
+ TestHelper.assertExpectedObjects(expectedResults, results, "subtotal");
+ }
+
+ @Test
+ public void testGroupByWithSubtotalsSpecWithLongDimensionColumn()
+ {
+ if (!config.getDefaultStrategy().equals(GroupByStrategySelector.STRATEGY_V2)) {
+ return;
+ }
+
+ GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.dataSource)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
+ .setDimensions(Lists.newArrayList(
+ new DefaultDimensionSpec("qualityLong", "ql", ValueType.LONG),
+ new DefaultDimensionSpec("market", "market")
+ ))
+ .setAggregatorSpecs(
+ Arrays.asList(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index"),
+ new FloatSumAggregatorFactory("idxFloat", "indexFloat"),
+ new DoubleSumAggregatorFactory("idxDouble", "index")
+ )
+ )
+ .setGranularity(QueryRunnerTestHelper.dayGran)
+ .setSubtotalsSpec(ImmutableList.of(
+ ImmutableList.of("ql"),
+ ImmutableList.of("market"),
+ ImmutableList.of()
+ ))
+ .build();
+
+ List<Row> expectedResults = Arrays.asList(
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 135.885094,
+ "idxFloat",
+ 135.8851,
+ "ql",
+ 1000L,
+ "rows",
+ 1L,
+ "idx",
+ 135L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 118.57034,
+ "idxFloat",
+ 118.57034,
+ "ql",
+ 1100L,
+ "rows",
+ 1L,
+ "idx",
+ 118L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 158.747224,
+ "idxFloat",
+ 158.74722,
+ "ql",
+ 1200L,
+ "rows",
+ 1L,
+ "idx",
+ 158L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 120.134704,
+ "idxFloat",
+ 120.134705,
+ "ql",
+ 1300L,
+ "rows",
+ 1L,
+ "idx",
+ 120L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 2871.8866900000003,
+ "idxFloat",
+ 2871.88671875,
+ "ql",
+ 1400L,
+ "rows",
+ 3L,
+ "idx",
+ 2870L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 121.583581,
+ "idxFloat",
+ 121.58358,
+ "ql",
+ 1500L,
+ "rows",
+ 1L,
+ "idx",
+ 121L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 2900.798647,
+ "idxFloat",
+ 2900.798583984375,
+ "ql",
+ 1600L,
+ "rows",
+ 3L,
+ "idx",
+ 2900L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 78.622547,
+ "idxFloat",
+ 78.62254,
+ "ql",
+ 1700L,
+ "rows",
+ 1L,
+ "idx",
+ 78L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 119.922742,
+ "idxFloat",
+ 119.922745,
+ "ql",
+ 1800L,
+ "rows",
+ 1L,
+ "idx",
+ 119L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 147.425935,
+ "idxFloat",
+ 147.42593,
+ "ql",
+ 1000L,
+ "rows",
+ 1L,
+ "idx",
+ 147L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 112.987027,
+ "idxFloat",
+ 112.98703,
+ "ql",
+ 1100L,
+ "rows",
+ 1L,
+ "idx",
+ 112L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 166.016049,
+ "idxFloat",
+ 166.01605,
+ "ql",
+ 1200L,
+ "rows",
+ 1L,
+ "idx",
+ 166L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 113.446008,
+ "idxFloat",
+ 113.44601,
+ "ql",
+ 1300L,
+ "rows",
+ 1L,
+ "idx",
+ 113L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 2448.830613,
+ "idxFloat",
+ 2448.83056640625,
+ "ql",
+ 1400L,
+ "rows",
+ 3L,
+ "idx",
+ 2447L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 114.290141,
+ "idxFloat",
+ 114.29014,
+ "ql",
+ 1500L,
+ "rows",
+ 1L,
+ "idx",
+ 114L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 2506.415148,
+ "idxFloat",
+ 2506.4150390625,
+ "ql",
+ 1600L,
+ "rows",
+ 3L,
+ "idx",
+ 2505L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 97.387433,
+ "idxFloat",
+ 97.387436,
+ "ql",
+ 1700L,
+ "rows",
+ 1L,
+ "idx",
+ 97L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 126.411364,
+ "idxFloat",
+ 126.41136,
+ "ql",
+ 1800L,
+ "rows",
+ 1L,
+ "idx",
+ 126L
+ ),
+
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 643.043177,
+ "idxFloat",
+ 643.043212890625,
+ "rows",
+ 5L,
+ "idx",
+ 640L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "total_market",
+ "idxDouble",
+ 1314.839715,
+ "idxFloat",
+ 1314.8397,
+ "rows",
+ 1L,
+ "idx",
+ 1314L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "upfront",
+ "idxDouble",
+ 1447.34116,
+ "idxFloat",
+ 1447.3412,
+ "rows",
+ 1L,
+ "idx",
+ 1447L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 266.090949,
+ "idxFloat",
+ 266.0909423828125,
+ "rows",
+ 2L,
+ "idx",
+ 265L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "total_market",
+ "idxDouble",
+ 1522.043733,
+ "idxFloat",
+ 1522.0437,
+ "rows",
+ 1L,
+ "idx",
+ 1522L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "upfront",
+ "idxDouble",
+ 1234.247546,
+ "idxFloat",
+ 1234.2476,
+ "rows",
+ 1L,
+ "idx",
+ 1234L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 198.545289,
+ "idxFloat",
+ 198.5452880859375,
+ "rows",
+ 2L,
+ "idx",
+ 197L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 650.806953,
+ "idxFloat",
+ 650.8069458007812,
+ "rows",
+ 5L,
"idx",
- 324L,
- "uniq",
- 1.0002442201269182
+ 648L
),
GroupByQueryRunnerTestHelper.createExpectedRow(
- "2011-04-01",
- "alias",
- "health",
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "total_market",
+ "idxDouble",
+ 1193.556278,
+ "idxFloat",
+ 1193.5563,
"rows",
- 2L,
+ 1L,
"idx",
- 233L,
- "uniq",
- 1.0002442201269182
+ 1193L
),
GroupByQueryRunnerTestHelper.createExpectedRow(
- "2011-04-01",
- "alias",
- "mezzanine",
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "upfront",
+ "idxDouble",
+ 1144.342401,
+ "idxFloat",
+ 1144.3424,
"rows",
- 6L,
+ 1L,
"idx",
- 5317L,
- "uniq",
- 1.0002442201269182
+ 1144L
),
GroupByQueryRunnerTestHelper.createExpectedRow(
- "2011-04-01",
- "alias",
- "news",
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 249.591647,
+ "idxFloat",
+ 249.59164428710938,
"rows",
2L,
"idx",
- 235L,
- "uniq",
- 1.0002442201269182
+ 249L
),
GroupByQueryRunnerTestHelper.createExpectedRow(
- "2011-04-01",
- "alias",
- "premium",
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "total_market",
+ "idxDouble",
+ 1321.375057,
+ "idxFloat",
+ 1321.375,
"rows",
- 6L,
+ 1L,
"idx",
- 5405L,
- "uniq",
- 1.0002442201269182
+ 1321L
),
GroupByQueryRunnerTestHelper.createExpectedRow(
- "2011-04-01",
- "alias",
- "technology",
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "upfront",
+ "idxDouble",
+ 1049.738585,
+ "idxFloat",
+ 1049.7385,
"rows",
- 2L,
+ 1L,
"idx",
- 175L,
- "uniq",
- 1.0002442201269182
+ 1049L
),
GroupByQueryRunnerTestHelper.createExpectedRow(
- "2011-04-01",
- "alias",
- "travel",
+ "2011-04-02T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 223.798797,
+ "idxFloat",
+ 223.79879760742188,
"rows",
2L,
"idx",
- 245L,
- "uniq",
- 1.0002442201269182
- )
- );
-
- // Subqueries are handled by the ToolChest
- Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
- TestHelper.assertExpectedObjects(expectedResults, results, "subquery-hyperunique");
- }
-
- @Test
- public void testSubqueryWithHyperUniquesPostAggregator()
- {
- GroupByQuery subquery = GroupByQuery
- .builder()
- .setDataSource(QueryRunnerTestHelper.dataSource)
- .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setDimensions(new ArrayList<>())
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new HyperUniquesAggregatorFactory("quality_uniques_inner", "quality_uniques"))
- .setPostAggregatorSpecs(
- Collections.singletonList(
- new FieldAccessPostAggregator("quality_uniques_inner_post", "quality_uniques_inner")
- )
- )
- .setGranularity(QueryRunnerTestHelper.dayGran)
- .build();
-
- GroupByQuery query = GroupByQuery
- .builder()
- .setDataSource(subquery)
- .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setDimensions(new ArrayList<>())
- .setAggregatorSpecs(new LongSumAggregatorFactory("rows", "rows"),
- new LongSumAggregatorFactory("idx", "idx"),
- new HyperUniquesAggregatorFactory("quality_uniques_outer", "quality_uniques_inner_post"))
- .setPostAggregatorSpecs(
- Collections.singletonList(
- new HyperUniqueFinalizingPostAggregator("quality_uniques_outer_post", "quality_uniques_outer")
- )
- )
- .setGranularity(QueryRunnerTestHelper.allGran)
- .build();
+ 223L
+ ),
- List<Row> expectedResults = Collections.singletonList(
GroupByQueryRunnerTestHelper.createExpectedRow(
- "2011-04-01",
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 6626.151569,
+ "idxFloat",
+ 6626.1513671875,
"rows",
- 26L,
+ 13L,
"idx",
- 12446L,
- "quality_uniques_outer",
- 9.019833517963864,
- "quality_uniques_outer_post",
- 9.019833517963864
- )
- );
-
- // Subqueries are handled by the ToolChest
- Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
- TestHelper.assertExpectedObjects(expectedResults, results, "subquery-hyperunique");
- }
-
- @Test
- public void testSubqueryWithFirstLast()
- {
- GroupByQuery subquery = GroupByQuery
- .builder()
- .setDataSource(QueryRunnerTestHelper.dataSource)
- .setQuerySegmentSpec(QueryRunnerTestHelper.fullOnInterval)
- .setDimensions(new DefaultDimensionSpec("market", "market"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongFirstAggregatorFactory("innerfirst", "index"),
- new LongLastAggregatorFactory("innerlast", "index"))
- .setGranularity(QueryRunnerTestHelper.dayGran)
- .setContext(ImmutableMap.of("finalize", true))
- .build();
-
- GroupByQuery query = GroupByQuery
- .builder()
- .setDataSource(subquery)
- .setQuerySegmentSpec(QueryRunnerTestHelper.fullOnInterval)
- .setDimensions(new ArrayList<>())
- .setAggregatorSpecs(new LongFirstAggregatorFactory("first", "innerfirst"),
- new LongLastAggregatorFactory("last", "innerlast"))
- .setGranularity(QueryRunnerTestHelper.monthGran)
- .build();
-
- List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-01-01", "first", 100L, "last", 943L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-02-01", "first", 132L, "last", 1101L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-03-01", "first", 153L, "last", 1063L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "first", 135L, "last", 780L)
- );
-
- Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
- TestHelper.assertExpectedObjects(expectedResults, results, "subquery-firstlast");
- }
-
- @Test
- public void testGroupByWithSubtotalsSpec()
- {
- if (!config.getDefaultStrategy().equals(GroupByStrategySelector.STRATEGY_V2)) {
- return;
- }
-
- GroupByQuery query = GroupByQuery
- .builder()
- .setDataSource(QueryRunnerTestHelper.dataSource)
- .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setDimensions(Lists.newArrayList(new DefaultDimensionSpec("quality", "alias"), new DefaultDimensionSpec("market", "market")))
- .setAggregatorSpecs(
- Arrays.asList(
- QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new FloatSumAggregatorFactory("idxFloat", "indexFloat"),
- new DoubleSumAggregatorFactory("idxDouble", "index")
- )
- )
- .setGranularity(QueryRunnerTestHelper.dayGran)
- .setSubtotalsSpec(ImmutableList.of(
- ImmutableList.of("alias"),
- ImmutableList.of("market"),
- ImmutableList.of()
- ))
- .build();
-
- List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "automotive", "rows", 1L, "idx", 135L, "idxFloat", 135.88510131835938f, "idxDouble", 135.88510131835938d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "business", "rows", 1L, "idx", 118L, "idxFloat", 118.57034, "idxDouble", 118.57034),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "entertainment", "rows", 1L, "idx", 158L, "idxFloat", 158.747224, "idxDouble", 158.747224),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "health", "rows", 1L, "idx", 120L, "idxFloat", 120.134704, "idxDouble", 120.134704),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "mezzanine", "rows", 3L, "idx", 2870L, "idxFloat", 2871.8866900000003f, "idxDouble", 2871.8866900000003d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "news", "rows", 1L, "idx", 121L, "idxFloat", 121.58358f, "idxDouble", 121.58358d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "premium", "rows", 3L, "idx", 2900L, "idxFloat", 2900.798647f, "idxDouble", 2900.798647d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technology", "rows", 1L, "idx", 78L, "idxFloat", 78.622547f, "idxDouble", 78.622547d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "travel", "rows", 1L, "idx", 119L, "idxFloat", 119.922742f, "idxDouble", 119.922742d),
-
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "automotive", "rows", 1L, "idx", 147L, "idxFloat", 147.42593f, "idxDouble", 147.42593d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "business", "rows", 1L, "idx", 112L, "idxFloat", 112.987027f, "idxDouble", 112.987027d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "entertainment", "rows", 1L, "idx", 166L, "idxFloat", 166.016049f, "idxDouble", 166.016049d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "health", "rows", 1L, "idx", 113L, "idxFloat", 113.446008f, "idxDouble", 113.446008d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "mezzanine", "rows", 3L, "idx", 2447L, "idxFloat", 2448.830613f, "idxDouble", 2448.830613d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "news", "rows", 1L, "idx", 114L, "idxFloat", 114.290141f, "idxDouble", 114.290141d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "rows", 3L, "idx", 2505L, "idxFloat", 2506.415148f, "idxDouble", 2506.415148d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technology", "rows", 1L, "idx", 97L, "idxFloat", 97.387433f, "idxDouble", 97.387433d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "travel", "rows", 1L, "idx", 126L, "idxFloat", 126.411364f, "idxDouble", 126.411364d),
-
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "spot", "idxDouble", 643.043177, "idxFloat", 643.043212890625, "rows", 5L, "idx", 640L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "total_market", "idxDouble", 1314.839715, "idxFloat", 1314.8397, "rows", 1L, "idx", 1314L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "upfront", "idxDouble", 1447.34116, "idxFloat", 1447.3412, "rows", 1L, "idx", 1447L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "spot", "idxDouble", 266.090949, "idxFloat", 266.0909423828125, "rows", 2L, "idx", 265L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "total_market", "idxDouble", 1522.043733, "idxFloat", 1522.0437, "rows", 1L, "idx", 1522L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "upfront", "idxDouble", 1234.247546, "idxFloat", 1234.2476, "rows", 1L, "idx", 1234L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "spot", "idxDouble", 198.545289, "idxFloat", 198.5452880859375, "rows", 2L, "idx", 197L),
-
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "spot", "idxDouble", 650.806953, "idxFloat", 650.8069458007812, "rows", 5L, "idx", 648L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "total_market", "idxDouble", 1193.556278, "idxFloat", 1193.5563, "rows", 1L, "idx", 1193L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "upfront", "idxDouble", 1144.342401, "idxFloat", 1144.3424, "rows", 1L, "idx", 1144L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "spot", "idxDouble", 249.591647, "idxFloat", 249.59164428710938, "rows", 2L, "idx", 249L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "total_market", "idxDouble", 1321.375057, "idxFloat", 1321.375, "rows", 1L, "idx", 1321L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "upfront", "idxDouble", 1049.738585, "idxFloat", 1049.7385, "rows", 1L, "idx", 1049L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "spot", "idxDouble", 223.798797, "idxFloat", 223.79879760742188, "rows", 2L, "idx", 223L),
-
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 6626.151575318359, "idxFloat", 6626.152f, "rows", 13L, "idx", 6619L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 5833.209713, "idxFloat", 5833.209f, "rows", 13L, "idx", 5827L)
- );
-
- Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
- TestHelper.assertExpectedObjects(expectedResults, results, "subtotal");
- }
-
- @Test
- public void testGroupByWithSubtotalsSpecWithLongDimensionColumn()
- {
- if (!config.getDefaultStrategy().equals(GroupByStrategySelector.STRATEGY_V2)) {
- return;
- }
-
- GroupByQuery query = GroupByQuery
- .builder()
- .setDataSource(QueryRunnerTestHelper.dataSource)
- .setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setDimensions(Lists.newArrayList(new DefaultDimensionSpec("qualityLong", "ql", ValueType.LONG), new DefaultDimensionSpec("market", "market")))
- .setAggregatorSpecs(
- Arrays.asList(
- QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"),
- new FloatSumAggregatorFactory("idxFloat", "indexFloat"),
- new DoubleSumAggregatorFactory("idxDouble", "index")
- )
+ 6619L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02T00:00:00.000Z",
+ "idxDouble",
+ 5833.209717999999,
+ "idxFloat",
+ 5833.20849609375,
+ "rows",
+ 13L,
+ "idx",
+ 5827L
)
- .setGranularity(QueryRunnerTestHelper.dayGran)
- .setSubtotalsSpec(ImmutableList.of(
- ImmutableList.of("ql"),
- ImmutableList.of("market"),
- ImmutableList.of()
- ))
- .build();
-
- List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 135.885094, "idxFloat", 135.8851, "ql", 1000L, "rows", 1L, "idx", 135L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 118.57034, "idxFloat", 118.57034, "ql", 1100L, "rows", 1L, "idx", 118L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 158.747224, "idxFloat", 158.74722, "ql", 1200L, "rows", 1L, "idx", 158L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 120.134704, "idxFloat", 120.134705, "ql", 1300L, "rows", 1L, "idx", 120L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 2871.8866900000003, "idxFloat", 2871.88671875, "ql", 1400L, "rows", 3L, "idx", 2870L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 121.583581, "idxFloat", 121.58358, "ql", 1500L, "rows", 1L, "idx", 121L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 2900.798647, "idxFloat", 2900.798583984375, "ql", 1600L, "rows", 3L, "idx", 2900L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 78.622547, "idxFloat", 78.62254, "ql", 1700L, "rows", 1L, "idx", 78L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 119.922742, "idxFloat", 119.922745, "ql", 1800L, "rows", 1L, "idx", 119L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 147.425935, "idxFloat", 147.42593, "ql", 1000L, "rows", 1L, "idx", 147L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 112.987027, "idxFloat", 112.98703, "ql", 1100L, "rows", 1L, "idx", 112L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 166.016049, "idxFloat", 166.01605, "ql", 1200L, "rows", 1L, "idx", 166L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 113.446008, "idxFloat", 113.44601, "ql", 1300L, "rows", 1L, "idx", 113L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 2448.830613, "idxFloat", 2448.83056640625, "ql", 1400L, "rows", 3L, "idx", 2447L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 114.290141, "idxFloat", 114.29014, "ql", 1500L, "rows", 1L, "idx", 114L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 2506.415148, "idxFloat", 2506.4150390625, "ql", 1600L, "rows", 3L, "idx", 2505L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 97.387433, "idxFloat", 97.387436, "ql", 1700L, "rows", 1L, "idx", 97L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 126.411364, "idxFloat", 126.41136, "ql", 1800L, "rows", 1L, "idx", 126L),
-
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "spot", "idxDouble", 643.043177, "idxFloat", 643.043212890625, "rows", 5L, "idx", 640L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "total_market", "idxDouble", 1314.839715, "idxFloat", 1314.8397, "rows", 1L, "idx", 1314L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "upfront", "idxDouble", 1447.34116, "idxFloat", 1447.3412, "rows", 1L, "idx", 1447L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "spot", "idxDouble", 266.090949, "idxFloat", 266.0909423828125, "rows", 2L, "idx", 265L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "total_market", "idxDouble", 1522.043733, "idxFloat", 1522.0437, "rows", 1L, "idx", 1522L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "upfront", "idxDouble", 1234.247546, "idxFloat", 1234.2476, "rows", 1L, "idx", 1234L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "spot", "idxDouble", 198.545289, "idxFloat", 198.5452880859375, "rows", 2L, "idx", 197L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "spot", "idxDouble", 650.806953, "idxFloat", 650.8069458007812, "rows", 5L, "idx", 648L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "total_market", "idxDouble", 1193.556278, "idxFloat", 1193.5563, "rows", 1L, "idx", 1193L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "upfront", "idxDouble", 1144.342401, "idxFloat", 1144.3424, "rows", 1L, "idx", 1144L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "spot", "idxDouble", 249.591647, "idxFloat", 249.59164428710938, "rows", 2L, "idx", 249L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "total_market", "idxDouble", 1321.375057, "idxFloat", 1321.375, "rows", 1L, "idx", 1321L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "upfront", "idxDouble", 1049.738585, "idxFloat", 1049.7385, "rows", 1L, "idx", 1049L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "market", "spot", "idxDouble", 223.798797, "idxFloat", 223.79879760742188, "rows", 2L, "idx", 223L),
-
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 6626.151569, "idxFloat", 6626.1513671875, "rows", 13L, "idx", 6619L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02T00:00:00.000Z", "idxDouble", 5833.209717999999, "idxFloat", 5833.20849609375, "rows", 13L, "idx", 5827L)
);
Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
@@ -5845,7 +7274,10 @@ public class GroupByQueryRunnerTest
.builder()
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setDimensions(Lists.newArrayList(new DefaultDimensionSpec("quality", "alias"), new DefaultDimensionSpec("market", "market")))
+ .setDimensions(Lists.newArrayList(
+ new DefaultDimensionSpec("quality", "alias"),
+ new DefaultDimensionSpec("market", "market")
+ ))
.setAggregatorSpecs(
Arrays.asList(
QueryRunnerTestHelper.rowsCount,
@@ -5865,9 +7297,43 @@ public class GroupByQueryRunnerTest
.build();
List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technology", "rows", 1L, "idx", 78L, "idxFloat", 78.622547f, "idxDouble", 78.622547d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "market", "spot", "idxDouble", 198.545289, "idxFloat", 198.5452880859375, "rows", 2L, "idx", 197L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01T00:00:00.000Z", "idxDouble", 6626.151575318359, "idxFloat", 6626.152f, "rows", 13L, "idx", 6619L)
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "technology",
+ "rows",
+ 1L,
+ "idx",
+ 78L,
+ "idxFloat",
+ 78.622547f,
+ "idxDouble",
+ 78.622547d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "market",
+ "spot",
+ "idxDouble",
+ 198.545289,
+ "idxFloat",
+ 198.5452880859375,
+ "rows",
+ 2L,
+ "idx",
+ 197L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01T00:00:00.000Z",
+ "idxDouble",
+ 6626.151575318359,
+ "idxFloat",
+ 6626.152f,
+ "rows",
+ 13L,
+ "idx",
+ 6619L
+ )
);
Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
@@ -5881,9 +7347,11 @@ public class GroupByQueryRunnerTest
.builder()
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- QueryRunnerTestHelper.jsCountIfTimeGreaterThan,
- QueryRunnerTestHelper.__timeLongSum)
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ QueryRunnerTestHelper.jsCountIfTimeGreaterThan,
+ QueryRunnerTestHelper.__timeLongSum
+ )
.setGranularity(QueryRunnerTestHelper.allGran)
.build();
@@ -6483,8 +7951,12 @@ public class GroupByQueryRunnerTest
toolChest
)
);
-
- TestHelper.assertExpectedObjects(bySegmentResults, theRunner.run(QueryPlus.wrap(fullQuery), new HashMap<>()), "bySegment");
+
+ TestHelper.assertExpectedObjects(
+ bySegmentResults,
+ theRunner.run(QueryPlus.wrap(fullQuery), new HashMap<>()),
+ "bySegment"
+ );
exec.shutdownNow();
}
@@ -6544,7 +8016,11 @@ public class GroupByQueryRunnerTest
)
);
- TestHelper.assertExpectedObjects(bySegmentResults, theRunner.run(QueryPlus.wrap(fullQuery), new HashMap<>()), "bySegment-dim-extraction");
+ TestHelper.assertExpectedObjects(
+ bySegmentResults,
+ theRunner.run(QueryPlus.wrap(fullQuery), new HashMap<>()),
+ "bySegment-dim-extraction"
+ );
exec.shutdownNow();
}
@@ -6575,8 +8051,10 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("quality", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.setDimFilter(new OrDimFilter(dimFilters))
.build();
@@ -6627,8 +8105,10 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("quality", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.setDimFilter(new ExtractionDimFilter("quality", "", lookupExtractionFn, null))
.build();
@@ -6665,8 +8145,10 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("quality", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.setDimFilter(
new ExtractionDimFilter("quality", "NOT_THERE", lookupExtractionFn, null)
@@ -6699,8 +8181,10 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("null_column", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.setDimFilter(
new ExtractionDimFilter(
@@ -6903,8 +8387,10 @@ public class GroupByQueryRunnerTest
.setDataSource(QueryRunnerTestHelper.dataSource)
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("quality", "alias"))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("idx", "index"))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("idx", "index")
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.setDimFilter(
new ExtractionDimFilter(
@@ -6989,7 +8475,11 @@ public class GroupByQueryRunnerTest
String extractionJsFn = "function(str) { return 'super-' + str; }";
String jsFn = "function(x) { return(x === 'super-mezzanine') }";
- ExtractionFn extractionFn = new JavaScriptExtractionFn(extractionJsFn, false, JavaScriptConfig.getEnabledInstance());
+ ExtractionFn extractionFn = new JavaScriptExtractionFn(
+ extractionJsFn,
+ false,
+ JavaScriptConfig.getEnabledInstance()
+ );
List<DimFilter> superFilterList = new ArrayList<>();
superFilterList.add(new SelectorDimFilter("quality", "super-mezzanine", extractionFn));
@@ -7041,7 +8531,11 @@ public class GroupByQueryRunnerTest
)
);
- TestHelper.assertExpectedObjects(bySegmentResults, theRunner.run(QueryPlus.wrap(fullQuery), new HashMap<>()), "bySegment-filter");
+ TestHelper.assertExpectedObjects(
+ bySegmentResults,
+ theRunner.run(QueryPlus.wrap(fullQuery), new HashMap<>()),
+ "bySegment-filter"
+ );
exec.shutdownNow();
}
@@ -7066,7 +8560,12 @@ public class GroupByQueryRunnerTest
superFilterList.add(
new SearchQueryDimFilter("null_column", new ContainsSearchQuerySpec("EMPTY", true), extractionFn)
);
- superFilterList.add(new JavaScriptDimFilter("null_column", jsFn, extractionFn, JavaScriptConfig.getEnabledInstance()));
+ superFilterList.add(new JavaScriptDimFilter(
+ "null_column",
+ jsFn,
+ extractionFn,
+ JavaScriptConfig.getEnabledInstance()
+ ));
DimFilter superFilter = new AndDimFilter(superFilterList);
GroupByQuery query = GroupByQuery.builder()
@@ -7113,12 +8612,60 @@ public class GroupByQueryRunnerTest
.build();
List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "spot", "rows", 9L, "numVals", 1.0002442201269182d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "total_market", "rows", 2L, "numVals", 1.0002442201269182d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "upfront", "rows", 2L, "numVals", 1.0002442201269182d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "spot", "rows", 9L, "numVals", 1.0002442201269182d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "total_market", "rows", 2L, "numVals", 1.0002442201269182d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "upfront", "rows", 2L, "numVals", 1.0002442201269182d)
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "spot",
+ "rows",
+ 9L,
+ "numVals",
+ 1.0002442201269182d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "total_market",
+ "rows",
+ 2L,
+ "numVals",
+ 1.0002442201269182d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "upfront",
+ "rows",
+ 2L,
+ "numVals",
+ 1.0002442201269182d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "spot",
+ "rows",
+ 9L,
+ "numVals",
+ 1.0002442201269182d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "total_market",
+ "rows",
+ 2L,
+ "numVals",
+ 1.0002442201269182d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "upfront",
+ "rows",
+ 2L,
+ "numVals",
+ 1.0002442201269182d
+ )
);
Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
@@ -7145,12 +8692,60 @@ public class GroupByQueryRunnerTest
.build();
List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "spot", "rows", 9L, "numVals", 8.015665809687173d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "total_market", "rows", 2L, "numVals", 2.000977198748901d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "upfront", "rows", 2L, "numVals", 2.000977198748901d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "spot", "rows", 9L, "numVals", 9.019833517963864d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "total_market", "rows", 2L, "numVals", 2.000977198748901d),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "upfront", "rows", 2L, "numVals", 2.000977198748901d)
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "spot",
+ "rows",
+ 9L,
+ "numVals",
+ 8.015665809687173d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "total_market",
+ "rows",
+ 2L,
+ "numVals",
+ 2.000977198748901d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "upfront",
+ "rows",
+ 2L,
+ "numVals",
+ 2.000977198748901d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "spot",
+ "rows",
+ 9L,
+ "numVals",
+ 9.019833517963864d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "total_market",
+ "rows",
+ 2L,
+ "numVals",
+ 2.000977198748901d
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "upfront",
+ "rows",
+ 2L,
+ "numVals",
+ 2.000977198748901d
+ )
);
Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, runner, query);
@@ -7956,8 +9551,10 @@ public class GroupByQueryRunnerTest
)
)
)
- .setAggregatorSpecs(new LongSumAggregatorFactory("ql_alias_sum", "ql_alias"),
- new DoubleSumAggregatorFactory("qf_alias_sum", "qf_alias"))
+ .setAggregatorSpecs(
+ new LongSumAggregatorFactory("ql_alias_sum", "ql_alias"),
+ new DoubleSumAggregatorFactory("qf_alias_sum", "qf_alias")
+ )
.setGranularity(QueryRunnerTestHelper.allGran)
.build();
@@ -8003,8 +9600,10 @@ public class GroupByQueryRunnerTest
new DefaultDimensionSpec("alias", "market"),
new DefaultDimensionSpec("time_alias", "time_alias2", ValueType.LONG)
)
- .setAggregatorSpecs(new LongMaxAggregatorFactory("time_alias_max", "time_alias"),
- new DoubleMaxAggregatorFactory("index_alias_max", "index_alias"))
+ .setAggregatorSpecs(
+ new LongMaxAggregatorFactory("time_alias_max", "time_alias"),
+ new DoubleMaxAggregatorFactory("index_alias_max", "index_alias")
+ )
.setGranularity(QueryRunnerTestHelper.allGran)
.build();
@@ -8116,27 +9715,29 @@ public class GroupByQueryRunnerTest
.setQuerySegmentSpec(QueryRunnerTestHelper.firstToThird)
.setDimensions(new DefaultDimensionSpec("quality", "alias"))
.setDimFilter(new SelectorDimFilter("quality", "technology", null))
- .setAggregatorSpecs(QueryRunnerTestHelper.rowsCount,
- new LongSumAggregatorFactory("qlLong", "qualityLong"),
- new DoubleSumAggregatorFactory("qlFloat", "qualityLong"),
- new JavaScriptAggregatorFactory(
- "qlJs",
- ImmutableList.of("qualityLong"),
- "function(a,b) { return a + b; }",
- "function() { return 0; }",
- "function(a,b) { return a + b }",
- JavaScriptConfig.getEnabledInstance()
- ),
- new DoubleSumAggregatorFactory("qfFloat", "qualityFloat"),
- new LongSumAggregatorFactory("qfLong", "qualityFloat"),
- new JavaScriptAggregatorFactory(
- "qfJs",
- ImmutableList.of("qualityFloat"),
- "function(a,b) { return a + b; }",
- "function() { return 0; }",
- "function(a,b) { return a + b }",
- JavaScriptConfig.getEnabledInstance()
- ))
+ .setAggregatorSpecs(
+ QueryRunnerTestHelper.rowsCount,
+ new LongSumAggregatorFactory("qlLong", "qualityLong"),
+ new DoubleSumAggregatorFactory("qlFloat", "qualityLong"),
+ new JavaScriptAggregatorFactory(
+ "qlJs",
+ ImmutableList.of("qualityLong"),
+ "function(a,b) { return a + b; }",
+ "function() { return 0; }",
+ "function(a,b) { return a + b }",
+ JavaScriptConfig.getEnabledInstance()
+ ),
+ new DoubleSumAggregatorFactory("qfFloat", "qualityFloat"),
+ new LongSumAggregatorFactory("qfLong", "qualityFloat"),
+ new JavaScriptAggregatorFactory(
+ "qfJs",
+ ImmutableList.of("qualityFloat"),
+ "function(a,b) { return a + b; }",
+ "function() { return 0; }",
+ "function(a,b) { return a + b }",
+ JavaScriptConfig.getEnabledInstance()
+ )
+ )
.setGranularity(QueryRunnerTestHelper.dayGran)
.build();
@@ -8510,11 +10111,61 @@ public class GroupByQueryRunnerTest
Map<String, Object> context = new HashMap<>();
List<Row> allGranExpectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "travel", "market", "spot", "rows", 2L, "idx", 243L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technology", "market", "spot", "rows", 2L, "idx", 177L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "market", "upfront", "rows", 2L, "idx", 1817L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "market", "total_market", "rows", 2L, "idx", 2342L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "market", "spot", "rows", 2L, "idx", 257L)
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "travel",
+ "market",
+ "spot",
+ "rows",
+ 2L,
+ "idx",
+ 243L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "technology",
+ "market",
+ "spot",
+ "rows",
+ 2L,
+ "idx",
+ 177L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "premium",
+ "market",
+ "upfront",
+ "rows",
+ 2L,
+ "idx",
+ 1817L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "premium",
+ "market",
+ "total_market",
+ "rows",
+ 2L,
+ "idx",
+ 2342L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "premium",
+ "market",
+ "spot",
+ "rows",
+ 2L,
+ "idx",
+ 257L
+ )
);
Iterable<Row> results = mergedRunner.run(QueryPlus.wrap(allGranQuery), context).toList();
@@ -8593,11 +10244,61 @@ public class GroupByQueryRunnerTest
Map<String, Object> context = new HashMap<>();
List<Row> allGranExpectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "travel", "market", "spot", "rows", 2L, "idx", 243L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technology", "market", "spot", "rows", 2L, "idx", 177L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "market", "total_market", "rows", 2L, "idx", 2342L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "market", "upfront", "rows", 2L, "idx", 1817L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "premium", "market", "spot", "rows", 2L, "idx", 257L)
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "travel",
+ "market",
+ "spot",
+ "rows",
+ 2L,
+ "idx",
+ 243L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "technology",
+ "market",
+ "spot",
+ "rows",
+ 2L,
+ "idx",
+ 177L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "premium",
+ "market",
+ "total_market",
+ "rows",
+ 2L,
+ "idx",
+ 2342L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "premium",
+ "market",
+ "upfront",
+ "rows",
+ 2L,
+ "idx",
+ 1817L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "premium",
+ "market",
+ "spot",
+ "rows",
+ 2L,
+ "idx",
+ 257L
+ )
);
Iterable<Row> results = mergedRunner.run(QueryPlus.wrap(allGranQuery), context).toList();
@@ -8652,9 +10353,9 @@ public class GroupByQueryRunnerTest
.setLimitSpec(
new DefaultLimitSpec(
Collections.singletonList(new OrderByColumnSpec(
- "alias",
- OrderByColumnSpec.Direction.DESCENDING
- )),
+ "alias",
+ OrderByColumnSpec.Direction.DESCENDING
+ )),
5
)
)
@@ -8794,19 +10495,39 @@ public class GroupByQueryRunnerTest
.build();
List<Row> expectedResults = Arrays.asList(
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-01", "alias", "technology", "qualityLen", 10L, "rows", 2L, "idx", 156L),
- GroupByQueryRunnerTestHelper.createExpectedRow("2011-04-02", "alias", "technology", "qualityLen", 10L, "rows", 2L, "idx", 194L)
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-01",
+ "alias",
+ "technology",
+ "qualityLen",
+ 10L,
+ "rows",
+ 2L,
+ "idx",
+ 156L
+ ),
+ GroupByQueryRunnerTestHelper.createExpectedRow(
+ "2011-04-02",
+ "alias",
+ "technology",
+ "qualityLen",
+ 10L,
+ "rows",
+ 2L,
+ "idx",
+ 194L
+ )
);
ChainedExecutionQueryRunner ceqr = new ChainedExecutionQueryRunner(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
(query1, future) -> {
return;
},
ImmutableList.<QueryRunner<Row>>of(runner, runner)
);
- QueryRunner<Row> mergingRunner = factory.mergeRunners(MoreExecutors.sameThreadExecutor(), ImmutableList.of(ceqr));
+ QueryRunner<Row> mergingRunner = factory.mergeRunners(Execs.directExecutor(), ImmutableList.of(ceqr));
Iterable<Row> results = GroupByQueryRunnerTestHelper.runQuery(factory, mergingRunner, query);
TestHelper.assertExpectedObjects(expectedResults, results, "type-conversion");
diff --git a/processing/src/test/java/org/apache/druid/query/groupby/GroupByTimeseriesQueryRunnerTest.java b/processing/src/test/java/org/apache/druid/query/groupby/GroupByTimeseriesQueryRunnerTest.java
index 06a3e5e..0dc2a63 100644
--- a/processing/src/test/java/org/apache/druid/query/groupby/GroupByTimeseriesQueryRunnerTest.java
+++ b/processing/src/test/java/org/apache/druid/query/groupby/GroupByTimeseriesQueryRunnerTest.java
@@ -22,12 +22,12 @@ package org.apache.druid.query.groupby;
import com.google.common.base.Function;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.data.input.MapBasedRow;
import org.apache.druid.data.input.Row;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.guava.Sequence;
import org.apache.druid.java.util.common.guava.Sequences;
@@ -55,6 +55,7 @@ import java.io.IOException;
import java.util.Map;
/**
+ *
*/
@RunWith(Parameterized.class)
public class GroupByTimeseriesQueryRunnerTest extends TimeseriesQueryRunnerTest
@@ -93,7 +94,7 @@ public class GroupByTimeseriesQueryRunnerTest extends TimeseriesQueryRunnerTest
{
TimeseriesQuery tsQuery = (TimeseriesQuery) queryPlus.getQuery();
QueryRunner<Row> newRunner = factory.mergeRunners(
- MoreExecutors.sameThreadExecutor(), ImmutableList.of(input)
+ Execs.directExecutor(), ImmutableList.of(input)
);
QueryToolChest toolChest = factory.getToolchest();
diff --git a/processing/src/test/java/org/apache/druid/query/metadata/SegmentMetadataQueryTest.java b/processing/src/test/java/org/apache/druid/query/metadata/SegmentMetadataQueryTest.java
index 13e1d3e..1a22544 100644
--- a/processing/src/test/java/org/apache/druid/query/metadata/SegmentMetadataQueryTest.java
+++ b/processing/src/test/java/org/apache/druid/query/metadata/SegmentMetadataQueryTest.java
@@ -24,10 +24,10 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.data.input.impl.TimestampSpec;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.query.BySegmentResultValue;
import org.apache.druid.query.BySegmentResultValueClass;
@@ -304,7 +304,7 @@ public class SegmentMetadataQueryTest
QueryRunner myRunner = new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
FACTORY.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Lists.newArrayList(
toolChest.preMergeQueryDecoration(runner1),
toolChest.preMergeQueryDecoration(runner2)
@@ -372,7 +372,7 @@ public class SegmentMetadataQueryTest
QueryRunner myRunner = new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
FACTORY.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Lists.newArrayList(
toolChest.preMergeQueryDecoration(runner1),
toolChest.preMergeQueryDecoration(runner2)
@@ -440,7 +440,7 @@ public class SegmentMetadataQueryTest
QueryRunner myRunner = new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
FACTORY.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Lists.newArrayList(
toolChest.preMergeQueryDecoration(runner1),
toolChest.preMergeQueryDecoration(runner2)
@@ -557,7 +557,7 @@ public class SegmentMetadataQueryTest
QueryRunner myRunner = new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
FACTORY.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Lists.newArrayList(
toolChest.preMergeQueryDecoration(runner1),
toolChest.preMergeQueryDecoration(runner2)
@@ -609,7 +609,7 @@ public class SegmentMetadataQueryTest
QueryRunner myRunner = new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
FACTORY.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Lists.newArrayList(
toolChest.preMergeQueryDecoration(runner1),
toolChest.preMergeQueryDecoration(runner2)
@@ -671,7 +671,7 @@ public class SegmentMetadataQueryTest
QueryRunner myRunner = new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
FACTORY.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Lists.newArrayList(
toolChest.preMergeQueryDecoration(runner1),
toolChest.preMergeQueryDecoration(runner2)
@@ -729,7 +729,7 @@ public class SegmentMetadataQueryTest
QueryRunner myRunner = new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
FACTORY.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Lists.newArrayList(
toolChest.preMergeQueryDecoration(runner1),
toolChest.preMergeQueryDecoration(runner2)
@@ -787,7 +787,7 @@ public class SegmentMetadataQueryTest
QueryRunner myRunner = new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
FACTORY.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Lists.newArrayList(
toolChest.preMergeQueryDecoration(runner1),
toolChest.preMergeQueryDecoration(runner2)
@@ -832,7 +832,7 @@ public class SegmentMetadataQueryTest
QueryRunner myRunner = new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
FACTORY.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
//Note: It is essential to have atleast 2 query runners merged to reproduce the regression bug described in
//https://github.com/apache/incubator-druid/pull/1172
//the bug surfaces only when ordering is used which happens only when you have 2 things to compare
diff --git a/processing/src/test/java/org/apache/druid/query/scan/MultiSegmentScanQueryTest.java b/processing/src/test/java/org/apache/druid/query/scan/MultiSegmentScanQueryTest.java
index cc6ad9d..9d48b6a 100644
--- a/processing/src/test/java/org/apache/druid/query/scan/MultiSegmentScanQueryTest.java
+++ b/processing/src/test/java/org/apache/druid/query/scan/MultiSegmentScanQueryTest.java
@@ -22,10 +22,10 @@ package org.apache.druid.query.scan;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.io.CharSource;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.guava.MergeSequence;
import org.apache.druid.java.util.common.guava.Sequence;
@@ -60,6 +60,7 @@ import java.util.List;
import java.util.Map;
/**
+ *
*/
@RunWith(Parameterized.class)
public class MultiSegmentScanQueryTest
@@ -167,7 +168,10 @@ public class MultiSegmentScanQueryTest
@Parameterized.Parameters(name = "limit={0},batchSize={1}")
public static Iterable<Object[]> constructorFeeder()
{
- return QueryRunnerTestHelper.cartesian(Arrays.asList(0, 1, 3, 7, 10, 20, 1000), Arrays.asList(0, 1, 3, 6, 7, 10, 123, 2000));
+ return QueryRunnerTestHelper.cartesian(
+ Arrays.asList(0, 1, 3, 7, 10, 20, 1000),
+ Arrays.asList(0, 1, 3, 6, 7, 10, 123, 2000)
+ );
}
private final int limit;
@@ -196,8 +200,9 @@ public class MultiSegmentScanQueryTest
ScanQuery query = newBuilder().build();
List<ScanResultValue> results = factory
.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
- ImmutableList.of(factory.createRunner(segment0), factory.createRunner(segment1)))
+ Execs.directExecutor(),
+ ImmutableList.of(factory.createRunner(segment0), factory.createRunner(segment1))
+ )
.run(QueryPlus.wrap(query), new HashMap<>())
.toList();
int totalCount = 0;
@@ -215,7 +220,8 @@ public class MultiSegmentScanQueryTest
public void testMergeResultsWithLimit()
{
QueryRunner<ScanResultValue> runner = toolChest.mergeResults(
- new QueryRunner<ScanResultValue>() {
+ new QueryRunner<ScanResultValue>()
+ {
@Override
public Sequence<ScanResultValue> run(
QueryPlus<ScanResultValue> queryPlus,
diff --git a/processing/src/test/java/org/apache/druid/query/spec/SpecificSegmentQueryRunnerTest.java b/processing/src/test/java/org/apache/druid/query/spec/SpecificSegmentQueryRunnerTest.java
index 930e321..af64fe2 100644
--- a/processing/src/test/java/org/apache/druid/query/spec/SpecificSegmentQueryRunnerTest.java
+++ b/processing/src/test/java/org/apache/druid/query/spec/SpecificSegmentQueryRunnerTest.java
@@ -21,10 +21,10 @@ package org.apache.druid.query.spec;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableList;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.guava.Accumulator;
import org.apache.druid.java.util.common.guava.Sequence;
@@ -116,6 +116,7 @@ public class SpecificSegmentQueryRunnerTest
new YieldingAccumulator()
{
final List lists = new ArrayList<>();
+
@Override
public Object accumulate(Object accumulated, Object in)
{
@@ -162,7 +163,7 @@ public class SpecificSegmentQueryRunnerTest
throw new SegmentMissingException("FAILSAUCE");
}
},
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
}
},
diff --git a/server/src/main/java/org/apache/druid/client/cache/BackgroundCachePopulator.java b/server/src/main/java/org/apache/druid/client/cache/BackgroundCachePopulator.java
index fa1a9ea..0e78e99 100644
--- a/server/src/main/java/org/apache/druid/client/cache/BackgroundCachePopulator.java
+++ b/server/src/main/java/org/apache/druid/client/cache/BackgroundCachePopulator.java
@@ -27,6 +27,7 @@ import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.guava.Sequence;
import org.apache.druid.java.util.common.guava.Sequences;
import org.apache.druid.java.util.common.logger.Logger;
@@ -101,7 +102,7 @@ public class BackgroundCachePopulator implements CachePopulator
exec
);
},
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
}
diff --git a/server/src/main/java/org/apache/druid/segment/realtime/RealtimeManager.java b/server/src/main/java/org/apache/druid/segment/realtime/RealtimeManager.java
index bc07ba8..4b64376 100644
--- a/server/src/main/java/org/apache/druid/segment/realtime/RealtimeManager.java
+++ b/server/src/main/java/org/apache/druid/segment/realtime/RealtimeManager.java
@@ -26,7 +26,6 @@ import com.google.common.base.Preconditions;
import com.google.common.base.Supplier;
import com.google.common.base.Throwables;
import com.google.common.collect.Iterables;
-import com.google.common.util.concurrent.MoreExecutors;
import com.google.inject.Inject;
import org.apache.druid.data.input.Committer;
import org.apache.druid.data.input.Firehose;
@@ -65,6 +64,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
/**
+ *
*/
public class RealtimeManager implements QuerySegmentWalker
{
@@ -175,7 +175,7 @@ public class RealtimeManager implements QuerySegmentWalker
return partitionChiefs == null ? new NoopQueryRunner<T>() : factory.getToolchest().mergeResults(
factory.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
// Chaining query runners which wait on submitted chain query runners can make executor pools deadlock
Iterables.transform(
partitionChiefs.values(), new Function<FireChief, QueryRunner<T>>()
@@ -202,7 +202,7 @@ public class RealtimeManager implements QuerySegmentWalker
? new NoopQueryRunner<T>()
: factory.getToolchest().mergeResults(
factory.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Iterables.transform(
specs,
new Function<SegmentDescriptor, QueryRunner<T>>()
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 53f68f5..6e88cf1 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
@@ -23,7 +23,6 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.Iterables;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.client.CachingQueryRunner;
import org.apache.druid.client.cache.Cache;
import org.apache.druid.client.cache.CacheConfig;
@@ -31,6 +30,7 @@ import org.apache.druid.client.cache.CachePopulatorStats;
import org.apache.druid.client.cache.ForegroundCachePopulator;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.guava.CloseQuietly;
import org.apache.druid.java.util.common.guava.FunctionalIterable;
import org.apache.druid.java.util.emitter.EmittingLogger;
@@ -206,7 +206,7 @@ public class SinkQuerySegmentWalker implements QuerySegmentWalker
sinkSegmentIdentifier,
descriptor.getInterval().getStart(),
factory.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Iterables.transform(
theSink,
new Function<FireHydrant, QueryRunner<T>>()
diff --git a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorDriver.java b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorDriver.java
index 95f50b4..5bb25f8 100644
--- a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorDriver.java
+++ b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorDriver.java
@@ -26,12 +26,12 @@ import com.google.common.base.Supplier;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import org.apache.druid.data.input.Committer;
import org.apache.druid.data.input.InputRow;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.concurrent.ListenableFutures;
import org.apache.druid.java.util.common.guava.Comparators;
import org.apache.druid.java.util.common.logger.Logger;
@@ -209,7 +209,7 @@ public class StreamAppenderatorDriver extends BaseAppenderatorDriver
/**
* Persist all data indexed through this driver so far. Blocks until complete.
- *
+ * <p>
* Should be called after all data has been added through {@link #add(InputRow, String, Supplier, boolean, boolean)}.
*
* @param committer committer representing all data that has been added so far
@@ -235,7 +235,7 @@ public class StreamAppenderatorDriver extends BaseAppenderatorDriver
/**
* Persist all data indexed through this driver so far. Returns a future of persisted commitMetadata.
- *
+ * <p>
* Should be called after all data has been added through {@link #add(InputRow, String, Supplier, boolean, boolean)}.
*
* @param committer committer representing all data that has been added so far
@@ -331,7 +331,7 @@ public class StreamAppenderatorDriver extends BaseAppenderatorDriver
segmentIdentifier.getVersion(),
segmentIdentifier.getShardSpec().getPartitionNum()
),
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
() -> {
log.info("Segment[%s] successfully handed off, dropping.", segmentIdentifier);
metrics.incrementHandOffCount();
diff --git a/server/src/main/java/org/apache/druid/server/QueryManager.java b/server/src/main/java/org/apache/druid/server/QueryManager.java
index 0684ed4..a90bf4b 100644
--- a/server/src/main/java/org/apache/druid/server/QueryManager.java
+++ b/server/src/main/java/org/apache/druid/server/QueryManager.java
@@ -23,7 +23,7 @@ import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimaps;
import com.google.common.collect.SetMultimap;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.query.Query;
import org.apache.druid.query.QueryWatcher;
@@ -76,7 +76,7 @@ public class QueryManager implements QueryWatcher
}
}
},
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
}
diff --git a/server/src/test/java/org/apache/druid/client/CachingClusteredClientTest.java b/server/src/test/java/org/apache/druid/client/CachingClusteredClientTest.java
index f65dc30..3612136 100644
--- a/server/src/test/java/org/apache/druid/client/CachingClusteredClientTest.java
+++ b/server/src/test/java/org/apache/druid/client/CachingClusteredClientTest.java
@@ -62,6 +62,7 @@ import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.granularity.Granularity;
import org.apache.druid.java.util.common.granularity.PeriodGranularity;
@@ -165,6 +166,7 @@ import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Executor;
/**
+ *
*/
@RunWith(Parameterized.class)
public class CachingClusteredClientTest
@@ -331,7 +333,7 @@ public class CachingClusteredClientTest
final ListeningExecutorService delegate = MoreExecutors.listeningDecorator(
// we need to run everything in the same thread to ensure all callbacks on futures in CachingClusteredClient
// are complete before moving on to the next query run.
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
@Override
@@ -1174,14 +1176,14 @@ public class CachingClusteredClientTest
public void testSearchCachingRenamedOutput()
{
final Druids.SearchQueryBuilder builder = Druids.newSearchQueryBuilder()
- .dataSource(DATA_SOURCE)
- .filters(DIM_FILTER)
- .granularity(GRANULARITY)
- .limit(1000)
- .intervals(SEG_SPEC)
- .dimensions(Collections.singletonList(TOP_DIM))
- .query("how")
- .context(CONTEXT);
+ .dataSource(DATA_SOURCE)
+ .filters(DIM_FILTER)
+ .granularity(GRANULARITY)
+ .limit(1000)
+ .intervals(SEG_SPEC)
+ .dimensions(Collections.singletonList(TOP_DIM))
+ .query("how")
+ .context(CONTEXT);
testQueryCaching(
getDefaultQueryRunner(),
@@ -1347,14 +1349,14 @@ public class CachingClusteredClientTest
final Set<String> metrics = Sets.newHashSet("rows");
Druids.SelectQueryBuilder builder = Druids.newSelectQueryBuilder()
- .dataSource(DATA_SOURCE)
- .intervals(SEG_SPEC)
- .filters(DIM_FILTER)
- .granularity(GRANULARITY)
- .dimensions(Collections.singletonList("a"))
- .metrics(Collections.singletonList("rows"))
- .pagingSpec(new PagingSpec(null, 3))
- .context(CONTEXT);
+ .dataSource(DATA_SOURCE)
+ .intervals(SEG_SPEC)
+ .filters(DIM_FILTER)
+ .granularity(GRANULARITY)
+ .dimensions(Collections.singletonList("a"))
+ .metrics(Collections.singletonList("rows"))
+ .pagingSpec(new PagingSpec(null, 3))
+ .context(CONTEXT);
testQueryCaching(
getDefaultQueryRunner(),
@@ -1446,9 +1448,9 @@ public class CachingClusteredClientTest
public void testGroupByCaching()
{
List<AggregatorFactory> aggsWithUniques = ImmutableList.<AggregatorFactory>builder()
- .addAll(AGGS)
- .add(new HyperUniquesAggregatorFactory("uniques", "uniques"))
- .build();
+ .addAll(AGGS)
+ .add(new HyperUniquesAggregatorFactory("uniques", "uniques"))
+ .build();
final HashFunction hashFn = Hashing.murmur3_128();
@@ -1650,9 +1652,11 @@ public class CachingClusteredClientTest
makeTimeResults(DateTimes.of("2011-01-01"), 50, 5000,
DateTimes.of("2011-01-02"), 10, 1252,
DateTimes.of("2011-01-03"), 20, 6213,
- DateTimes.of("2011-01-04"), 30, 743),
+ DateTimes.of("2011-01-04"), 30, 743
+ ),
makeTimeResults(DateTimes.of("2011-01-07"), 60, 6020,
- DateTimes.of("2011-01-08"), 70, 250)
+ DateTimes.of("2011-01-08"), 70, 250
+ )
);
testQueryCachingWithFilter(
@@ -1690,14 +1694,14 @@ public class CachingClusteredClientTest
);
final Druids.TimeseriesQueryBuilder builder = Druids.newTimeseriesQueryBuilder()
- .dataSource(DATA_SOURCE)
- .filters(filter)
- .granularity(GRANULARITY)
- .intervals(SEG_SPEC)
- .context(CONTEXT)
- .intervals("2011-01-05/2011-01-10")
- .aggregators(RENAMED_AGGS)
- .postAggregators(RENAMED_POST_AGGS);
+ .dataSource(DATA_SOURCE)
+ .filters(filter)
+ .granularity(GRANULARITY)
+ .intervals(SEG_SPEC)
+ .context(CONTEXT)
+ .intervals("2011-01-05/2011-01-10")
+ .aggregators(RENAMED_AGGS)
+ .postAggregators(RENAMED_POST_AGGS);
TimeseriesQuery query = builder.build();
Map<String, Object> context = new HashMap<>();
@@ -1888,7 +1892,12 @@ public class CachingClusteredClientTest
@Override
public Sequence answer()
{
- return toFilteredQueryableTimeseriesResults((TimeseriesQuery) capture.getValue().getQuery(), segmentIds, queryIntervals, results);
+ return toFilteredQueryableTimeseriesResults(
+ (TimeseriesQuery) capture.getValue().getQuery(),
+ segmentIds,
+ queryIntervals,
+ results
+ );
}
})
.times(0, 1);
@@ -1946,7 +1955,11 @@ public class CachingClusteredClientTest
MultipleSpecificSegmentSpec spec = (MultipleSpecificSegmentSpec) query.getQuerySegmentSpec();
List<Result<TimeseriesResultValue>> ret = new ArrayList<>();
for (SegmentDescriptor descriptor : spec.getDescriptors()) {
- String id = StringUtils.format("%s_%s", queryIntervals.indexOf(descriptor.getInterval()), descriptor.getPartitionNumber());
+ String id = StringUtils.format(
+ "%s_%s",
+ queryIntervals.indexOf(descriptor.getInterval()),
+ descriptor.getPartitionNumber()
+ );
int index = segmentIds.indexOf(id);
if (index != -1) {
ret.add(new Result(
@@ -2466,13 +2479,13 @@ public class CachingClusteredClientTest
(DateTime) objects[i],
new TimeseriesResultValue(
ImmutableMap.<String, Object>builder()
- .put("rows", objects[i + 1])
- .put("imps", objects[i + 2])
- .put("impers", objects[i + 2])
- .put("avg_imps_per_row", avg_impr)
- .put("avg_imps_per_row_half", avg_impr / 2)
- .put("avg_imps_per_row_double", avg_impr * 2)
- .build()
+ .put("rows", objects[i + 1])
+ .put("imps", objects[i + 2])
+ .put("impers", objects[i + 2])
+ .put("avg_imps_per_row", avg_impr)
+ .put("avg_imps_per_row_half", avg_impr / 2)
+ .put("avg_imps_per_row_double", avg_impr * 2)
+ .build()
)
)
);
@@ -2722,7 +2735,8 @@ public class CachingClusteredClientTest
return mergeLimit;
}
},
- new DruidHttpClientConfig() {
+ new DruidHttpClientConfig()
+ {
@Override
public long getMaxQueuedBytes()
{
@@ -3126,7 +3140,8 @@ public class CachingClusteredClientTest
@SuppressWarnings("unchecked")
private QueryRunner getDefaultQueryRunner()
{
- return new QueryRunner() {
+ return new QueryRunner()
+ {
@Override
public Sequence run(final QueryPlus queryPlus, final Map responseContext)
{
diff --git a/server/src/test/java/org/apache/druid/client/HttpServerInventoryViewTest.java b/server/src/test/java/org/apache/druid/client/HttpServerInventoryViewTest.java
index be49011..bb43104 100644
--- a/server/src/test/java/org/apache/druid/client/HttpServerInventoryViewTest.java
+++ b/server/src/test/java/org/apache/druid/client/HttpServerInventoryViewTest.java
@@ -24,7 +24,6 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.discovery.DataNodeService;
import org.apache.druid.discovery.DiscoveryDruidNode;
import org.apache.druid.discovery.DruidNodeDiscovery;
@@ -32,6 +31,7 @@ import org.apache.druid.discovery.DruidNodeDiscoveryProvider;
import org.apache.druid.discovery.NodeType;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.http.client.HttpClient;
import org.apache.druid.java.util.http.client.Request;
import org.apache.druid.java.util.http.client.response.HttpResponseHandler;
@@ -64,6 +64,7 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
/**
+ *
*/
public class HttpServerInventoryViewTest
{
@@ -197,7 +198,7 @@ public class HttpServerInventoryViewTest
);
httpServerInventoryView.registerSegmentCallback(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
new ServerView.SegmentCallback()
{
@Override
@@ -225,7 +226,7 @@ public class HttpServerInventoryViewTest
final CountDownLatch serverRemovedCalled = new CountDownLatch(1);
httpServerInventoryView.registerServerRemovedCallback(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
new ServerView.ServerRemovedCallback()
{
@Override
@@ -254,8 +255,10 @@ public class HttpServerInventoryViewTest
segmentDropLatches.get(segment2.getIdentifier()).await();
DruidServer druidServer = httpServerInventoryView.getInventoryValue("host:8080");
- Assert.assertEquals(ImmutableMap.of(segment3.getIdentifier(), segment3, segment4.getIdentifier(), segment4),
- druidServer.getSegments());
+ Assert.assertEquals(
+ ImmutableMap.of(segment3.getIdentifier(), segment3, segment4.getIdentifier(), segment4),
+ druidServer.getSegments()
+ );
druidNodeDiscovery.listener.nodesRemoved(ImmutableList.of(druidNode));
@@ -320,7 +323,10 @@ public class HttpServerInventoryViewTest
if (requestNum.get() == 2) {
//fail scenario where request is sent to server but we got an unexpected response.
- HttpResponse httpResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR);
+ HttpResponse httpResponse = new DefaultHttpResponse(
+ HttpVersion.HTTP_1_1,
+ HttpResponseStatus.INTERNAL_SERVER_ERROR
+ );
httpResponse.setContent(ChannelBuffers.buffer(0));
httpResponseHandler.handleResponse(httpResponse, null);
return Futures.immediateFailedFuture(new RuntimeException("server error"));
diff --git a/server/src/test/java/org/apache/druid/client/client/BatchServerInventoryViewTest.java b/server/src/test/java/org/apache/druid/client/client/BatchServerInventoryViewTest.java
index c002469..987db9c 100644
--- a/server/src/test/java/org/apache/druid/client/client/BatchServerInventoryViewTest.java
+++ b/server/src/test/java/org/apache/druid/client/client/BatchServerInventoryViewTest.java
@@ -43,6 +43,7 @@ import org.apache.druid.curator.announcement.Announcer;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.guava.Comparators;
import org.apache.druid.segment.TestHelper;
import org.apache.druid.server.coordination.BatchDataSegmentAnnouncer;
@@ -77,6 +78,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
+ *
*/
public class BatchServerInventoryViewTest
{
@@ -118,7 +120,7 @@ public class BatchServerInventoryViewTest
announcer = new Announcer(
cf,
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
announcer.start();
@@ -204,7 +206,8 @@ public class BatchServerInventoryViewTest
return input.rhs.getInterval().getStart().isBefore(SEGMENT_INTERVAL_START.plusDays(INITIAL_SEGMENTS));
}
}
- ) {
+ )
+ {
@Override
protected DruidServer addInnerInventory(DruidServer container, String inventoryKey, Set<DataSegment> inventory)
{
@@ -337,7 +340,7 @@ public class BatchServerInventoryViewTest
EasyMock.replay(callback);
filteredBatchServerInventoryView.registerSegmentCallback(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
callback,
new Predicate<Pair<DruidServerMetadata, DataSegment>>()
{
@@ -407,7 +410,11 @@ public class BatchServerInventoryViewTest
while (inventoryUpdateCounter.get() != count) {
Thread.sleep(100);
if (stopwatch.elapsed(TimeUnit.MILLISECONDS) > forWaitingTiming.milliseconds()) {
- throw new ISE("BatchServerInventoryView is not updating counter expected[%d] value[%d]", count, inventoryUpdateCounter.get());
+ throw new ISE(
+ "BatchServerInventoryView is not updating counter expected[%d] value[%d]",
+ count,
+ inventoryUpdateCounter.get()
+ );
}
}
}
diff --git a/server/src/test/java/org/apache/druid/curator/discovery/CuratorDruidNodeAnnouncerAndDiscoveryTest.java b/server/src/test/java/org/apache/druid/curator/discovery/CuratorDruidNodeAnnouncerAndDiscoveryTest.java
index fcf6b6b..c2121a7 100644
--- a/server/src/test/java/org/apache/druid/curator/discovery/CuratorDruidNodeAnnouncerAndDiscoveryTest.java
+++ b/server/src/test/java/org/apache/druid/curator/discovery/CuratorDruidNodeAnnouncerAndDiscoveryTest.java
@@ -23,13 +23,13 @@ import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.curator.CuratorTestBase;
import org.apache.druid.curator.announcement.Announcer;
import org.apache.druid.discovery.DiscoveryDruidNode;
import org.apache.druid.discovery.DruidNodeDiscovery;
import org.apache.druid.discovery.NodeType;
import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.initialization.ServerConfig;
import org.apache.druid.server.initialization.ZkPathsConfig;
@@ -42,6 +42,7 @@ import java.util.HashSet;
import java.util.Set;
/**
+ *
*/
public class CuratorDruidNodeAnnouncerAndDiscoveryTest extends CuratorTestBase
{
@@ -68,7 +69,7 @@ public class CuratorDruidNodeAnnouncerAndDiscoveryTest extends CuratorTestBase
Announcer announcer = new Announcer(
curator,
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
announcer.start();
diff --git a/server/src/test/java/org/apache/druid/segment/realtime/RealtimeManagerTest.java b/server/src/test/java/org/apache/druid/segment/realtime/RealtimeManagerTest.java
index 0ab42c8..4e69322 100644
--- a/server/src/test/java/org/apache/druid/segment/realtime/RealtimeManagerTest.java
+++ b/server/src/test/java/org/apache/druid/segment/realtime/RealtimeManagerTest.java
@@ -27,7 +27,6 @@ import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.data.input.Committer;
import org.apache.druid.data.input.Firehose;
import org.apache.druid.data.input.FirehoseFactory;
@@ -41,6 +40,7 @@ import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.io.Closer;
import org.apache.druid.java.util.common.parsers.ParseException;
@@ -100,6 +100,7 @@ import java.util.Map;
import java.util.concurrent.TimeUnit;
/**
+ *
*/
public class RealtimeManagerTest
{
@@ -1018,7 +1019,8 @@ public class RealtimeManagerTest
}
@Override
- public IncrementalIndexAddResult add(InputRow row, Supplier<Committer> committerSupplier) throws IndexSizeExceededException
+ public IncrementalIndexAddResult add(InputRow row, Supplier<Committer> committerSupplier)
+ throws IndexSizeExceededException
{
if (row == null) {
return Plumber.THROWAWAY;
@@ -1055,7 +1057,7 @@ public class RealtimeManagerTest
return factory.getToolchest()
.mergeResults(
factory.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
Iterables.transform(
baseQuery.getIntervals(),
new Function<Interval, QueryRunner<T>>()
diff --git a/server/src/test/java/org/apache/druid/segment/realtime/plumber/CoordinatorBasedSegmentHandoffNotifierTest.java b/server/src/test/java/org/apache/druid/segment/realtime/plumber/CoordinatorBasedSegmentHandoffNotifierTest.java
index 228d795..3fcac21 100644
--- a/server/src/test/java/org/apache/druid/segment/realtime/plumber/CoordinatorBasedSegmentHandoffNotifierTest.java
+++ b/server/src/test/java/org/apache/druid/segment/realtime/plumber/CoordinatorBasedSegmentHandoffNotifierTest.java
@@ -19,9 +19,9 @@
package org.apache.druid.segment.realtime.plumber;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.client.coordinator.CoordinatorClient;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.query.SegmentDescriptor;
import org.easymock.EasyMock;
import org.joda.time.Duration;
@@ -62,7 +62,7 @@ public class CoordinatorBasedSegmentHandoffNotifierTest
final AtomicBoolean callbackCalled = new AtomicBoolean(false);
notifier.registerSegmentHandoffCallback(
descriptor,
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
() -> callbackCalled.set(true)
);
notifier.checkForSegmentHandoffs();
@@ -93,7 +93,7 @@ public class CoordinatorBasedSegmentHandoffNotifierTest
notifier.registerSegmentHandoffCallback(
descriptor,
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
() -> callbackCalled.set(true)
);
Assert.assertEquals(1, notifier.getHandOffCallbacks().size());
diff --git a/server/src/test/java/org/apache/druid/segment/realtime/plumber/RealtimePlumberSchoolTest.java b/server/src/test/java/org/apache/druid/segment/realtime/plumber/RealtimePlumberSchoolTest.java
index a3a9b5b..4052cc6 100644
--- a/server/src/test/java/org/apache/druid/segment/realtime/plumber/RealtimePlumberSchoolTest.java
+++ b/server/src/test/java/org/apache/druid/segment/realtime/plumber/RealtimePlumberSchoolTest.java
@@ -24,7 +24,6 @@ import com.google.common.base.Suppliers;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.io.Files;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.io.FileUtils;
import org.apache.druid.client.cache.CachePopulatorStats;
import org.apache.druid.client.cache.MapCache;
@@ -38,6 +37,7 @@ import org.apache.druid.data.input.impl.TimestampSpec;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.query.DefaultQueryRunnerFactoryConglomerate;
@@ -79,6 +79,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
/**
+ *
*/
@RunWith(Parameterized.class)
public class RealtimePlumberSchoolTest
@@ -115,7 +116,10 @@ public class RealtimePlumberSchoolTest
private FireDepartmentMetrics metrics;
private File tmpDir;
- public RealtimePlumberSchoolTest(RejectionPolicyFactory rejectionPolicy, SegmentWriteOutMediumFactory segmentWriteOutMediumFactory)
+ public RealtimePlumberSchoolTest(
+ RejectionPolicyFactory rejectionPolicy,
+ SegmentWriteOutMediumFactory segmentWriteOutMediumFactory
+ )
{
this.rejectionPolicy = rejectionPolicy;
this.segmentWriteOutMediumFactory = segmentWriteOutMediumFactory;
@@ -219,7 +223,7 @@ public class RealtimePlumberSchoolTest
announcer,
segmentPublisher,
handoffNotifierFactory,
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
TestHelper.getTestIndexMergerV9(segmentWriteOutMediumFactory),
TestHelper.getTestIndexIO(),
MapCache.create(0),
diff --git a/server/src/test/java/org/apache/druid/server/coordination/coordination/BatchDataSegmentAnnouncerTest.java b/server/src/test/java/org/apache/druid/server/coordination/coordination/BatchDataSegmentAnnouncerTest.java
index e52c278..52006f5 100644
--- a/server/src/test/java/org/apache/druid/server/coordination/coordination/BatchDataSegmentAnnouncerTest.java
+++ b/server/src/test/java/org/apache/druid/server/coordination/coordination/BatchDataSegmentAnnouncerTest.java
@@ -27,7 +27,6 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
@@ -35,6 +34,7 @@ import org.apache.curator.test.TestingCluster;
import org.apache.druid.curator.PotentiallyGzippedCompressionProvider;
import org.apache.druid.curator.announcement.Announcer;
import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.segment.TestHelper;
import org.apache.druid.server.coordination.BatchDataSegmentAnnouncer;
import org.apache.druid.server.coordination.ChangeRequestHistory;
@@ -58,6 +58,7 @@ import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
/**
+ *
*/
public class BatchDataSegmentAnnouncerTest
{
@@ -97,7 +98,7 @@ public class BatchDataSegmentAnnouncerTest
announcer = new Announcer(
cf,
- MoreExecutors.sameThreadExecutor()
+ Execs.directExecutor()
);
announcer.start();
diff --git a/services/src/main/java/org/apache/druid/cli/DumpSegment.java b/services/src/main/java/org/apache/druid/cli/DumpSegment.java
index 66ebd54..e1e2307 100644
--- a/services/src/main/java/org/apache/druid/cli/DumpSegment.java
+++ b/services/src/main/java/org/apache/druid/cli/DumpSegment.java
@@ -27,7 +27,6 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.MoreExecutors;
import com.google.inject.Binder;
import com.google.inject.Injector;
import com.google.inject.Key;
@@ -48,6 +47,7 @@ import org.apache.druid.guice.annotations.Json;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.guava.Sequence;
import org.apache.druid.java.util.common.guava.Sequences;
@@ -485,7 +485,7 @@ public class DumpSegment extends GuiceRunnable
final QueryRunner<T> runner = factory.createRunner(new QueryableIndexSegment("segment", index));
return factory
.getToolchest()
- .mergeResults(factory.mergeRunners(MoreExecutors.sameThreadExecutor(), ImmutableList.of(runner)))
+ .mergeResults(factory.mergeRunners(Execs.directExecutor(), ImmutableList.of(runner)))
.run(QueryPlus.wrap(query), new HashMap<>());
}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/schema/DruidSchema.java b/sql/src/main/java/org/apache/druid/sql/calcite/schema/DruidSchema.java
index 87d97ef..4e1bb09 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/schema/DruidSchema.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/schema/DruidSchema.java
@@ -25,7 +25,6 @@ import com.google.common.collect.ImmutableMultimap;
import com.google.common.collect.Iterables;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.MoreExecutors;
import com.google.inject.Inject;
import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.impl.AbstractSchema;
@@ -34,6 +33,7 @@ import org.apache.druid.client.TimelineServerView;
import org.apache.druid.guice.ManageLifecycle;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.concurrent.ScheduledExecutors;
import org.apache.druid.java.util.common.guava.Sequence;
import org.apache.druid.java.util.common.guava.Yielder;
@@ -141,7 +141,7 @@ public class DruidSchema extends AbstractSchema
this.escalator = escalator;
serverView.registerTimelineCallback(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
new TimelineServerView.TimelineCallback()
{
@Override
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/util/SpecificSegmentsQuerySegmentWalker.java b/sql/src/test/java/org/apache/druid/sql/calcite/util/SpecificSegmentsQuerySegmentWalker.java
index 55fff60..0840803 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/util/SpecificSegmentsQuerySegmentWalker.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/util/SpecificSegmentsQuerySegmentWalker.java
@@ -23,9 +23,9 @@ import com.google.common.base.Function;
import com.google.common.collect.Iterables;
import com.google.common.collect.Ordering;
import com.google.common.io.Closeables;
-import com.google.common.util.concurrent.MoreExecutors;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.UOE;
+import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.guava.FunctionalIterable;
import org.apache.druid.java.util.common.guava.Sequence;
import org.apache.druid.query.FinalizeResultsQueryRunner;
@@ -224,7 +224,7 @@ public class SpecificSegmentsQuerySegmentWalker implements QuerySegmentWalker, C
return new FinalizeResultsQueryRunner<>(
toolChest.mergeResults(
factory.mergeRunners(
- MoreExecutors.sameThreadExecutor(),
+ Execs.directExecutor(),
FunctionalIterable
.create(specs)
.transformCat(
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org