This is an automated email from the ASF dual-hosted git repository.
gianm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new ce9886ae1c8 fix: Time-ordered scan correctness and limit performance.
(#20655)
ce9886ae1c8 is described below
commit ce9886ae1c8dc5fe97c03405cbd5c6532a2bec0c
Author: Gian Merlino <[email protected]>
AuthorDate: Mon Oct 26 21:53:21 2026 -0700
fix: Time-ordered scan correctness and limit performance. (#20655)
Three changes related to how Scan queries schedule segments
for processing:
1) ScanQueryRunnerFactory#stableLimitingSort (the priority queue
strategy that runs when ordering by time with a low limit) now
skips segments that cannot possibly contribute to the results.
There was previously some logic partially doing this, but it
scanned time chunks beyond what was really necessary.
2) ScanQueryRunnerFactory#stableLimitingSort now truncates the
interval of segments being scanned, such that we avoid gathering
rows from within a segment when those rows cannot possibly
contribute to the result.
3) ServerManager now builds one segment runner per descriptor in
descriptor order when calling QueryRunnerFactory#mergeRunners.
This enables ScanQueryRunnerFactory to correctly match up segments
and intervals. Previously this could cause out-of-order results
with VSF on-demand loading.
---
.../druid/benchmark/query/ScanBenchmark.java | 85 +++--
.../query/ServerManagerForQueryErrorTest.java | 259 +++++++------
.../druid/collections/StableLimitingSorter.java | 12 +
.../main/java/org/apache/druid/query/Queries.java | 44 ++-
.../druid/query/scan/ScanQueryRunnerFactory.java | 189 +++++-----
.../query/spec/SpecificSegmentQueryRunner.java | 9 +-
.../collections/StableLimitingSorterTest.java | 28 ++
.../java/org/apache/druid/query/QueriesTest.java | 68 ++++
.../query/scan/ScanQueryRunnerFactoryTest.java | 403 ++++++++++++++++++++-
.../query/spec/SpecificSegmentQueryRunnerTest.java | 52 +++
.../apache/druid/client/CachingQueryRunner.java | 11 +-
.../org/apache/druid/server/ServerManager.java | 68 ++--
.../druid/client/CachingQueryRunnerTest.java | 50 ++-
.../ServerManagerScanOrderingTest.java | 357 ++++++++++++++++++
14 files changed, 1352 insertions(+), 283 deletions(-)
diff --git
a/benchmarks/src/test/java/org/apache/druid/benchmark/query/ScanBenchmark.java
b/benchmarks/src/test/java/org/apache/druid/benchmark/query/ScanBenchmark.java
index 2662071188d..ef6c17fb29c 100644
---
a/benchmarks/src/test/java/org/apache/druid/benchmark/query/ScanBenchmark.java
+++
b/benchmarks/src/test/java/org/apache/druid/benchmark/query/ScanBenchmark.java
@@ -58,6 +58,8 @@ import org.apache.druid.query.scan.ScanResultValue;
import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
import org.apache.druid.query.spec.MultipleSpecificSegmentSpec;
import org.apache.druid.query.spec.QuerySegmentSpec;
+import org.apache.druid.query.spec.SpecificSegmentQueryRunner;
+import org.apache.druid.query.spec.SpecificSegmentSpec;
import org.apache.druid.segment.IncrementalIndexSegment;
import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.IndexMergerV9;
@@ -75,6 +77,7 @@ import
org.apache.druid.segment.incremental.OnheapIncrementalIndex;
import org.apache.druid.segment.serde.ComplexMetrics;
import
org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
import org.apache.druid.timeline.SegmentId;
+import org.joda.time.Interval;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
@@ -112,7 +115,7 @@ public class ScanBenchmark
@Param({"basic.A", "basic.B", "basic.C", "basic.D"})
private String schemaAndQuery;
- @Param({"1000", "99999"})
+ @Param({"10", "1000", "99999"})
private int limit;
@Param({"NONE", "DESCENDING", "ASCENDING"})
@@ -301,8 +304,17 @@ public class ScanBenchmark
@State(Scope.Benchmark)
public static class QueryableIndexState
{
+ /**
+ * Number of time chunks that evenly divide the data interval of the
schema.
+ */
+ @Param({"1", "8"})
+ private int numTimeChunks;
+
+ /**
+ * Number of segments per time chunk. Each one spans its whole time chunk.
+ */
@Param({"2", "4"})
- private int numSegments;
+ private int numPartitionsPerTimeChunk;
@Param({"2"})
private int numProcessingThreads;
@@ -311,6 +323,11 @@ public class ScanBenchmark
private File qIndexesDir;
private List<QueryableIndex> qIndexes;
+ /**
+ * IDs of {@link #qIndexes}, in time order.
+ */
+ private List<SegmentId> segmentIds;
+
@Setup
public void setup(ScanBenchmark global) throws IOException
{
@@ -320,22 +337,39 @@ public class ScanBenchmark
qIndexesDir = FileUtils.createTempDir();
qIndexes = new ArrayList<>();
+ segmentIds = new ArrayList<>();
- for (int i = 0; i < numSegments; i++) {
- log.info("Generating rows for segment " + i);
-
- IncrementalIndex incIndex = global.makeIncIndex();
- global.generator.reset(RNG_SEED + i).addToIndex(incIndex,
global.rowsPerSegment);
-
- File indexFile = INDEX_MERGER_V9.persist(
- incIndex,
- new File(qIndexesDir, String.valueOf(i)),
- IndexSpec.getDefault(),
- null
+ final Interval dataInterval = global.schemaInfo.getDataInterval();
+ for (int chunk = 0; chunk < numTimeChunks; chunk++) {
+ final Interval chunkInterval = Intervals.utc(
+ dataInterval.getStartMillis() + dataInterval.toDurationMillis() *
chunk / numTimeChunks,
+ dataInterval.getStartMillis() + dataInterval.toDurationMillis() *
(chunk + 1) / numTimeChunks
);
- incIndex.close();
- qIndexes.add(INDEX_IO.loadIndex(indexFile));
+ for (int partition = 0; partition < numPartitionsPerTimeChunk;
partition++) {
+ final int i = qIndexes.size();
+ log.info("Generating rows for segment " + i);
+
+ final DataGenerator generator = new DataGenerator(
+ global.schemaInfo.getColumnSchemas(),
+ RNG_SEED + i,
+ chunkInterval,
+ global.rowsPerSegment
+ );
+ IncrementalIndex incIndex = global.makeIncIndex();
+ generator.addToIndex(incIndex, global.rowsPerSegment);
+
+ File indexFile = INDEX_MERGER_V9.persist(
+ incIndex,
+ new File(qIndexesDir, String.valueOf(i)),
+ IndexSpec.getDefault(),
+ null
+ );
+ incIndex.close();
+
+ qIndexes.add(INDEX_IO.loadIndex(indexFile));
+ segmentIds.add(SegmentId.of("qIndex", chunkInterval,
"dummy_version", partition));
+ }
}
}
@@ -441,21 +475,22 @@ public class ScanBenchmark
List<SegmentDescriptor> segmentDescriptors = new ArrayList<>();
List<QueryRunner<Row>> runners = new ArrayList<>();
QueryToolChest toolChest = factory.getToolchest();
- for (int i = 0; i < state.numSegments; i++) {
- String segmentName = "qIndex";
+ for (int i = 0; i < state.qIndexes.size(); i++) {
+ final SegmentId segmentId = state.segmentIds.get(i);
final QueryRunner<Result<ScanResultValue>> runner =
QueryBenchmarkUtil.makeQueryRunner(
factory,
- SegmentId.dummy(segmentName),
- new QueryableIndexSegment(state.qIndexes.get(i),
SegmentId.dummy(segmentName, i))
+ segmentId,
+ new QueryableIndexSegment(state.qIndexes.get(i), segmentId)
);
- segmentDescriptors.add(
- new SegmentDescriptor(
- Intervals.ETERNITY,
- "dummy_version",
- i
+ segmentDescriptors.add(segmentId.toDescriptor());
+
+ // Restrict each segment to its descriptor, as data servers do.
+ runners.add(
+ new SpecificSegmentQueryRunner<>(
+ toolChest.preMergeQueryDecoration(runner),
+ new SpecificSegmentSpec(segmentId.toDescriptor())
)
);
- runners.add(toolChest.preMergeQueryDecoration(runner));
}
QueryRunner theRunner = toolChest.postMergeQueryDecoration(
diff --git
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/ServerManagerForQueryErrorTest.java
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/ServerManagerForQueryErrorTest.java
index 847ed8e8933..0d8f4f8105a 100644
---
a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/ServerManagerForQueryErrorTest.java
+++
b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/query/ServerManagerForQueryErrorTest.java
@@ -26,10 +26,8 @@ import org.apache.commons.lang3.mutable.MutableBoolean;
import org.apache.druid.client.cache.Cache;
import org.apache.druid.client.cache.CacheConfig;
import org.apache.druid.client.cache.CachePopulator;
-import org.apache.druid.error.DruidException;
import org.apache.druid.guice.annotations.Smile;
import org.apache.druid.java.util.common.guava.Accumulator;
-import org.apache.druid.java.util.common.guava.FunctionalIterable;
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.guava.Yielder;
@@ -246,145 +244,138 @@ public class ServerManagerForQueryErrorTest extends
ServerManager
}
@Override
- protected <T> FunctionalIterable<QueryRunner<T>> getQueryRunnersForSegments(
- Query<T> query,
- QueryRunnerFactory<T, Query<T>> factory,
- QueryToolChest<T, Query<T>> toolChest,
- List<SegmentReference> segmentReferences,
- AtomicLong cpuTimeAccumulator,
- Optional<byte[]> cacheKeyPrefix
+ protected <T> QueryRunner<T> buildQueryRunnerForSegment(
+ final SegmentDescriptor segmentDescriptor,
+ final Segment segment,
+ final QueryRunnerFactory<T, Query<T>> factory,
+ final QueryToolChest<T, Query<T>> toolChest,
+ final AtomicLong cpuTimeAccumulator,
+ final Optional<byte[]> cacheKeyPrefix
)
{
- return FunctionalIterable
- .create(segmentReferences)
- .transform(
- ref ->
- ref.getSegmentReference()
- .map(segment -> {
- final QueryContext queryContext = query.context();
- if
(queryContext.getBoolean(QUERY_TIMEOUT_TEST_CONTEXT_KEY, false)) {
- return (QueryRunner<T>) (queryPlus, responseContext) ->
new Sequence<>()
- {
- @Override
- public <OutType> OutType accumulate(
- OutType initValue,
- Accumulator<OutType, T> accumulator
- )
- {
- throw new QueryTimeoutException("query timeout
test");
- }
+ final QueryRunner<T> baseRunner = super.buildQueryRunnerForSegment(
+ segmentDescriptor,
+ segment,
+ factory,
+ toolChest,
+ cpuTimeAccumulator,
+ cacheKeyPrefix
+ );
+
+ return (queryPlus, responseContext) -> {
+ final QueryContext queryContext = queryPlus.getQuery().context();
+ if (queryContext.getBoolean(QUERY_TIMEOUT_TEST_CONTEXT_KEY, false)) {
+ return new Sequence<>()
+ {
+ @Override
+ public <OutType> OutType accumulate(
+ OutType initValue,
+ Accumulator<OutType, T> accumulator
+ )
+ {
+ throw new QueryTimeoutException("query timeout test");
+ }
- @Override
- public <OutType> Yielder<OutType> toYielder(
- OutType initValue,
- YieldingAccumulator<OutType, T> accumulator
- )
- {
- throw new QueryTimeoutException("query timeout
test");
- }
- };
- } else if
(queryContext.getBoolean(QUERY_CAPACITY_EXCEEDED_TEST_CONTEXT_KEY, false)) {
- return (QueryRunner<T>) (queryPlus, responseContext) ->
new Sequence<>()
- {
- @Override
- public <OutType> OutType accumulate(
- OutType initValue,
- Accumulator<OutType, T> accumulator
- )
- {
- throw
QueryCapacityExceededException.withErrorMessageAndResolvedHost(
- "query capacity exceeded test"
- );
- }
+ @Override
+ public <OutType> Yielder<OutType> toYielder(
+ OutType initValue,
+ YieldingAccumulator<OutType, T> accumulator
+ )
+ {
+ throw new QueryTimeoutException("query timeout test");
+ }
+ };
+ } else if
(queryContext.getBoolean(QUERY_CAPACITY_EXCEEDED_TEST_CONTEXT_KEY, false)) {
+ return new Sequence<>()
+ {
+ @Override
+ public <OutType> OutType accumulate(
+ OutType initValue,
+ Accumulator<OutType, T> accumulator
+ )
+ {
+ throw
QueryCapacityExceededException.withErrorMessageAndResolvedHost(
+ "query capacity exceeded test"
+ );
+ }
- @Override
- public <OutType> Yielder<OutType> toYielder(
- OutType initValue,
- YieldingAccumulator<OutType, T> accumulator
- )
- {
- throw
QueryCapacityExceededException.withErrorMessageAndResolvedHost(
- "query capacity exceeded test"
- );
- }
- };
- } else if
(queryContext.getBoolean(QUERY_UNSUPPORTED_TEST_CONTEXT_KEY, false)) {
- return (QueryRunner<T>) (queryPlus, responseContext) ->
new Sequence<>()
- {
- @Override
- public <OutType> OutType accumulate(
- OutType initValue,
- Accumulator<OutType, T> accumulator
- )
- {
- throw new QueryUnsupportedException("query
unsupported test");
- }
+ @Override
+ public <OutType> Yielder<OutType> toYielder(
+ OutType initValue,
+ YieldingAccumulator<OutType, T> accumulator
+ )
+ {
+ throw
QueryCapacityExceededException.withErrorMessageAndResolvedHost(
+ "query capacity exceeded test"
+ );
+ }
+ };
+ } else if (queryContext.getBoolean(QUERY_UNSUPPORTED_TEST_CONTEXT_KEY,
false)) {
+ return new Sequence<>()
+ {
+ @Override
+ public <OutType> OutType accumulate(
+ OutType initValue,
+ Accumulator<OutType, T> accumulator
+ )
+ {
+ throw new QueryUnsupportedException("query unsupported test");
+ }
- @Override
- public <OutType> Yielder<OutType> toYielder(
- OutType initValue,
- YieldingAccumulator<OutType, T> accumulator
- )
- {
- throw new QueryUnsupportedException("query
unsupported test");
- }
- };
- } else if
(queryContext.getBoolean(RESOURCE_LIMIT_EXCEEDED_TEST_CONTEXT_KEY, false)) {
- return (QueryRunner<T>) (queryPlus, responseContext) ->
new Sequence<>()
- {
- @Override
- public <OutType> OutType accumulate(
- OutType initValue,
- Accumulator<OutType, T> accumulator
- )
- {
- throw new ResourceLimitExceededException("resource
limit exceeded test");
- }
+ @Override
+ public <OutType> Yielder<OutType> toYielder(
+ OutType initValue,
+ YieldingAccumulator<OutType, T> accumulator
+ )
+ {
+ throw new QueryUnsupportedException("query unsupported test");
+ }
+ };
+ } else if
(queryContext.getBoolean(RESOURCE_LIMIT_EXCEEDED_TEST_CONTEXT_KEY, false)) {
+ return new Sequence<>()
+ {
+ @Override
+ public <OutType> OutType accumulate(
+ OutType initValue,
+ Accumulator<OutType, T> accumulator
+ )
+ {
+ throw new ResourceLimitExceededException("resource limit exceeded
test");
+ }
- @Override
- public <OutType> Yielder<OutType> toYielder(
- OutType initValue,
- YieldingAccumulator<OutType, T> accumulator
- )
- {
- throw new ResourceLimitExceededException("resource
limit exceeded test");
- }
- };
- } else if
(queryContext.getBoolean(QUERY_FAILURE_TEST_CONTEXT_KEY, false)) {
- return (QueryRunner<T>) (queryPlus, responseContext) ->
new Sequence<>()
- {
- @Override
- public <OutType> OutType accumulate(
- OutType initValue,
- Accumulator<OutType, T> accumulator
- )
- {
- throw new RuntimeException("query failure test");
- }
+ @Override
+ public <OutType> Yielder<OutType> toYielder(
+ OutType initValue,
+ YieldingAccumulator<OutType, T> accumulator
+ )
+ {
+ throw new ResourceLimitExceededException("resource limit exceeded
test");
+ }
+ };
+ } else if (queryContext.getBoolean(QUERY_FAILURE_TEST_CONTEXT_KEY,
false)) {
+ return new Sequence<>()
+ {
+ @Override
+ public <OutType> OutType accumulate(
+ OutType initValue,
+ Accumulator<OutType, T> accumulator
+ )
+ {
+ throw new RuntimeException("query failure test");
+ }
- @Override
- public <OutType> Yielder<OutType> toYielder(
- OutType initValue,
- YieldingAccumulator<OutType, T> accumulator
- )
- {
- throw new RuntimeException("query failure test");
- }
- };
- }
+ @Override
+ public <OutType> Yielder<OutType> toYielder(
+ OutType initValue,
+ YieldingAccumulator<OutType, T> accumulator
+ )
+ {
+ throw new RuntimeException("query failure test");
+ }
+ };
+ }
- return buildQueryRunnerForSegment(
- ref.getSegmentDescriptor(),
- segment,
- factory,
- toolChest,
- cpuTimeAccumulator,
- cacheKeyPrefix
- );
- })
- .orElseThrow(
- () -> DruidException.defensive("Unexpected missing
segment[%s]", ref.getSegmentDescriptor())
- )
- );
+ return baseRunner.run(queryPlus, responseContext);
+ };
}
}
diff --git
a/processing/src/main/java/org/apache/druid/collections/StableLimitingSorter.java
b/processing/src/main/java/org/apache/druid/collections/StableLimitingSorter.java
index d0c53781ca7..6ce8313a1ac 100644
---
a/processing/src/main/java/org/apache/druid/collections/StableLimitingSorter.java
+++
b/processing/src/main/java/org/apache/druid/collections/StableLimitingSorter.java
@@ -24,6 +24,7 @@ import com.google.common.collect.MinMaxPriorityQueue;
import com.google.common.collect.Ordering;
import com.google.common.primitives.Ints;
+import javax.annotation.Nullable;
import java.util.Comparator;
import java.util.Iterator;
import java.util.Objects;
@@ -75,6 +76,17 @@ public class StableLimitingSorter<T>
return Ints.checkedCast(Math.min(size, limit));
}
+ /**
+ * Returns the greatest element currently retained, or null if there are
none. Once the sorter is full, an element
+ * offered to {@link #add} is retained only if it sorts strictly before this
one.
+ */
+ @Nullable
+ public T peekLast()
+ {
+ final NumberedElement<T> last = queue.peekLast();
+ return last == null ? null : last.getElement();
+ }
+
/**
* Drain elements in sorted order (least first).
*/
diff --git a/processing/src/main/java/org/apache/druid/query/Queries.java
b/processing/src/main/java/org/apache/druid/query/Queries.java
index 842a047edbe..51d75cb8e79 100644
--- a/processing/src/main/java/org/apache/druid/query/Queries.java
+++ b/processing/src/main/java/org/apache/druid/query/Queries.java
@@ -33,6 +33,7 @@ import
org.apache.druid.query.spec.MultipleSpecificSegmentSpec;
import org.apache.druid.segment.VirtualColumn;
import org.apache.druid.segment.VirtualColumns;
import org.apache.druid.segment.column.ColumnHolder;
+import org.joda.time.Interval;
import javax.annotation.Nullable;
@@ -178,11 +179,52 @@ public class Queries
return retVal;
}
+ /**
+ * Returns a {@link SegmentDescriptor} with its interval narrowed to the
intervals of the lowest-level
+ * subquery of the provided {@link Query} (the same one that {@link
#withSpecificSegments(Query, List)} modifies),
+ * or null if they do not overlap.
+ *
+ * <p>Data servers build each descriptor from the query's original
intervals, so this normally returns
+ * {@code descriptor} unchanged. It differs only when a query engine
rewrites its query to narrow the
+ * interval further.
+ */
+ @Nullable
+ public static SegmentDescriptor narrowToQuery(final SegmentDescriptor
descriptor, final Query<?> query)
+ {
+ if (query.getDataSource() instanceof QueryDataSource) {
+ return narrowToQuery(descriptor, ((QueryDataSource)
query.getDataSource()).getQuery());
+ }
+
+ // Query intervals are condensed, so each one overlaps a different,
non-abutting part of the descriptor.
+ Interval narrowedInterval = null;
+ for (final Interval queryInterval : query.getIntervals()) {
+ final Interval overlap = descriptor.getInterval().overlap(queryInterval);
+ if (overlap != null && overlap.toDurationMillis() > 0) {
+ if (narrowedInterval != null) {
+ throw DruidException.defensive(
+ "Query intervals[%s] split segment interval[%s]",
+ query.getIntervals(),
+ descriptor.getInterval()
+ );
+ }
+ narrowedInterval = overlap;
+ }
+ }
+
+ if (narrowedInterval == null) {
+ return null;
+ } else if (narrowedInterval.equals(descriptor.getInterval())) {
+ return descriptor;
+ } else {
+ return new SegmentDescriptor(narrowedInterval, descriptor.getVersion(),
descriptor.getPartitionNumber());
+ }
+ }
+
/**
* Rewrite "query" to refer to some specific base datasource, instead of the
one it currently refers to.
*
* Unlike the seemingly-similar {@link Query#withDataSource}, this will walk
down the datasource tree and replace
- * only the base datasource (in the sense defined in {@link
DataSourceAnalysis}).
+ * only the base datasource (in the sense defined in {@link
ExecutionVertex#getBaseDataSource()}).
*/
public static Query withBaseDataSource(final Query query, final DataSource
newBaseDataSource)
{
diff --git
a/processing/src/main/java/org/apache/druid/query/scan/ScanQueryRunnerFactory.java
b/processing/src/main/java/org/apache/druid/query/scan/ScanQueryRunnerFactory.java
index f09b42c2795..466f081e39a 100644
---
a/processing/src/main/java/org/apache/druid/query/scan/ScanQueryRunnerFactory.java
+++
b/processing/src/main/java/org/apache/druid/query/scan/ScanQueryRunnerFactory.java
@@ -24,14 +24,14 @@ import com.google.common.collect.Iterators;
import com.google.common.collect.Lists;
import com.google.inject.Inject;
import org.apache.druid.collections.StableLimitingSorter;
+import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.JodaUtils;
import org.apache.druid.java.util.common.Pair;
import org.apache.druid.java.util.common.UOE;
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.guava.Yielder;
-import org.apache.druid.java.util.common.guava.Yielders;
import org.apache.druid.query.Order;
import org.apache.druid.query.Query;
import org.apache.druid.query.QueryPlus;
@@ -43,13 +43,13 @@ import
org.apache.druid.query.ResourceLimitExceededException;
import org.apache.druid.query.SegmentDescriptor;
import org.apache.druid.query.SinkQueryRunners;
import org.apache.druid.query.context.ResponseContext;
+import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
import org.apache.druid.query.spec.MultipleSpecificSegmentSpec;
import org.apache.druid.query.spec.QuerySegmentSpec;
import org.apache.druid.query.spec.SpecificSegmentSpec;
import org.apache.druid.segment.Segment;
import org.joda.time.Interval;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
@@ -111,46 +111,16 @@ public class ScanQueryRunnerFactory implements
QueryRunnerFactory<ScanResultValu
return returnedRows;
}
} else {
- List<Interval> intervalsOrdered =
getIntervalsFromSpecificQuerySpec(query.getQuerySegmentSpec());
- List<QueryRunner<ScanResultValue>> queryRunnersOrdered =
Lists.newArrayList(queryRunners);
-
- if (Order.DESCENDING.equals(query.getTimeOrder())) {
- intervalsOrdered = Lists.reverse(intervalsOrdered);
- queryRunnersOrdered = Lists.reverse(queryRunnersOrdered);
- }
+ final List<Pair<Interval, QueryRunner<ScanResultValue>>>
intervalsAndRunnersOrdered =
+ getIntervalsAndRunnersOrdered(query, queryRunners);
int maxRowsQueuedForOrdering = (query.getMaxRowsQueuedForOrdering() ==
null
?
scanQueryConfig.getMaxRowsQueuedForOrdering()
: query.getMaxRowsQueuedForOrdering());
if (query.getScanRowsLimit() <= maxRowsQueuedForOrdering) {
// Use priority queue strategy
- try {
- return stableLimitingSort(
- Sequences.concat(Sequences.map(
- Sequences.simple(queryRunnersOrdered),
- input -> input.run(queryPlus, responseContext)
- )),
- query,
- intervalsOrdered
- );
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
+ return stableLimitingSort(intervalsAndRunnersOrdered, queryPlus,
responseContext);
} else {
// Use n-way merge strategy
- List<Pair<Interval, QueryRunner<ScanResultValue>>>
intervalsAndRunnersOrdered = new ArrayList<>();
- if (intervalsOrdered.size() == queryRunnersOrdered.size()) {
- for (int i = 0; i < queryRunnersOrdered.size(); i++) {
- intervalsAndRunnersOrdered.add(new
Pair<>(intervalsOrdered.get(i), queryRunnersOrdered.get(i)));
- }
- } else if (queryRunners instanceof SinkQueryRunners) {
- ((SinkQueryRunners<ScanResultValue>)
queryRunners).runnerIntervalMappingIterator()
-
.forEachRemaining(intervalsAndRunnersOrdered::add);
- } else {
- throw new ISE("Number of segment descriptors does not equal number
of "
- + "query runners...something went wrong!");
- }
-
// Group the list of pairs by interval. The LinkedHashMap will have
an interval paired with a list of all the
// query runners for that segment
LinkedHashMap<Interval, List<Pair<Interval,
QueryRunner<ScanResultValue>>>> partitionsGroupedByInterval =
@@ -180,12 +150,12 @@ public class ScanQueryRunnerFactory implements
QueryRunnerFactory<ScanResultValu
// there should be no interval overlap. We create a list of lists
so we can create a sequence of sequences.
// There's no easy way to convert a LinkedHashMap to a sequence
because it's non-iterable.
List<List<QueryRunner<ScanResultValue>>> groupedRunners =
- partitionsGroupedByInterval.entrySet()
+ partitionsGroupedByInterval.values()
.stream()
- .map(entry -> entry.getValue()
- .stream()
-
.map(segQueryRunnerPair -> segQueryRunnerPair.rhs)
-
.collect(Collectors.toList()))
+ .map(pairs -> pairs
+ .stream()
+ .map(segQueryRunnerPair ->
segQueryRunnerPair.rhs)
+ .collect(Collectors.toList()))
.collect(Collectors.toList());
return nWayMergeAndLimit(groupedRunners, queryPlus,
responseContext);
@@ -206,17 +176,18 @@ public class ScanQueryRunnerFactory implements
QueryRunnerFactory<ScanResultValu
}
/**
- * Returns a sorted and limited copy of the provided {@param inputSequence}.
Materializes the full sequence
- * in memory before returning it. The amount of memory use is limited by the
limit of the {@param scanQuery}.
+ * Returns the first rows, up to the limit of the query, from the provided
runners in the order given by
+ * {@link ScanQuery#getResultOrdering()}. Runs the runners one at a time, in
the provided order, and materializes the
+ * result in memory. Skips runners that could not possibly contribute to the
results.
*/
@VisibleForTesting
Sequence<ScanResultValue> stableLimitingSort(
- Sequence<ScanResultValue> inputSequence,
- ScanQuery scanQuery,
- List<Interval> intervalsOrdered
- ) throws IOException
+ final List<Pair<Interval, QueryRunner<ScanResultValue>>>
intervalsAndRunnersOrdered,
+ final QueryPlus<ScanResultValue> queryPlus,
+ final ResponseContext responseContext
+ )
{
- Comparator<ScanResultValue> comparator = scanQuery.getResultOrdering();
+ final ScanQuery scanQuery = (ScanQuery) queryPlus.getQuery();
if (scanQuery.getScanRowsLimit() > Integer.MAX_VALUE) {
throw new UOE(
@@ -227,52 +198,92 @@ public class ScanQueryRunnerFactory implements
QueryRunnerFactory<ScanResultValu
// Converting the limit from long to int could theoretically throw an
ArithmeticException but this branch
// only runs if limit < MAX_LIMIT_FOR_IN_MEMORY_TIME_ORDERING (which
should be < Integer.MAX_VALUE)
- int limit = Math.toIntExact(scanQuery.getScanRowsLimit());
-
- final StableLimitingSorter<ScanResultValue> sorter = new
StableLimitingSorter<>(comparator, limit);
-
- Yielder<ScanResultValue> yielder = Yielders.each(inputSequence);
-
- try {
- boolean doneScanning = yielder.isDone();
- // We need to scan limit elements and anything else in the last segment
- int numRowsScanned = 0;
- Interval finalInterval = null;
- while (!doneScanning) {
- ScanResultValue next = yielder.get();
- List<ScanResultValue> singleEventScanResultValues =
next.toSingleEventScanResultValues();
- for (ScanResultValue srv : singleEventScanResultValues) {
- numRowsScanned++;
- // Using an intermediate unbatched ScanResultValue is not that great
memory-wise, but the column list
- // needs to be preserved for queries using the compactedList result
format
- sorter.add(srv);
-
- // Finish scanning the interval containing the limit row
- if (numRowsScanned > limit && finalInterval == null) {
- long timestampOfLimitRow =
srv.getFirstEventTimestamp(scanQuery.getResultFormat());
- for (Interval interval : intervalsOrdered) {
- if (interval.contains(timestampOfLimitRow)) {
- finalInterval = interval;
- }
- }
- if (finalInterval == null) {
- throw new ISE("Row came from an unscanned interval");
- }
- }
+ final int limit = Math.toIntExact(scanQuery.getScanRowsLimit());
+
+ final StableLimitingSorter<ScanResultValue> sorter =
+ new StableLimitingSorter<>(scanQuery.getResultOrdering(), limit);
+
+ for (final Pair<Interval, QueryRunner<ScanResultValue>> intervalAndRunner
: intervalsAndRunnersOrdered) {
+ QueryPlus<ScanResultValue> runnerQueryPlus = queryPlus;
+
+ if (sorter.size() == limit) {
+ final Interval includableInterval =
computeIncludableInterval(scanQuery, sorter.peekLast());
+ if (!intervalAndRunner.lhs.overlaps(includableInterval)) {
+ continue;
}
- yielder = yielder.next(null);
- doneScanning = yielder.isDone() ||
- (finalInterval != null &&
-
!finalInterval.contains(next.getFirstEventTimestamp(scanQuery.getResultFormat())));
+
+ // Narrow to the part of this runner's interval that could still be
included, so its cursor can skip the rest.
+ // The overlap is non-null, since we just checked that the intervals
overlap.
+ final Interval narrowedInterval =
intervalAndRunner.lhs.overlap(includableInterval);
+ runnerQueryPlus = queryPlus.withQuery(
+ scanQuery.withQuerySegmentSpec(new
MultipleIntervalSegmentSpec(List.of(narrowedInterval)))
+ );
}
- final List<ScanResultValue> sortedElements = new
ArrayList<>(sorter.size());
- Iterators.addAll(sortedElements, sorter.drain());
- return Sequences.simple(sortedElements);
+ intervalAndRunner.rhs.run(runnerQueryPlus, responseContext).accumulate(
+ sorter,
+ (theSorter, batch) -> {
+ for (final ScanResultValue srv :
batch.toSingleEventScanResultValues()) {
+ // Using an intermediate unbatched ScanResultValue is not that
great memory-wise, but the column list
+ // needs to be preserved for queries using the compactedList
result format
+ theSorter.add(srv);
+ }
+ return theSorter;
+ }
+ );
+ }
+
+ final List<ScanResultValue> sortedElements = new
ArrayList<>(sorter.size());
+ Iterators.addAll(sortedElements, sorter.drain());
+ return Sequences.simple(sortedElements);
+ }
+
+ /**
+ * Returns the timestamps that a row must have to be included in the result,
given that the result is full and its
+ * current last row is {@code lastRow}. Includes the timestamp of {@code
lastRow}, since rows with equal timestamps
+ * are ordered by segment ID, so a later row with that timestamp may still
be included.
+ */
+ private static Interval computeIncludableInterval(final ScanQuery query,
final ScanResultValue lastRow)
+ {
+ final long lastTimestamp =
lastRow.getFirstEventTimestamp(query.getResultFormat());
+ if (Order.DESCENDING.equals(query.getTimeOrder())) {
+ return Intervals.utc(lastTimestamp, JodaUtils.MAX_INSTANT);
+ } else {
+ return Intervals.utc(JodaUtils.MIN_INSTANT, lastTimestamp + 1);
}
- finally {
- yielder.close();
+ }
+
+ /**
+ * Returns each runner paired with the interval of the segment it reads, in
the order given by
+ * {@link ScanQuery#getTimeOrder()}.
+ */
+ private List<Pair<Interval, QueryRunner<ScanResultValue>>>
getIntervalsAndRunnersOrdered(
+ final ScanQuery query,
+ final Iterable<QueryRunner<ScanResultValue>> queryRunners
+ )
+ {
+ final List<Pair<Interval, QueryRunner<ScanResultValue>>>
intervalsAndRunners = new ArrayList<>();
+
+ if (queryRunners instanceof SinkQueryRunners) {
+ ((SinkQueryRunners<ScanResultValue>)
queryRunners).runnerIntervalMappingIterator()
+
.forEachRemaining(intervalsAndRunners::add);
+ } else {
+ final List<Interval> intervals =
getIntervalsFromSpecificQuerySpec(query.getQuerySegmentSpec());
+ final List<QueryRunner<ScanResultValue>> runners =
Lists.newArrayList(queryRunners);
+ if (intervals.size() != runners.size()) {
+ throw DruidException.defensive(
+ "Number of segment descriptors[%d] does not equal number of query
runners[%d]",
+ intervals.size(),
+ runners.size()
+ );
+ }
+ for (int i = 0; i < runners.size(); i++) {
+ intervalsAndRunners.add(Pair.of(intervals.get(i), runners.get(i)));
+ }
}
+
+ // Descriptors and runners are in ascending time order.
+ return Order.DESCENDING.equals(query.getTimeOrder()) ?
Lists.reverse(intervalsAndRunners) : intervalsAndRunners;
}
@VisibleForTesting
diff --git
a/processing/src/main/java/org/apache/druid/query/spec/SpecificSegmentQueryRunner.java
b/processing/src/main/java/org/apache/druid/query/spec/SpecificSegmentQueryRunner.java
index c686316a9c7..24c02ee8906 100644
---
a/processing/src/main/java/org/apache/druid/query/spec/SpecificSegmentQueryRunner.java
+++
b/processing/src/main/java/org/apache/druid/query/spec/SpecificSegmentQueryRunner.java
@@ -31,6 +31,7 @@ import org.apache.druid.query.Queries;
import org.apache.druid.query.Query;
import org.apache.druid.query.QueryPlus;
import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.SegmentDescriptor;
import org.apache.druid.query.context.QueryContextParameters;
import org.apache.druid.query.context.ResponseContext;
import org.apache.druid.segment.SegmentMissingException;
@@ -58,10 +59,16 @@ public class SpecificSegmentQueryRunner<T> implements
QueryRunner<T>
@Override
public Sequence<T> run(final QueryPlus<T> input, final ResponseContext
responseContext)
{
+ final SegmentDescriptor descriptor =
Queries.narrowToQuery(specificSpec.getDescriptor(), input.getQuery());
+ if (descriptor == null) {
+ // Nothing to read, since the query does not overlap this segment.
+ return Sequences.empty();
+ }
+
final QueryPlus<T> queryPlus = input.withQuery(
Queries.withSpecificSegments(
input.getQuery(),
- Collections.singletonList(specificSpec.getDescriptor())
+ Collections.singletonList(descriptor)
)
);
diff --git
a/processing/src/test/java/org/apache/druid/collections/StableLimitingSorterTest.java
b/processing/src/test/java/org/apache/druid/collections/StableLimitingSorterTest.java
index fe2a92fbfc8..a82f5494a35 100644
---
a/processing/src/test/java/org/apache/druid/collections/StableLimitingSorterTest.java
+++
b/processing/src/test/java/org/apache/druid/collections/StableLimitingSorterTest.java
@@ -19,9 +19,14 @@
package org.apache.druid.collections;
+import com.google.common.collect.Lists;
import nl.jqno.equalsverifier.EqualsVerifier;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.Comparator;
+import java.util.List;
+
public class StableLimitingSorterTest
{
@Test
@@ -29,4 +34,27 @@ public class StableLimitingSorterTest
{
EqualsVerifier.forClass(StableLimitingSorter.NumberedElement.class).usingGetClass().verify();
}
+
+ @Test
+ public void testPeekLast()
+ {
+ // Compare only the first character, so "b1" and "b2" are equal.
+ final StableLimitingSorter<String> sorter =
+ new StableLimitingSorter<>(Comparator.comparing(s -> s.charAt(0)), 2);
+ Assertions.assertNull(sorter.peekLast());
+
+ sorter.add("c");
+ Assertions.assertEquals("c", sorter.peekLast());
+
+ sorter.add("b1");
+ Assertions.assertEquals("c", sorter.peekLast());
+
+ sorter.add("a");
+ Assertions.assertEquals("b1", sorter.peekLast());
+
+ // Equal to the greatest element, but added later, so it is not retained.
+ sorter.add("b2");
+ Assertions.assertEquals("b1", sorter.peekLast());
+ Assertions.assertEquals(List.of("a", "b1"),
Lists.newArrayList(sorter.drain()));
+ }
}
diff --git a/processing/src/test/java/org/apache/druid/query/QueriesTest.java
b/processing/src/test/java/org/apache/druid/query/QueriesTest.java
index ce3a4cd2da1..aa1a3a84d73 100644
--- a/processing/src/test/java/org/apache/druid/query/QueriesTest.java
+++ b/processing/src/test/java/org/apache/druid/query/QueriesTest.java
@@ -35,6 +35,7 @@ import
org.apache.druid.query.aggregation.post.FieldAccessPostAggregator;
import org.apache.druid.query.context.QueryContextParameters;
import org.apache.druid.query.filter.TrueDimFilter;
import org.apache.druid.query.planning.ExecutionVertexTest;
+import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
import org.apache.druid.query.spec.MultipleSpecificSegmentSpec;
import org.apache.druid.query.timeseries.TimeseriesQuery;
import org.apache.druid.segment.join.JoinType;
@@ -337,6 +338,73 @@ public class QueriesTest
Assertions.assertEquals("Base
dataSource[LookupDataSource{lookupName='lookyloo'}] is not a table!",
e.getMessage());
}
+ @Test
+ public void testNarrowToQueryCoveringQuery()
+ {
+ final SegmentDescriptor descriptor = new
SegmentDescriptor(Intervals.of("2000/2001"), "0", 0);
+ Assertions.assertSame(descriptor, Queries.narrowToQuery(descriptor,
makeTimeseriesQuery("1999/2001", "2005/2006")));
+ }
+
+ @Test
+ public void testNarrowToQueryPartialOverlap()
+ {
+ final SegmentDescriptor descriptor = new
SegmentDescriptor(Intervals.of("2000/2001"), "0", 1);
+ Assertions.assertEquals(
+ new SegmentDescriptor(Intervals.of("2000-03/2001"), "0", 1),
+ Queries.narrowToQuery(descriptor, makeTimeseriesQuery("1998/1999",
"2000-03/2002"))
+ );
+ }
+
+ @Test
+ public void testNarrowToQueryNoOverlap()
+ {
+ final SegmentDescriptor descriptor = new
SegmentDescriptor(Intervals.of("2000/2001"), "0", 0);
+ Assertions.assertNull(Queries.narrowToQuery(descriptor,
makeTimeseriesQuery("1999/2000", "2001/2002")));
+ }
+
+ @Test
+ public void testNarrowToQuerySplitIsAnError()
+ {
+ final SegmentDescriptor descriptor = new
SegmentDescriptor(Intervals.of("2000/2001"), "0", 0);
+ final TimeseriesQuery query = makeTimeseriesQuery("1999/2000-03",
"2000-06/2002");
+
+ DruidException e = assertThrows(DruidException.class, () ->
Queries.narrowToQuery(descriptor, query));
+ Assertions.assertEquals(
+ "Query intervals[[1999-01-01T00:00:00.000Z/2000-03-01T00:00:00.000Z, "
+ + "2000-06-01T00:00:00.000Z/2002-01-01T00:00:00.000Z]] split segment
interval"
+ + "[2000-01-01T00:00:00.000Z/2001-01-01T00:00:00.000Z]",
+ e.getMessage()
+ );
+ }
+
+ @Test
+ public void testNarrowToQuerySubQueryStack()
+ {
+ final SegmentDescriptor descriptor = new
SegmentDescriptor(Intervals.of("2000/2001"), "0", 0);
+ final TimeseriesQuery query =
+ Druids.newTimeseriesQueryBuilder()
+ .dataSource(new
QueryDataSource(makeTimeseriesQuery("2000-03/2002")))
+ .intervals("1999/2000-02")
+ .granularity(Granularities.ALL)
+ .build();
+
+ Assertions.assertEquals(
+ new SegmentDescriptor(Intervals.of("2000-03/2001"), "0", 0),
+ Queries.narrowToQuery(descriptor, query)
+ );
+ }
+
+ private static TimeseriesQuery makeTimeseriesQuery(final String... intervals)
+ {
+ return Druids.newTimeseriesQueryBuilder()
+ .dataSource("foo")
+ .intervals(
+ new
MultipleIntervalSegmentSpec(Arrays.stream(intervals).map(Intervals::of).toList())
+ )
+ .granularity(Granularities.ALL)
+ .build();
+ }
+
@Test
public void testWithBaseDataSourceBasic()
{
diff --git
a/processing/src/test/java/org/apache/druid/query/scan/ScanQueryRunnerFactoryTest.java
b/processing/src/test/java/org/apache/druid/query/scan/ScanQueryRunnerFactoryTest.java
index c7031f636a3..1e6c5ee183c 100644
---
a/processing/src/test/java/org/apache/druid/query/scan/ScanQueryRunnerFactoryTest.java
+++
b/processing/src/test/java/org/apache/druid/query/scan/ScanQueryRunnerFactoryTest.java
@@ -20,8 +20,10 @@
package org.apache.druid.query.scan;
import com.google.common.collect.ImmutableList;
+import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.Pair;
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.Sequence;
@@ -34,11 +36,13 @@ import org.apache.druid.query.QueryRunner;
import org.apache.druid.query.QueryRunnerTestHelper;
import org.apache.druid.query.ResourceLimitExceededException;
import org.apache.druid.query.SegmentDescriptor;
+import org.apache.druid.query.SinkQueryRunners;
import org.apache.druid.query.context.ResponseContext;
import org.apache.druid.query.spec.LegacySegmentSpec;
import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
import org.apache.druid.query.spec.MultipleSpecificSegmentSpec;
import org.apache.druid.query.spec.QuerySegmentSpec;
+import org.apache.druid.query.spec.SpecificSegmentQueryRunner;
import org.apache.druid.query.spec.SpecificSegmentSpec;
import org.joda.time.Interval;
import org.junit.jupiter.api.Assertions;
@@ -46,7 +50,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedClass;
import org.junit.jupiter.params.provider.MethodSource;
-import java.io.IOException;
+import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -129,7 +133,7 @@ public class ScanQueryRunnerFactoryTest
}
@Test
- public void testSortAndLimitScanResultValues() throws IOException
+ public void testSortAndLimitScanResultValues()
{
List<ScanResultValue> srvs = new ArrayList<>(numElements);
List<Long> expectedEventTimestamps = new ArrayList<>();
@@ -153,12 +157,14 @@ public class ScanQueryRunnerFactoryTest
Sequence<ScanResultValue> inputSequence = Sequences.simple(srvs);
try {
List<ScanResultValue> output = FACTORY.stableLimitingSort(
- inputSequence,
- query,
- ImmutableList.of(new Interval(
- DateTimes.of("2010-01-01"),
- DateTimes.of("2019-01-01").plusHours(1)
- ))
+ ImmutableList.of(
+ Pair.of(
+ new Interval(DateTimes.of("2010-01-01"),
DateTimes.of("2019-01-01").plusHours(1)),
+ (queryPlus, responseContext) -> inputSequence
+ )
+ ),
+ QueryPlus.wrap(query),
+ ResponseContext.createEmpty()
).toList();
if (query.getScanRowsLimit() > Integer.MAX_VALUE) {
Assertions.fail("Unsupported exception should have been thrown due
to high limit");
@@ -284,6 +290,10 @@ public class ScanQueryRunnerFactoryTest
public static class ScanQueryRunnerFactoryNonParameterizedTest
{
+ private static final Interval HOUR_0 = Intervals.of("2000-01-01T00/PT1H");
+ private static final Interval HOUR_1 = Intervals.of("2000-01-01T01/PT1H");
+ private static final Interval HOUR_2 = Intervals.of("2000-01-01T02/PT1H");
+
private SegmentDescriptor descriptor = new SegmentDescriptor(new Interval(
DateTimes.of("2010-01-01"),
DateTimes.of("2019-01-01").plusHours(1)
@@ -373,5 +383,382 @@ public class ScanQueryRunnerFactoryTest
ex.getMessage()
);
}
+
+ @Test
+ public void testPriorityQueueDescendingSkipsOlderIntervals()
+ {
+ final RecordingRunner hour0 = new RecordingRunner(HOUR_0, 10, 20, 30);
+ final RecordingRunner hour1 = new RecordingRunner(HOUR_1, 10, 20, 30);
+ final RecordingRunner hour2Partition0 = new RecordingRunner(HOUR_2, 10,
20, 30);
+ final RecordingRunner hour2Partition1 = new RecordingRunner(HOUR_2, 15,
25, 35);
+
+ Assertions.assertEquals(
+ List.of(minute(HOUR_2, 35), minute(HOUR_2, 30), minute(HOUR_2, 25)),
+ runMerged(
+ Order.DESCENDING,
+ 3,
+ List.of(HOUR_0, HOUR_1, HOUR_2, HOUR_2),
+ List.of(hour0, hour1, hour2Partition0, hour2Partition1)
+ )
+ );
+ Assertions.assertEquals(
+ List.of(false, false, true, true),
+ ranFlags(hour0, hour1, hour2Partition0, hour2Partition1)
+ );
+ }
+
+ @Test
+ public void testPriorityQueueDescendingRunsIntervalsUntilLimitReached()
+ {
+ final RecordingRunner hour0 = new RecordingRunner(HOUR_0, 10, 20, 30);
+ final RecordingRunner hour1 = new RecordingRunner(HOUR_1, 10, 20, 30);
+ final RecordingRunner hour2 = new RecordingRunner(HOUR_2, 10, 20);
+
+ Assertions.assertEquals(
+ List.of(minute(HOUR_2, 20), minute(HOUR_2, 10), minute(HOUR_1, 30),
minute(HOUR_1, 20)),
+ runMerged(Order.DESCENDING, 4, List.of(HOUR_0, HOUR_1, HOUR_2),
List.of(hour0, hour1, hour2))
+ );
+ Assertions.assertEquals(List.of(false, true, true), ranFlags(hour0,
hour1, hour2));
+ }
+
+ @Test
+ public void testPriorityQueueAscendingSkipsLaterIntervals()
+ {
+ final RecordingRunner hour0Partition0 = new RecordingRunner(HOUR_0, 10,
20, 30);
+ final RecordingRunner hour0Partition1 = new RecordingRunner(HOUR_0, 15,
25, 35);
+ final RecordingRunner hour1 = new RecordingRunner(HOUR_1, 10, 20, 30);
+ final RecordingRunner hour2 = new RecordingRunner(HOUR_2, 10, 20, 30);
+
+ Assertions.assertEquals(
+ List.of(minute(HOUR_0, 10), minute(HOUR_0, 15), minute(HOUR_0, 20)),
+ runMerged(
+ Order.ASCENDING,
+ 3,
+ List.of(HOUR_0, HOUR_0, HOUR_1, HOUR_2),
+ List.of(hour0Partition0, hour0Partition1, hour1, hour2)
+ )
+ );
+ Assertions.assertEquals(
+ List.of(true, true, false, false),
+ ranFlags(hour0Partition0, hour0Partition1, hour1, hour2)
+ );
+ }
+
+ @Test
+ public void testPriorityQueueAscendingNarrowsLaterRunners()
+ {
+ final RecordingRunner hour0Partition0 = new RecordingRunner(HOUR_0, 10,
20, 30);
+ final RecordingRunner hour0Partition1 = new RecordingRunner(HOUR_0, 5,
15, 25, 35);
+ final RecordingRunner hour1 = new RecordingRunner(HOUR_1, 10);
+
+ Assertions.assertEquals(
+ List.of(minute(HOUR_0, 5), minute(HOUR_0, 10), minute(HOUR_0, 15)),
+ runMerged(
+ Order.ASCENDING,
+ 3,
+ List.of(HOUR_0, HOUR_0, HOUR_1),
+ List.of(hour0Partition0, hour0Partition1, hour1)
+ )
+ );
+
+ // The second partition reads only up to the last row of the first,
inclusive.
+
Assertions.assertEquals(List.of(Intervals.of("2000-01-01T00/2000-01-01T02")),
hour0Partition0.query.getIntervals());
+ Assertions.assertEquals(
+ List.of(Intervals.utc(HOUR_0.getStartMillis(), minute(HOUR_0, 30) +
1)),
+ hour0Partition1.query.getIntervals()
+ );
+ Assertions.assertNull(hour1.query);
+ }
+
+ @Test
+ public void testPriorityQueueDescendingNarrowsLaterRunners()
+ {
+ final RecordingRunner hour0 = new RecordingRunner(HOUR_0, 50);
+ final RecordingRunner hour1Partition0 = new RecordingRunner(HOUR_1, 10,
20, 30);
+ final RecordingRunner hour1Partition1 = new RecordingRunner(HOUR_1, 5,
25, 40);
+
+ Assertions.assertEquals(
+ List.of(minute(HOUR_1, 40), minute(HOUR_1, 30), minute(HOUR_1, 25)),
+ runMerged(
+ Order.DESCENDING,
+ 3,
+ List.of(HOUR_0, HOUR_1, HOUR_1),
+ List.of(hour0, hour1Partition0, hour1Partition1)
+ )
+ );
+
+ // Runners run latest first, so the first partition reads only down to
the last row of the second, inclusive.
+
Assertions.assertEquals(List.of(Intervals.of("2000-01-01T00/2000-01-01T02")),
hour1Partition1.query.getIntervals());
+ Assertions.assertEquals(
+ List.of(Intervals.utc(minute(HOUR_1, 5), HOUR_1.getEndMillis())),
+ hour1Partition0.query.getIntervals()
+ );
+ Assertions.assertNull(hour0.query);
+ }
+
+ @Test
+ public void testPriorityQueueRetainsEqualTimestampFromLaterRunner()
+ {
+ // Rows with equal timestamps are ordered by segment ID, so the later
runner's row at minute 10 displaces the
+ // earlier runner's row at minute 10.
+ final RecordingRunner first = new RecordingRunner("b", HOUR_0, 0, 10);
+ final RecordingRunner second = new RecordingRunner("a", HOUR_0, 10, 20);
+
+ final List<ScanResultValue> rows =
+ runMergedRows(Order.ASCENDING, 2, List.of(HOUR_0, HOUR_0),
List.of(first, second));
+
+ Assertions.assertEquals(
+ List.of("b", "a"),
+
rows.stream().map(ScanResultValue::getSegmentId).collect(Collectors.toList())
+ );
+ Assertions.assertEquals(List.of(minute(HOUR_0, 0), minute(HOUR_0, 10)),
toTimestamps(rows));
+ Assertions.assertEquals(
+ List.of(Intervals.utc(HOUR_0.getStartMillis(), minute(HOUR_0, 10) +
1)),
+ second.query.getIntervals()
+ );
+ }
+
+ @Test
+ public void testPriorityQueueNarrowsSpecificSegmentDescriptors()
+ {
+ // Wrap runners in SpecificSegmentQueryRunner, as data servers do, so
narrowed intervals reach each segment as a
+ // narrowed descriptor.
+ final List<SegmentDescriptor> descriptors =
makeDescriptors(List.of(HOUR_0, HOUR_1, HOUR_1));
+ final List<RecordingRunner> recordingRunners = List.of(
+ new RecordingRunner(HOUR_0, 50),
+ new RecordingRunner(HOUR_1, 10, 40),
+ new RecordingRunner(HOUR_1, 20, 50)
+ );
+
+ Assertions.assertEquals(
+ List.of(minute(HOUR_1, 50), minute(HOUR_1, 40)),
+ runMerged(
+ Order.DESCENDING,
+ 2,
+ List.of(HOUR_0, HOUR_1, HOUR_1),
+ IntStream.range(0, descriptors.size())
+ .mapToObj(
+ i -> (QueryRunner<ScanResultValue>) new
SpecificSegmentQueryRunner<>(
+ recordingRunners.get(i),
+ new SpecificSegmentSpec(descriptors.get(i))
+ )
+ )
+ .collect(Collectors.toList())
+ )
+ );
+
+ Assertions.assertNull(recordingRunners.get(0).query);
+ Assertions.assertEquals(
+ new MultipleSpecificSegmentSpec(
+ List.of(new SegmentDescriptor(Intervals.utc(minute(HOUR_1, 20),
HOUR_1.getEndMillis()), "v0", 1))
+ ),
+ recordingRunners.get(1).query.getQuerySegmentSpec()
+ );
+ Assertions.assertEquals(
+ new MultipleSpecificSegmentSpec(List.of(descriptors.get(2))),
+ recordingRunners.get(2).query.getQuerySegmentSpec()
+ );
+ }
+
+ @Test
+ public void testRunnersThatDoNotMatchDescriptors()
+ {
+ // One fewer runner than descriptors, so runners cannot be matched to
intervals.
+ final List<QueryRunner<ScanResultValue>> runners = List.of(
+ new RecordingRunner(HOUR_0, 10, 20, 30),
+ new RecordingRunner(HOUR_2, 10, 20, 30)
+ );
+ final List<Interval> descriptorIntervals = List.of(HOUR_0, HOUR_1,
HOUR_2);
+
+ // Priority queue strategy.
+ Assertions.assertThrows(
+ DruidException.class,
+ () -> runMerged(Order.DESCENDING, 2, descriptorIntervals, runners)
+ );
+
+ // N-way merge strategy.
+ Assertions.assertThrows(
+ DruidException.class,
+ () -> runMerged(Order.DESCENDING,
CONFIG.getMaxRowsQueuedForOrdering() + 1, descriptorIntervals, runners)
+ );
+ }
+
+ @Test
+ public void
testPriorityQueueDescendingSinkQueryRunnersSkipsOlderIntervals()
+ {
+ final RecordingRunner hour0Hydrant0 = new RecordingRunner(HOUR_0, 10,
20);
+ final RecordingRunner hour0Hydrant1 = new RecordingRunner(HOUR_0, 30);
+ final RecordingRunner hour1Hydrant0 = new RecordingRunner(HOUR_1, 10,
30);
+ final RecordingRunner hour1Hydrant1 = new RecordingRunner(HOUR_1, 20);
+
+ Assertions.assertEquals(
+ List.of(minute(HOUR_1, 30), minute(HOUR_1, 20)),
+ runMerged(
+ Order.DESCENDING,
+ 2,
+ List.of(HOUR_0, HOUR_1),
+ new SinkQueryRunners<>(
+ List.of(
+ Pair.of(HOUR_0, hour0Hydrant0),
+ Pair.of(HOUR_0, hour0Hydrant1),
+ Pair.of(HOUR_1, hour1Hydrant0),
+ Pair.of(HOUR_1, hour1Hydrant1)
+ )
+ )
+ )
+ );
+ Assertions.assertEquals(
+ List.of(false, false, true, true),
+ ranFlags(hour0Hydrant0, hour0Hydrant1, hour1Hydrant0, hour1Hydrant1)
+ );
+ }
+
+ @Test
+ public void testNWayMergeDescendingSinkQueryRunners()
+ {
+ Assertions.assertEquals(
+ List.of(
+ minute(HOUR_1, 30),
+ minute(HOUR_1, 20),
+ minute(HOUR_1, 10),
+ minute(HOUR_0, 30),
+ minute(HOUR_0, 20),
+ minute(HOUR_0, 10)
+ ),
+ runMerged(
+ Order.DESCENDING,
+ CONFIG.getMaxRowsQueuedForOrdering() + 1,
+ List.of(HOUR_0, HOUR_1),
+ new SinkQueryRunners<>(
+ List.of(
+ Pair.of(HOUR_0, new RecordingRunner(HOUR_0, 10, 20)),
+ Pair.of(HOUR_0, new RecordingRunner(HOUR_0, 30)),
+ Pair.of(HOUR_1, new RecordingRunner(HOUR_1, 10, 30)),
+ Pair.of(HOUR_1, new RecordingRunner(HOUR_1, 20))
+ )
+ )
+ )
+ );
+ }
+
+ private static List<Long> runMerged(
+ final Order order,
+ final long limit,
+ final List<Interval> descriptorIntervals,
+ final Iterable<QueryRunner<ScanResultValue>> runners
+ )
+ {
+ return toTimestamps(runMergedRows(order, limit, descriptorIntervals,
runners));
+ }
+
+ private static List<ScanResultValue> runMergedRows(
+ final Order order,
+ final long limit,
+ final List<Interval> descriptorIntervals,
+ final Iterable<QueryRunner<ScanResultValue>> runners
+ )
+ {
+ final ScanQuery query =
+ Druids.newScanQueryBuilder()
+ .dataSource("foo")
+ .intervals(new
MultipleSpecificSegmentSpec(makeDescriptors(descriptorIntervals)))
+ .resultFormat(ScanQuery.ResultFormat.RESULT_FORMAT_LIST)
+ .order(order)
+ .limit(limit)
+ .build();
+
+ return FACTORY.mergeRunners(Execs.directExecutor(), runners)
+ .run(QueryPlus.wrap(query), ResponseContext.createEmpty())
+ .toList()
+ .stream()
+ .flatMap(batch ->
batch.toSingleEventScanResultValues().stream())
+ .collect(Collectors.toList());
+ }
+
+ private static List<SegmentDescriptor> makeDescriptors(final
List<Interval> intervals)
+ {
+ return IntStream.range(0, intervals.size())
+ .mapToObj(i -> new SegmentDescriptor(intervals.get(i),
"v0", i))
+ .collect(Collectors.toList());
+ }
+
+ private static List<Long> toTimestamps(final List<ScanResultValue> rows)
+ {
+ return rows.stream()
+ .map(row ->
row.getFirstEventTimestamp(ScanQuery.ResultFormat.RESULT_FORMAT_LIST))
+ .collect(Collectors.toList());
+ }
+
+ private static List<Boolean> ranFlags(final RecordingRunner... runners)
+ {
+ return Arrays.stream(runners).map(runner -> runner.query !=
null).collect(Collectors.toList());
+ }
+
+ private static long minute(final Interval hour, final int minute)
+ {
+ return hour.getStart().plusMinutes(minute).getMillis();
+ }
+
+ /**
+ * Runner that returns one single-row batch per timestamp within the query
intervals, in the time order of the
+ * query, as a cursor would. Records the query it ran with.
+ */
+ private static class RecordingRunner implements
QueryRunner<ScanResultValue>
+ {
+ @Nullable
+ private final String segmentId;
+ private final List<Long> timestamps;
+
+ /**
+ * Query this runner ran with, or null if it did not run.
+ */
+ @Nullable
+ private ScanQuery query;
+
+ RecordingRunner(final Interval hour, final int... minutes)
+ {
+ this(null, hour, minutes);
+ }
+
+ /**
+ * Creates a runner whose rows have {@code segmentId}, or the default of
+ * {@link ScanQueryTestHelper#generateScanResultValue} if null.
+ */
+ RecordingRunner(@Nullable final String segmentId, final Interval hour,
final int... minutes)
+ {
+ this.segmentId = segmentId;
+ this.timestamps = Arrays.stream(minutes)
+ .mapToObj(minute ->
hour.getStart().plusMinutes(minute).getMillis())
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public Sequence<ScanResultValue> run(
+ final QueryPlus<ScanResultValue> queryPlus,
+ final ResponseContext responseContext
+ )
+ {
+ query = (ScanQuery) queryPlus.getQuery();
+ final List<Interval> queryIntervals = query.getIntervals();
+ final List<Long> orderedTimestamps =
+ timestamps.stream()
+ .filter(t -> queryIntervals.stream().anyMatch(interval
-> interval.contains(t)))
+ .collect(Collectors.toList());
+ if (Order.DESCENDING.equals(query.getTimeOrder())) {
+ Collections.reverse(orderedTimestamps);
+ }
+ return Sequences.simple(
+ orderedTimestamps.stream()
+ .map(t -> makeRow(t, query.getResultFormat()))
+ .collect(Collectors.toList())
+ );
+ }
+
+ private ScanResultValue makeRow(final long timestamp, final
ScanQuery.ResultFormat resultFormat)
+ {
+ final ScanResultValue row =
ScanQueryTestHelper.generateScanResultValue(timestamp, resultFormat, 1);
+ return segmentId == null ? row : new ScanResultValue(segmentId,
row.getColumns(), row.getEvents());
+ }
+ }
}
}
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 337a5652026..a11fe9a9626 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
@@ -31,6 +31,7 @@ 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.guava.Yielder;
import org.apache.druid.java.util.common.guava.YieldingAccumulator;
+import org.apache.druid.query.BaseQuery;
import org.apache.druid.query.Druids;
import org.apache.druid.query.QueryContext;
import org.apache.druid.query.QueryPlus;
@@ -206,6 +207,47 @@ public class SpecificSegmentQueryRunnerTest
assertThreadNameDuringProcessing(true, "processing_thread-name-query");
}
+ @Test
+ public void testNarrowsDescriptorToQueryIntervals()
+ {
+ final SegmentDescriptor descriptor = new
SegmentDescriptor(Intervals.of("2012-01-01/P1D"), "version", 0);
+ final AtomicReference<QuerySegmentSpec> baseSegmentSpec = new
AtomicReference<>();
+ final SpecificSegmentQueryRunner<Result<TimeseriesResultValue>>
queryRunner = new SpecificSegmentQueryRunner<>(
+ (queryPlus, responseContext) -> {
+ baseSegmentSpec.set(((BaseQuery<?>)
queryPlus.getQuery()).getQuerySegmentSpec());
+ return Sequences.empty();
+ },
+ new SpecificSegmentSpec(descriptor)
+ );
+
+
queryRunner.run(QueryPlus.wrap(makeCountQuery("2011-12-31T12/2012-01-01T06")),
ResponseContext.createEmpty())
+ .toList();
+
+ Assertions.assertEquals(
+ new MultipleSpecificSegmentSpec(
+ List.of(new
SegmentDescriptor(Intervals.of("2012-01-01/2012-01-01T06"), "version", 0))
+ ),
+ baseSegmentSpec.get()
+ );
+ }
+
+ @Test
+ public void testSkipsSegmentOutsideQueryIntervals()
+ {
+ final SegmentDescriptor descriptor = new
SegmentDescriptor(Intervals.of("2012-01-01/P1D"), "version", 0);
+ final SpecificSegmentQueryRunner<Result<TimeseriesResultValue>>
queryRunner = new SpecificSegmentQueryRunner<>(
+ (queryPlus, responseContext) -> Assertions.fail("Segment should not
run"),
+ new SpecificSegmentSpec(descriptor)
+ );
+
+ final ResponseContext responseContext = ResponseContext.createEmpty();
+ Assertions.assertEquals(
+ List.of(),
+ queryRunner.run(QueryPlus.wrap(makeCountQuery("2012-01-02/P1D")),
responseContext).toList()
+ );
+ Assertions.assertNull(responseContext.getMissingSegments());
+ }
+
private void assertThreadNameDuringProcessing(
final Boolean setProcessingThreadNames,
final String expectedThreadNameDuringProcessing
@@ -282,6 +324,16 @@ public class SpecificSegmentQueryRunnerTest
}
}
+ private static TimeseriesQuery makeCountQuery(final String interval)
+ {
+ return Druids.newTimeseriesQueryBuilder()
+ .dataSource("foo")
+ .granularity(Granularities.ALL)
+ .intervals(ImmutableList.of(Intervals.of(interval)))
+ .aggregators(ImmutableList.of(new
CountAggregatorFactory("rows")))
+ .build();
+ }
+
private static Result<TimeseriesResultValue> makeResult()
{
final TimeseriesResultBuilder builder = new TimeseriesResultBuilder(
diff --git
a/server/src/main/java/org/apache/druid/client/CachingQueryRunner.java
b/server/src/main/java/org/apache/druid/client/CachingQueryRunner.java
index 41d4bb4ea63..d718254412d 100644
--- a/server/src/main/java/org/apache/druid/client/CachingQueryRunner.java
+++ b/server/src/main/java/org/apache/druid/client/CachingQueryRunner.java
@@ -31,6 +31,7 @@ import org.apache.druid.java.util.common.guava.BaseSequence;
import org.apache.druid.java.util.common.guava.Sequence;
import org.apache.druid.java.util.common.guava.Sequences;
import org.apache.druid.query.CacheStrategy;
+import org.apache.druid.query.Queries;
import org.apache.druid.query.Query;
import org.apache.druid.query.QueryPlus;
import org.apache.druid.query.QueryRunner;
@@ -87,6 +88,14 @@ public class CachingQueryRunner<T> implements QueryRunner<T>
{
Query<T> query = queryPlus.getQuery();
final CacheStrategy strategy = toolChest.getCacheStrategy(query, mapper);
+
+ // Key on the part of the segment that the query covers, since callers may
narrow the query to skip part of it.
+ final SegmentDescriptor descriptor =
Queries.narrowToQuery(segmentDescriptor, query);
+ if (descriptor == null) {
+ // Nothing worth caching, since the query does not overlap this segment.
+ return base.run(queryPlus, responseContext);
+ }
+
final boolean populateCache = canPopulateCache(query, strategy);
final boolean useCache = canUseCache(query, strategy);
@@ -94,7 +103,7 @@ public class CachingQueryRunner<T> implements QueryRunner<T>
if (useCache || populateCache) {
key = CacheUtil.computeSegmentCacheKey(
cacheId,
- alignToActualDataInterval(segmentDescriptor),
+ alignToActualDataInterval(descriptor),
Bytes.concat(cacheKeyPrefix.get(), strategy.computeCacheKey(query))
);
} else {
diff --git a/server/src/main/java/org/apache/druid/server/ServerManager.java
b/server/src/main/java/org/apache/druid/server/ServerManager.java
index a85e3e15874..3845f38a6f5 100644
--- a/server/src/main/java/org/apache/druid/server/ServerManager.java
+++ b/server/src/main/java/org/apache/druid/server/ServerManager.java
@@ -21,6 +21,7 @@ package org.apache.druid.server;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.inject.Inject;
import org.apache.druid.client.CachingQueryRunner;
@@ -62,6 +63,7 @@ import org.apache.druid.query.context.ResponseContext;
import org.apache.druid.query.metadata.metadata.SegmentMetadataQuery;
import org.apache.druid.query.planning.ExecutionVertex;
import org.apache.druid.query.policy.PolicyEnforcer;
+import org.apache.druid.query.scan.ScanQueryRunnerFactory;
import org.apache.druid.query.spec.SpecificSegmentQueryRunner;
import org.apache.druid.query.spec.SpecificSegmentSpec;
import org.apache.druid.segment.Segment;
@@ -83,7 +85,9 @@ import org.joda.time.Interval;
import javax.annotation.Nullable;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
@@ -410,32 +414,51 @@ public class ServerManager implements QuerySegmentWalker
);
}
- protected <T> FunctionalIterable<QueryRunner<T>> getQueryRunnersForSegments(
- final Query<T> query,
+ /**
+ * Returns one runner per spec, in the order of {@code specs}. The ordering
is important because
+ * {@link QueryRunnerFactory} implementations may rely on it. For example,
+ * {@link ScanQueryRunnerFactory#mergeRunners} matches runners to segment
descriptors positionally.
+ *
+ * <p>The order of {@code segmentReferences} need not match the order of
{@code specs}.
+ */
+ private <T> FunctionalIterable<QueryRunner<T>> getQueryRunnersForSegments(
+ final List<SegmentDescriptor> specs,
+ final List<SegmentReference> segmentReferences,
final QueryRunnerFactory<T, Query<T>> factory,
final QueryToolChest<T, Query<T>> toolChest,
- final List<SegmentReference> segmentReferences,
final AtomicLong cpuTimeAccumulator,
final Optional<byte[]> cacheKeyPrefix
)
{
+ final Map<SegmentDescriptor, SegmentReference> referencesByDescriptor =
new HashMap<>();
+ for (final SegmentReference reference : segmentReferences) {
+ referencesByDescriptor.put(reference.getSegmentDescriptor(), reference);
+ }
+
return FunctionalIterable
- .create(segmentReferences)
+ .create(specs)
.transform(
- ref ->
- ref.getSegmentReference()
- .map(segment ->
- buildQueryRunnerForSegment(
- ref.getSegmentDescriptor(),
- segment,
- factory,
- toolChest,
- cpuTimeAccumulator,
- cacheKeyPrefix
- )
- ).orElseThrow(
- () -> DruidException.defensive("Unexpected missing
segment[%s]", ref.getSegmentDescriptor())
- )
+ spec -> {
+ final SegmentReference ref = referencesByDescriptor.get(spec);
+ if (ref == null) {
+ // Return nothing: ResourceManagingQueryRunner already
reported this spec as missing.
+ return new NoopQueryRunner<>();
+ }
+
+ return ref.getSegmentReference()
+ .map(segment ->
+ buildQueryRunnerForSegment(
+ ref.getSegmentDescriptor(),
+ segment,
+ factory,
+ toolChest,
+ cpuTimeAccumulator,
+ cacheKeyPrefix
+ )
+ ).orElseThrow(
+ () -> DruidException.defensive("Unexpected missing
segment[%s]", ref.getSegmentDescriptor())
+ );
+ }
);
}
@@ -558,7 +581,7 @@ public class ServerManager implements QuerySegmentWalker
private final QueryRunnerFactory<T, Query<T>> factory;
private final QueryToolChest<T, Query<T>> toolChest;
private final ExecutionVertex ev;
- private final Iterable<SegmentDescriptor> specs;
+ private final List<SegmentDescriptor> specs;
public ResourceManagingQueryRunner(
VersionedIntervalTimeline<String, DataSegment> timeline,
@@ -572,7 +595,8 @@ public class ServerManager implements QuerySegmentWalker
this.factory = factory;
this.toolChest = toolChest;
this.ev = ev;
- this.specs = specs;
+ // Copy, since run() iterates specs twice and both passes must see the
same descriptors.
+ this.specs = ImmutableList.copyOf(specs);
}
@Override
@@ -607,10 +631,10 @@ public class ServerManager implements QuerySegmentWalker
queryPlus.getQueryMetrics().reportQueriedSegmentCount(segmentReferences.size()).emit(emitter);
final FunctionalIterable<QueryRunner<T>> queryRunners =
getQueryRunnersForSegments(
- query,
+ specs,
+ segmentReferences,
factory,
toolChest,
- segmentReferences,
cpuTimeAccumulator,
cacheKeyPrefix
);
diff --git
a/server/src/test/java/org/apache/druid/client/CachingQueryRunnerTest.java
b/server/src/test/java/org/apache/druid/client/CachingQueryRunnerTest.java
index 3a4d1759db8..61e465b7fcb 100644
--- a/server/src/test/java/org/apache/druid/client/CachingQueryRunnerTest.java
+++ b/server/src/test/java/org/apache/druid/client/CachingQueryRunnerTest.java
@@ -214,6 +214,39 @@ public class CachingQueryRunnerTest extends
InitializedNullHandlingTest
}
}
+ @Test
+ public void testQueryOnPartOfSegmentDoesNotUseWholeSegmentCacheEntry()
throws IOException
+ {
+ final Query query = new TopNQueryBuilder()
+ .dataSource("ds")
+ .dimension("top_dim")
+ .metric("imps")
+ .threshold(3)
+ .intervals("2011-01-05/2011-01-10")
+ .aggregators(AGGS)
+ .granularity(Granularities.ALL)
+ .build();
+ final QueryToolChest toolchest = new TopNQueryQueryToolChest(new
TopNQueryConfig());
+ final CacheStrategy cacheStrategy = toolchest.getCacheStrategy(query,
null);
+ final byte[] cacheKeyPrefix = RandomUtils.nextBytes(10);
+
+ // Cache results under the key for the whole segment, as a query covering
all of it would.
+ final Cache cache = MapCache.create(1024 * 1024);
+ cache.put(
+ CacheUtil.computeSegmentCacheKey(
+ CACHE_ID,
+ SEGMENT_DESCRIPTOR,
+ Bytes.concat(cacheKeyPrefix, cacheStrategy.computeCacheKey(query))
+ ),
+ toByteArray(
+ Iterables.transform(makeTopNResults(true, OBJECTS),
cacheStrategy.prepareForSegmentLevelCache())
+ )
+ );
+
+ final CachingQueryRunner runner = makeCachingQueryRunner(cacheKeyPrefix,
cache, toolchest, Sequences.empty());
+ Assertions.assertEquals(List.of(),
runner.run(QueryPlus.wrap(query)).toList());
+ }
+
@Test
public void testNullCacheKeyPrefix()
{
@@ -351,7 +384,7 @@ public class CachingQueryRunnerTest extends
InitializedNullHandlingTest
CacheStrategy cacheStrategy = toolchest.getCacheStrategy(query, null);
Cache.NamedKey cacheKey = CacheUtil.computeSegmentCacheKey(
CACHE_ID,
- SEGMENT_DESCRIPTOR,
+ narrowSegmentDescriptor(query),
Bytes.concat(keyPrefix, cacheStrategy.computeCacheKey(query))
);
Assertions.assertTrue(runner.canPopulateCache(query, cacheStrategy));
@@ -395,7 +428,7 @@ public class CachingQueryRunnerTest extends
InitializedNullHandlingTest
CacheStrategy cacheStrategy = toolchest.getCacheStrategy(query, null);
Cache.NamedKey cacheKey = CacheUtil.computeSegmentCacheKey(
CACHE_ID,
- SEGMENT_DESCRIPTOR,
+ narrowSegmentDescriptor(query),
Bytes.concat(cacheKeyPrefix, cacheStrategy.computeCacheKey(query))
);
@@ -413,6 +446,19 @@ public class CachingQueryRunnerTest extends
InitializedNullHandlingTest
Assertions.assertEquals(expectedResults.toString(), results.toString());
}
+ /**
+ * Returns {@link #SEGMENT_DESCRIPTOR} narrowed to the interval of {@code
query}, which is where the runner keys its
+ * cache entries.
+ */
+ private static SegmentDescriptor narrowSegmentDescriptor(final Query<?>
query)
+ {
+ return new SegmentDescriptor(
+
Iterables.getOnlyElement(query.getIntervals()).overlap(SEGMENT_DESCRIPTOR.getInterval()),
+ SEGMENT_DESCRIPTOR.getVersion(),
+ SEGMENT_DESCRIPTOR.getPartitionNumber()
+ );
+ }
+
private CachingQueryRunner makeCachingQueryRunner(
byte[] cacheKeyPrefix,
Cache cache,
diff --git
a/server/src/test/java/org/apache/druid/server/coordination/ServerManagerScanOrderingTest.java
b/server/src/test/java/org/apache/druid/server/coordination/ServerManagerScanOrderingTest.java
new file mode 100644
index 00000000000..5eb856b95d8
--- /dev/null
+++
b/server/src/test/java/org/apache/druid/server/coordination/ServerManagerScanOrderingTest.java
@@ -0,0 +1,357 @@
+/*
+ * 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.server.coordination;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.google.inject.Guice;
+import com.google.inject.Inject;
+import com.google.inject.testing.fieldbinder.Bind;
+import com.google.inject.testing.fieldbinder.BoundFieldModule;
+import org.apache.druid.client.cache.Cache;
+import org.apache.druid.client.cache.CacheConfig;
+import org.apache.druid.client.cache.CachePopulator;
+import org.apache.druid.client.cache.CachePopulatorStats;
+import org.apache.druid.client.cache.ForegroundCachePopulator;
+import org.apache.druid.client.cache.LocalCacheProvider;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.guice.annotations.Smile;
+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.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.query.DefaultGenericQueryMetricsFactory;
+import org.apache.druid.query.DefaultQueryRunnerFactoryConglomerate;
+import org.apache.druid.query.Druids;
+import org.apache.druid.query.ForwardingQueryProcessingPool;
+import org.apache.druid.query.Order;
+import org.apache.druid.query.QueryPlus;
+import org.apache.druid.query.QueryProcessingPool;
+import org.apache.druid.query.QueryRunnerFactoryConglomerate;
+import org.apache.druid.query.SegmentDescriptor;
+import org.apache.druid.query.context.DefaultResponseContext;
+import org.apache.druid.query.policy.NoopPolicyEnforcer;
+import org.apache.druid.query.policy.PolicyEnforcer;
+import org.apache.druid.query.scan.ScanQuery;
+import org.apache.druid.query.scan.ScanQueryConfig;
+import org.apache.druid.query.scan.ScanQueryEngine;
+import org.apache.druid.query.scan.ScanQueryQueryToolChest;
+import org.apache.druid.query.scan.ScanQueryRunnerFactory;
+import org.apache.druid.query.scan.ScanResultValue;
+import org.apache.druid.query.spec.MultipleSpecificSegmentSpec;
+import org.apache.druid.segment.IncrementalIndexSegment;
+import org.apache.druid.segment.IndexBuilder;
+import org.apache.druid.segment.Segment;
+import org.apache.druid.segment.loading.AcquireMode;
+import org.apache.druid.server.SegmentManager;
+import org.apache.druid.server.ServerManager;
+import org.apache.druid.server.initialization.ServerConfig;
+import org.apache.druid.server.metrics.NoopServiceEmitter;
+import org.apache.druid.test.utils.TestSegmentCacheManager;
+import org.apache.druid.test.utils.TestSegmentManager;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.joda.time.Interval;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ExecutorService;
+
+/**
+ * Tests for time-ordered scan queries through {@link ServerManager} when some
segments are loaded on demand.
+ */
+public class ServerManagerScanOrderingTest
+{
+ private static final ScanQuery.ResultFormat RESULT_FORMAT =
ScanQuery.ResultFormat.RESULT_FORMAT_LIST;
+ private static final List<Interval> DAYS = List.of(
+ Intervals.of("2011-04-01/2011-04-02"),
+ Intervals.of("2011-04-02/2011-04-03"),
+ Intervals.of("2011-04-03/2011-04-04")
+ );
+ private static final int ROWS_PER_DAY = 3;
+
+ @Bind
+ private QueryRunnerFactoryConglomerate conglomerate;
+ @Bind
+ private SegmentManager segmentManager;
+ @Bind
+ private PolicyEnforcer policyEnforcer;
+ @Bind
+ private ServerConfig serverConfig;
+ @Bind
+ private ServiceEmitter serviceEmitter;
+ @Bind
+ private QueryProcessingPool queryProcessingPool;
+ @Bind
+ private CachePopulator cachePopulator;
+ @Bind
+ @Smile
+ private ObjectMapper objectMapper;
+ @Bind
+ private Cache cache;
+ @Bind
+ private CacheConfig cacheConfig;
+
+ @Inject
+ private ServerManager serverManager;
+
+ private OnDemandSegmentCacheManager cacheManager;
+ private ExecutorService exec;
+ private List<SegmentDescriptor> descriptors;
+ private List<DataSegment> dataSegments;
+
+ @BeforeEach
+ public void setUp() throws Exception
+ {
+ serviceEmitter = new NoopServiceEmitter();
+ EmittingLogger.registerEmitter(new NoopServiceEmitter());
+ cacheManager = new OnDemandSegmentCacheManager();
+ segmentManager = new SegmentManager(cacheManager);
+
+ descriptors = new ArrayList<>();
+ dataSegments = new ArrayList<>();
+ for (final Interval day : DAYS) {
+ final SegmentId segmentId = SegmentId.of("test", day, "1", 0);
+ final DataSegment dataSegment =
TestSegmentManager.createDataSegmentForTest(segmentId);
+ final List<InputRow> rows = new ArrayList<>();
+ for (int hour = 0; hour < ROWS_PER_DAY; hour++) {
+ rows.add(
+ new MapBasedInputRow(
+ day.getStart().plusHours(hour),
+ List.of("dim"),
+ Map.of("dim", "x")
+ )
+ );
+ }
+ cacheManager.registerSegment(
+ dataSegment,
+ new
IncrementalIndexSegment(IndexBuilder.create().rows(rows).buildIncrementalIndex(),
segmentId)
+ );
+ segmentManager.loadSegment(dataSegment);
+ dataSegments.add(dataSegment);
+ descriptors.add(segmentId.toDescriptor());
+ }
+
+ conglomerate =
DefaultQueryRunnerFactoryConglomerate.buildFromQueryRunnerFactories(
+ ImmutableMap.of(
+ ScanQuery.class,
+ new ScanQueryRunnerFactory(
+ new
ScanQueryQueryToolChest(DefaultGenericQueryMetricsFactory.instance()),
+ new ScanQueryEngine(),
+ new ScanQueryConfig()
+ )
+ )
+ );
+ exec = Execs.multiThreaded(2, "ServerManagerScanOrderingTest-%d");
+ queryProcessingPool = new ForwardingQueryProcessingPool(exec);
+ cachePopulator = new ForegroundCachePopulator(new DefaultObjectMapper(),
new CachePopulatorStats(), -1);
+ objectMapper = new DefaultObjectMapper();
+ cache = new LocalCacheProvider().get();
+ cacheConfig = new CacheConfig();
+ serverConfig = new ServerConfig();
+ policyEnforcer = NoopPolicyEnforcer.instance();
+
+ Guice.createInjector(BoundFieldModule.of(this)).injectMembers(this);
+ }
+
+ @AfterEach
+ public void tearDown()
+ {
+ exec.shutdownNow();
+ }
+
+ @Test
+ public void testAscendingUnlimited_allCached()
+ {
+ Assertions.assertEquals(expectedTimes(Order.ASCENDING, Long.MAX_VALUE),
runScan(Order.ASCENDING, Long.MAX_VALUE));
+ }
+
+ @Test
+ public void testAscendingUnlimited_firstDayOnDemand()
+ {
+ cacheManager.evictFromCache(dataSegments.get(0));
+ Assertions.assertEquals(expectedTimes(Order.ASCENDING, Long.MAX_VALUE),
runScan(Order.ASCENDING, Long.MAX_VALUE));
+ }
+
+ @Test
+ public void testDescendingUnlimited_firstDayOnDemand()
+ {
+ cacheManager.evictFromCache(dataSegments.get(0));
+ Assertions.assertEquals(
+ expectedTimes(Order.DESCENDING, Long.MAX_VALUE),
+ runScan(Order.DESCENDING, Long.MAX_VALUE)
+ );
+ }
+
+ @Test
+ public void testAscendingLimited_allCached()
+ {
+ Assertions.assertEquals(expectedTimes(Order.ASCENDING, 5),
runScan(Order.ASCENDING, 5));
+ }
+
+ @Test
+ public void testAscendingLimited_firstDayOnDemand()
+ {
+ cacheManager.evictFromCache(dataSegments.get(0));
+ Assertions.assertEquals(expectedTimes(Order.ASCENDING, 5),
runScan(Order.ASCENDING, 5));
+ }
+
+ @Test
+ public void testDescendingLimited_firstDayOnDemand()
+ {
+ cacheManager.evictFromCache(dataSegments.get(0));
+ Assertions.assertEquals(expectedTimes(Order.DESCENDING, 5),
runScan(Order.DESCENDING, 5));
+ }
+
+ @Test
+ public void testAscendingUnlimited_missingSegment()
+ {
+ final SegmentDescriptor missing = new SegmentDescriptor(DAYS.get(1),
"unknown_version", 0);
+ descriptors.add(1, missing);
+ final DefaultResponseContext responseContext =
DefaultResponseContext.createEmpty();
+ Assertions.assertEquals(
+ expectedTimes(Order.ASCENDING, Long.MAX_VALUE),
+ runScan(Order.ASCENDING, Long.MAX_VALUE, responseContext)
+ );
+ Assertions.assertEquals(List.of(missing),
responseContext.getMissingSegments());
+ }
+
+ @Test
+ public void testDescendingLimited_missingSegment()
+ {
+ final SegmentDescriptor missing = new SegmentDescriptor(DAYS.get(1),
"unknown_version", 0);
+ descriptors.add(1, missing);
+ final DefaultResponseContext responseContext =
DefaultResponseContext.createEmpty();
+ Assertions.assertEquals(expectedTimes(Order.DESCENDING, 5),
runScan(Order.DESCENDING, 5, responseContext));
+ Assertions.assertEquals(List.of(missing),
responseContext.getMissingSegments());
+ }
+
+ @Test
+ public void testAscendingUnlimited_onDemandAndMissingSegments()
+ {
+ cacheManager.evictFromCache(dataSegments.get(0));
+ final SegmentDescriptor missing = new SegmentDescriptor(DAYS.get(1),
"unknown_version", 0);
+ descriptors.add(1, missing);
+ final DefaultResponseContext responseContext =
DefaultResponseContext.createEmpty();
+ Assertions.assertEquals(
+ expectedTimes(Order.ASCENDING, Long.MAX_VALUE),
+ runScan(Order.ASCENDING, Long.MAX_VALUE, responseContext)
+ );
+ Assertions.assertEquals(List.of(missing),
responseContext.getMissingSegments());
+ }
+
+ @Test
+ public void testDescendingLimited_onDemandAndMissingSegments()
+ {
+ cacheManager.evictFromCache(dataSegments.get(0));
+ final SegmentDescriptor missing = new SegmentDescriptor(DAYS.get(1),
"unknown_version", 0);
+ descriptors.add(1, missing);
+ final DefaultResponseContext responseContext =
DefaultResponseContext.createEmpty();
+ Assertions.assertEquals(expectedTimes(Order.DESCENDING, 5),
runScan(Order.DESCENDING, 5, responseContext));
+ Assertions.assertEquals(List.of(missing),
responseContext.getMissingSegments());
+ }
+
+ private List<Long> runScan(final Order order, final long limit)
+ {
+ return runScan(order, limit, DefaultResponseContext.createEmpty());
+ }
+
+ private List<Long> runScan(final Order order, final long limit, final
DefaultResponseContext responseContext)
+ {
+ final ScanQuery query = Druids.newScanQueryBuilder()
+ .dataSource("test")
+ .intervals(new
MultipleSpecificSegmentSpec(descriptors))
+ .columns("__time", "dim")
+ .resultFormat(RESULT_FORMAT)
+ .order(order)
+ .limit(limit)
+ .batchSize(1)
+ .build();
+ final List<ScanResultValue> results =
serverManager.getQueryRunnerForSegments(query, descriptors)
+
.run(QueryPlus.wrap(query), responseContext)
+ .toList();
+ final List<Long> times = new ArrayList<>();
+ for (final ScanResultValue result : results) {
+ for (final ScanResultValue single :
result.toSingleEventScanResultValues()) {
+ times.add(single.getFirstEventTimestamp(RESULT_FORMAT));
+ }
+ }
+ return times;
+ }
+
+ private static List<Long> expectedTimes(final Order order, final long limit)
+ {
+ final List<Long> times = new ArrayList<>();
+ for (final Interval day : DAYS) {
+ for (int hour = 0; hour < ROWS_PER_DAY; hour++) {
+ times.add(day.getStart().plusHours(hour).getMillis());
+ }
+ }
+ if (order == Order.DESCENDING) {
+ Collections.reverse(times);
+ }
+ return new ArrayList<>(times.subList(0, (int) Math.min(limit,
times.size())));
+ }
+
+ /**
+ * Cache manager that can report specific segments as absent from the cache
but loadable on demand.
+ */
+ private static class OnDemandSegmentCacheManager extends
TestSegmentCacheManager
+ {
+ private final Set<SegmentId> evicted = new HashSet<>();
+
+ void evictFromCache(final DataSegment dataSegment)
+ {
+ evicted.add(dataSegment.getId());
+ }
+
+ @Override
+ public Optional<Segment> acquireCachedSegment(final SegmentId segmentId,
final AcquireMode acquireMode)
+ {
+ if (evicted.contains(segmentId)) {
+ return Optional.empty();
+ }
+ return super.acquireCachedSegment(segmentId, acquireMode);
+ }
+
+ @Override
+ public boolean canLoadSegmentsOnDemand()
+ {
+ return true;
+ }
+
+ @Override
+ public boolean canLoadSegmentOnDemand(final DataSegment dataSegment)
+ {
+ return true;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
