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 b4882aa503a perf: Descending vector cursors. (#19947)
b4882aa503a is described below
commit b4882aa503a9b78391e33ac63c6ea9af9780e294
Author: Gian Merlino <[email protected]>
AuthorDate: Mon Aug 24 12:20:38 2026 -0700
perf: Descending vector cursors. (#19947)
This patch adds descending-order vector cursors, enabling faster
execution for timeseries, timeBoundary, and scan queries that run in
descending order.
---
.../druid/benchmark/query/TimeseriesBenchmark.java | 21 +-
docs/querying/query-context-reference.md | 11 +-
.../querykit/scan/ScanQueryFrameProcessorTest.java | 197 +++++-----
.../query/vector/VectorCursorGranularizer.java | 52 ++-
.../org/apache/druid/segment/BitmapOffset.java | 9 +-
.../druid/segment/QueryableIndexCursorHolder.java | 58 ++-
.../apache/druid/segment/data/ColumnarDoubles.java | 5 +-
.../apache/druid/segment/data/ColumnarFloats.java | 5 +-
.../apache/druid/segment/data/ColumnarLongs.java | 5 +-
.../nested/NestedFieldDictionaryEncodedColumn.java | 15 +-
.../druid/segment/nested/ScalarDoubleColumn.java | 5 +-
.../druid/segment/nested/ScalarLongColumn.java | 5 +-
.../apache/druid/segment/nested/VariantColumn.java | 5 +-
.../vector/DescendingBitmapVectorOffset.java | 154 ++++++++
.../vector/DescendingNoFilterVectorOffset.java | 107 +++++
.../druid/segment/vector/ReadableVectorOffset.java | 8 +-
.../vector/ReverseVectorColumnSelectorFactory.java | 434 +++++++++++++++++++++
.../apache/druid/query/QueryRunnerTestHelper.java | 65 ---
.../timeboundary/TimeBoundaryQueryRunnerTest.java | 255 +++++-------
.../timeseries/TimeseriesQueryRunnerTest.java | 4 +-
.../segment/DescendingVectorCursorDiffTest.java | 306 +++++++++++++++
.../java/org/apache/druid/segment/TestHelper.java | 10 -
.../segment/vector/DescendingVectorOffsetTest.java | 253 ++++++++++++
.../druid/sql/calcite/CalciteJoinQueryTest.java | 6 +-
.../apache/druid/sql/calcite/CalciteQueryTest.java | 6 -
.../sql/calcite/CalciteTimeBoundaryQueryTest.java | 2 +-
26 files changed, 1603 insertions(+), 400 deletions(-)
diff --git
a/benchmarks/src/test/java/org/apache/druid/benchmark/query/TimeseriesBenchmark.java
b/benchmarks/src/test/java/org/apache/druid/benchmark/query/TimeseriesBenchmark.java
index 0ddb941ea0b..54bfca56f86 100644
---
a/benchmarks/src/test/java/org/apache/druid/benchmark/query/TimeseriesBenchmark.java
+++
b/benchmarks/src/test/java/org/apache/druid/benchmark/query/TimeseriesBenchmark.java
@@ -28,6 +28,7 @@ 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.Sequence;
import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.math.expr.ExpressionProcessing;
import org.apache.druid.query.Druids;
import org.apache.druid.query.FinalizeResultsQueryRunner;
import org.apache.druid.query.Query;
@@ -99,7 +100,17 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
@State(Scope.Benchmark)
-@Fork(value = 1)
+@Fork(
+ value = 1,
+ jvmArgsAppend = {
+ "--add-exports=java.base/sun.nio.ch=ALL-UNNAMED",
+ "--add-opens=java.base/java.nio=ALL-UNNAMED",
+ "--add-opens=java.base/sun.nio.ch=ALL-UNNAMED",
+ "--add-opens=java.base/java.lang=ALL-UNNAMED",
+ "--add-opens=java.base/java.lang.reflect=ALL-UNNAMED",
+ "--add-opens=java.base/jdk.internal.ref=ALL-UNNAMED"
+ }
+)
@Warmup(iterations = 5)
@Measurement(iterations = 15)
public class TimeseriesBenchmark
@@ -113,6 +124,9 @@ public class TimeseriesBenchmark
@Param({"true", "false"})
private boolean descending;
+ @Param({"false", "true"})
+ private String vectorize;
+
@Param({"all", "hour"})
private String queryGranularity;
@@ -164,6 +178,7 @@ public class TimeseriesBenchmark
.intervals(intervalSpec)
.aggregators(queryAggs)
.descending(descending)
+ .context(Map.of("vectorize", vectorize))
.build();
basicQueries.put("A", queryA);
@@ -184,6 +199,7 @@ public class TimeseriesBenchmark
.intervals(intervalSpec)
.aggregators(queryAggs)
.descending(descending)
+ .context(Map.of("vectorize", vectorize))
.build();
basicQueries.put("timeFilterNumeric", timeFilterQuery);
@@ -204,6 +220,7 @@ public class TimeseriesBenchmark
.intervals(intervalSpec)
.aggregators(queryAggs)
.descending(descending)
+ .context(Map.of("vectorize", vectorize))
.build();
basicQueries.put("timeFilterAlphanumeric", timeFilterQuery);
@@ -221,6 +238,7 @@ public class TimeseriesBenchmark
.intervals(intervalSpec)
.aggregators(queryAggs)
.descending(descending)
+ .context(Map.of("vectorize", vectorize))
.build();
basicQueries.put("timeFilterByInterval", timeFilterQuery);
@@ -238,6 +256,7 @@ public class TimeseriesBenchmark
{
log.info("SETUP CALLED AT " + System.currentTimeMillis());
+ ExpressionProcessing.initializeForTests();
ComplexMetrics.registerSerde(HyperUniquesSerde.TYPE_NAME, new
HyperUniquesSerde());
setupQueries();
diff --git a/docs/querying/query-context-reference.md
b/docs/querying/query-context-reference.md
index d37d2b80799..3006869f175 100644
--- a/docs/querying/query-context-reference.md
+++ b/docs/querying/query-context-reference.md
@@ -107,9 +107,9 @@ query page.
## Vectorization parameters
-The GroupBy and Timeseries query types can run in _vectorized_ mode, which
speeds up query execution by processing
-batches of rows at a time. Not all queries can be vectorized. In particular,
vectorization currently has the following
-requirements:
+The GroupBy, Timeseries, TimeBoundary, and Scan (in MSQ only) query types can
run in _vectorized_ mode, which speeds up
+query execution by processing batches of rows at a time. Not all queries can
be vectorized. In particular, vectorization
+currently has the following requirements:
- All query-level filters must either be able to run on bitmap indexes or must
offer vectorized row-matchers. These
include `selector`, `bound`, `in`, `like`, `regex`, `search`, `and`, `or`, and
`not`.
@@ -120,16 +120,15 @@ include `selector`, `bound`, `in`, `like`, `regex`,
`search`, `and`, `or`, and `
- All virtual columns must offer vectorized implementations. Currently for
expression virtual columns, support for vectorization is decided on a per
expression basis, depending on the type of input and the functions used by the
expression. See the currently supported list in the [expression
documentation](math-expr.md#vectorization-support).
- For GroupBy: All dimension specs must be "default" (no extraction functions
or filtered dimension specs).
- For GroupBy: No multi-value dimensions.
-- For Timeseries: No "descending" order.
- Only immutable segments (not real-time).
- Only [table datasources](datasource.md#table) (not joins, subqueries,
lookups, or inline datasources).
-Other query types (like TopN, Scan, Select, and Search) ignore the `vectorize`
parameter, and will execute without
+Other query types (like TopN, Search, and native Scan) ignore the `vectorize`
parameter, and will execute without
vectorization. These query types will ignore the `vectorize` parameter even if
it is set to `"force"`.
|Parameter|Default| Description|
|---------|-------|------------|
-|`vectorize`|`true`|Enables or disables vectorized query execution. Possible
values are `false` (disabled), `true` (enabled if possible, disabled otherwise,
on a per-segment basis), and `force` (enabled, and groupBy or timeseries
queries that cannot be vectorized will fail). The `"force"` setting is meant to
aid in testing, and is not generally useful in production (since real-time
segments can never be processed with vectorized execution, any queries on
real-time data will fail). This w [...]
+|`vectorize`|`true`|Enables or disables vectorized query execution. Possible
values are `false` (disabled), `true` (enabled if possible, disabled otherwise,
on a per-segment basis), and `force` (enabled, and query types that support
vectorization will fail if they cannot be vectorized). The `"force"` setting is
meant to aid in testing, and is not generally useful in production (since
real-time segments can never be processed with vectorized execution, any
queries on real-time data will f [...]
|`vectorSize`|`512`|Sets the row batching size for a particular query. This
will override `druid.query.default.context.vectorSize` if it's set.|
|`vectorizeVirtualColumns`|`true`|Enables or disables vectorized query
processing of queries with virtual columns, layered on top of `vectorize`
(`vectorize` must also be set to true for a query to utilize vectorization).
Possible values are `false` (disabled), `true` (enabled if possible, disabled
otherwise, on a per-segment basis), and `force` (enabled, and groupBy or
timeseries queries with virtual columns that cannot be vectorized will fail).
The `"force"` setting is meant to aid in [...]
diff --git
a/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessorTest.java
b/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessorTest.java
index f3500915f96..d72b4bdfb30 100644
---
a/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessorTest.java
+++
b/multi-stage-query/src/test/java/org/apache/druid/msq/querykit/scan/ScanQueryFrameProcessorTest.java
@@ -20,6 +20,7 @@
package org.apache.druid.msq.querykit.scan;
import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
import com.google.common.util.concurrent.ListenableFuture;
import org.apache.druid.collections.ReferenceCountingResourceHolder;
import org.apache.druid.collections.ResourceHolder;
@@ -40,11 +41,14 @@ import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.Unit;
import org.apache.druid.java.util.common.guava.Sequence;
+import org.apache.druid.java.util.common.guava.Sequences;
import org.apache.druid.msq.querykit.FrameProcessorTestBase;
import org.apache.druid.msq.querykit.ReadableInput;
import org.apache.druid.msq.querykit.SegmentReferenceHolder;
import org.apache.druid.msq.test.LimitedFrameWriterFactory;
import org.apache.druid.query.Druids;
+import org.apache.druid.query.Order;
+import org.apache.druid.query.QueryContexts;
import org.apache.druid.query.policy.PolicyEnforcer;
import org.apache.druid.query.scan.ScanQuery;
import org.apache.druid.query.scan.ScanQueryEngine;
@@ -71,8 +75,6 @@ import org.junit.jupiter.api.Test;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
-
-import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -103,74 +105,44 @@ public class ScanQueryFrameProcessorTest extends
FrameProcessorTestBase
.columns(cursorFactory.getRowSignature().getColumnNames())
.build();
- final BlockingQueueFrameChannel outputChannel =
BlockingQueueFrameChannel.minimal();
-
- // Limit output frames to 1 row to ensure we test edge cases
- final FrameWriterFactory frameWriterFactory = new
LimitedFrameWriterFactory(
- FrameWriters.makeFrameWriterFactory(
- FrameType.latestRowBased(),
- new SingleMemoryAllocatorFactory(HeapMemoryAllocator.unlimited()),
- cursorFactory.getRowSignature(),
- Collections.emptyList(),
- false
- ),
- 1
- );
-
- final ReferenceCountedSegmentProvider segmentReferenceProvider =
- new ReferenceCountedSegmentProvider(new
QueryableIndexSegment(queryableIndex, SegmentId.dummy("test")));
- Assertions.assertEquals(0, segmentReferenceProvider.getNumReferences());
- final ScanQueryFrameProcessor processor = new ScanQueryFrameProcessor(
- query,
- null,
- new DefaultObjectMapper(),
- ReadableInput.segment(
- new SegmentReferenceHolder(
- new SegmentReference(
- SegmentId.dummy("test").toDescriptor(),
- segmentReferenceProvider.acquireReference(),
- null
- ),
- null
- )
- ),
- SegmentMapFunction.IDENTITY,
- new ResourceHolder<>()
- {
- @Override
- public WritableFrameChannel get()
- {
- return outputChannel.writable();
- }
-
- @Override
- public void close()
- {
- try {
- outputChannel.writable().close();
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- },
- new ReferenceCountingResourceHolder<>(frameWriterFactory, () -> {})
- );
-
- ListenableFuture<Object> retVal = exec.runFully(processor, null);
-
- final Sequence<List<Object>> rowsFromProcessor =
FrameTestUtil.readRowsFromFrameChannel(
- outputChannel.readable(),
- FrameReader.create(cursorFactory.getRowSignature())
- );
-
FrameTestUtil.assertRowsEqual(
FrameTestUtil.readRowsFromCursorFactory(cursorFactory,
cursorFactory.getRowSignature(), false),
- rowsFromProcessor
+ Sequences.simple(runScanOverSegment(queryableIndex, query))
);
+ }
- Assertions.assertEquals(Unit.instance(), retVal.get());
- Assertions.assertEquals(0, segmentReferenceProvider.getNumReferences());
// Segment reference was closed
+ @Test
+ public void test_runWithSegments_descendingMatchesNonVectorized() throws
Exception
+ {
+ final QueryableIndex queryableIndex = TestIndex.getMMappedTestIndex();
+
+ final Druids.ScanQueryBuilder baseBuilder =
+ Druids.newScanQueryBuilder()
+ .dataSource("test")
+ .intervals(new
MultipleIntervalSegmentSpec(ImmutableList.of(Intervals.of("2000-01-01T00Z/2030-01-01T00Z"))))
+ .columns("__time", "market", "quality", "index")
+ .order(Order.DESCENDING);
+
+ final ScanQuery nonVectorizedQuery =
+ baseBuilder.context(ImmutableMap.of(QueryContexts.VECTORIZE_KEY,
"false")).build();
+ final ScanQuery vectorizedQuery =
+ baseBuilder.context(ImmutableMap.of(QueryContexts.VECTORIZE_KEY,
"force", QueryContexts.VECTOR_SIZE_KEY, 7))
+ .build();
+
+ final List<List<Object>> nonVectorizedRows =
runScanOverSegment(queryableIndex, nonVectorizedQuery);
+ final List<List<Object>> vectorizedRows =
runScanOverSegment(queryableIndex, vectorizedQuery);
+
+ // The vectorized descending scan must match the non-vectorized descending
scan exactly.
+ Assertions.assertFalse(vectorizedRows.isEmpty());
+ Assertions.assertEquals(nonVectorizedRows, vectorizedRows);
+
+ // Sanity check: __time (column 0) is non-increasing.
+ long previousTime = Long.MAX_VALUE;
+ for (final List<Object> row : vectorizedRows) {
+ final long time = ((Number) row.getFirst()).longValue();
+ Assertions.assertTrue(time <= previousTime, "descending __time");
+ previousTime = time;
+ }
}
@Test
@@ -229,25 +201,7 @@ public class ScanQueryFrameProcessorTest extends
FrameProcessorTestBase
new DefaultObjectMapper(),
ReadableInput.channel(inputChannel.readable(),
FrameReader.create(signature), 0, 0),
SegmentMapFunction.IDENTITY,
- new ResourceHolder<>()
- {
- @Override
- public WritableFrameChannel get()
- {
- return outputChannel.writable();
- }
-
- @Override
- public void close()
- {
- try {
- outputChannel.writable().close();
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- },
+ ResourceHolder.fromCloseable(outputChannel.writable()),
new ReferenceCountingResourceHolder<>(frameWriterFactory, () -> {})
);
@@ -397,25 +351,7 @@ public class ScanQueryFrameProcessorTest extends
FrameProcessorTestBase
)
),
SegmentMapFunction.IDENTITY,
- new ResourceHolder<>()
- {
- @Override
- public WritableFrameChannel get()
- {
- return outputChannel.writable();
- }
-
- @Override
- public void close()
- {
- try {
- outputChannel.writable().close();
- }
- catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- },
+ ResourceHolder.fromCloseable(outputChannel.writable()),
new ReferenceCountingResourceHolder<>(frameWriterFactory, () -> {})
);
@@ -441,4 +377,57 @@ public class ScanQueryFrameProcessorTest extends
FrameProcessorTestBase
Assertions.assertEquals(Unit.instance(), retVal.get(30, TimeUnit.SECONDS));
}
+
+ private List<List<Object>> runScanOverSegment(final QueryableIndex
queryableIndex, final ScanQuery query)
+ throws Exception
+ {
+ final CursorFactory cursorFactory = new
QueryableIndexCursorFactory(queryableIndex);
+ final RowSignature signature = cursorFactory.getRowSignature();
+
+ final BlockingQueueFrameChannel outputChannel =
BlockingQueueFrameChannel.minimal();
+
+ // Limit output frames to 1 row to ensure we test edge cases
+ final FrameWriterFactory frameWriterFactory = new
LimitedFrameWriterFactory(
+ FrameWriters.makeFrameWriterFactory(
+ FrameType.latestRowBased(),
+ new SingleMemoryAllocatorFactory(HeapMemoryAllocator.unlimited()),
+ signature,
+ Collections.emptyList(),
+ false
+ ),
+ 1
+ );
+
+ final ReferenceCountedSegmentProvider segmentReferenceProvider =
+ new ReferenceCountedSegmentProvider(new
QueryableIndexSegment(queryableIndex, SegmentId.dummy("test")));
+ final ScanQueryFrameProcessor processor = new ScanQueryFrameProcessor(
+ query,
+ null,
+ new DefaultObjectMapper(),
+ ReadableInput.segment(
+ new SegmentReferenceHolder(
+ new SegmentReference(
+ SegmentId.dummy("test").toDescriptor(),
+ segmentReferenceProvider.acquireReference(),
+ null
+ ),
+ null
+ )
+ ),
+ SegmentMapFunction.IDENTITY,
+ ResourceHolder.fromCloseable(outputChannel.writable()),
+ new ReferenceCountingResourceHolder<>(frameWriterFactory, () -> {})
+ );
+
+ final ListenableFuture<Object> retVal = exec.runFully(processor, null);
+
+ final List<List<Object>> rows = FrameTestUtil.readRowsFromFrameChannel(
+ outputChannel.readable(),
+ FrameReader.create(signature)
+ ).toList();
+
+ Assertions.assertEquals(Unit.instance(), retVal.get());
+ Assertions.assertEquals(0, segmentReferenceProvider.getNumReferences());
// Segment reference was closed
+ return rows;
+ }
}
diff --git
a/processing/src/main/java/org/apache/druid/query/vector/VectorCursorGranularizer.java
b/processing/src/main/java/org/apache/druid/query/vector/VectorCursorGranularizer.java
index f262a4ea486..164be369001 100644
---
a/processing/src/main/java/org/apache/druid/query/vector/VectorCursorGranularizer.java
+++
b/processing/src/main/java/org/apache/druid/query/vector/VectorCursorGranularizer.java
@@ -19,7 +19,9 @@
package org.apache.druid.query.vector;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
import org.apache.druid.error.DruidException;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.granularity.Granularity;
@@ -75,7 +77,11 @@ public class VectorCursorGranularizer
return null;
}
- final Iterable<Interval> bucketIterable =
granularity.getIterable(clippedQueryInterval);
+ Iterable<Interval> bucketIterable =
granularity.getIterable(clippedQueryInterval);
+ if (timeOrder == Order.DESCENDING) {
+ // Descending cursors emit rows latest-first, so iterate the buckets
latest-first as well.
+ bucketIterable = Lists.reverse(ImmutableList.copyOf(bucketIterable));
+ }
final Interval firstBucket =
granularity.bucket(clippedQueryInterval.getStart());
final VectorValueSelector timeSelector;
@@ -88,7 +94,7 @@ public class VectorCursorGranularizer
timeSelector =
cursor.getColumnSelectorFactory().makeValueSelector(ColumnHolder.TIME_COLUMN_NAME);
}
- return new VectorCursorGranularizer(cursor, bucketIterable, timeSelector);
+ return new VectorCursorGranularizer(cursor, bucketIterable, timeSelector,
timeOrder == Order.DESCENDING);
}
private final VectorCursor cursor;
@@ -100,6 +106,9 @@ public class VectorCursorGranularizer
@Nullable
private final VectorValueSelector timeSelector;
+ // Whether the cursor (and hence the timestamps within each vector) is
time-descending.
+ private final boolean descending;
+
// Current time vector.
@Nullable
private long[] timestamps = null;
@@ -113,12 +122,14 @@ public class VectorCursorGranularizer
private VectorCursorGranularizer(
VectorCursor cursor,
Iterable<Interval> bucketIterable,
- @Nullable VectorValueSelector timeSelector
+ @Nullable VectorValueSelector timeSelector,
+ boolean descending
)
{
this.cursor = cursor;
this.bucketIterable = bucketIterable;
this.timeSelector = timeSelector;
+ this.descending = descending;
}
public void setCurrentOffsets(final Interval bucketInterval)
@@ -134,16 +145,31 @@ public class VectorCursorGranularizer
timestamps = timeSelector.getLongVector();
}
- // Skip "offset" to start of bucketInterval.
- while (startOffset < vectorSize && timestamps[startOffset] < timeStart) {
- startOffset++;
- }
-
- // Find end of bucketInterval.
- for (endOffset = vectorSize - 1;
- endOffset >= startOffset && timestamps[endOffset] >= timeEnd;
- endOffset--) {
- // nothing needed, "for" is doing the work.
+ if (descending) {
+ // Timestamps within the vector are descending. Skip rows at or after
the (exclusive) end of the bucket; these
+ // belong to later buckets, which have already been processed.
+ while (startOffset < vectorSize && timestamps[startOffset] >= timeEnd)
{
+ startOffset++;
+ }
+
+ // Find end of bucketInterval: the first row (from the back) whose
timestamp is still within the bucket.
+ for (endOffset = vectorSize - 1;
+ endOffset >= startOffset && timestamps[endOffset] < timeStart;
+ endOffset--) {
+ // nothing needed, "for" is doing the work.
+ }
+ } else {
+ // Skip "offset" to start of bucketInterval.
+ while (startOffset < vectorSize && timestamps[startOffset] <
timeStart) {
+ startOffset++;
+ }
+
+ // Find end of bucketInterval.
+ for (endOffset = vectorSize - 1;
+ endOffset >= startOffset && timestamps[endOffset] >= timeEnd;
+ endOffset--) {
+ // nothing needed, "for" is doing the work.
+ }
}
// Adjust: endOffset is now pointing at the last row to aggregate, but
we want it
diff --git
a/processing/src/main/java/org/apache/druid/segment/BitmapOffset.java
b/processing/src/main/java/org/apache/druid/segment/BitmapOffset.java
index 94324fe2206..8363a2cde04 100644
--- a/processing/src/main/java/org/apache/druid/segment/BitmapOffset.java
+++ b/processing/src/main/java/org/apache/druid/segment/BitmapOffset.java
@@ -28,6 +28,7 @@ import
org.apache.druid.query.monomorphicprocessing.RuntimeShapeInspector;
import org.apache.druid.segment.data.Offset;
import org.apache.druid.segment.data.ReadableOffset;
import org.roaringbitmap.IntIterator;
+import org.roaringbitmap.PeekableIntIterator;
import javax.annotation.Nullable;
import java.util.Arrays;
@@ -130,7 +131,13 @@ public class BitmapOffset extends Offset
private IntIterator iteratorForMark;
private int valueForMark;
- public static IntIterator getReverseBitmapOffsetIterator(ImmutableBitmap
bitmapIndex)
+ /**
+ * Returns an iterator over the set bits of "bitmapIndex", from highest to
lowest. The returned iterator's
+ * {@link PeekableIntIterator#advanceIfNeeded(int)} deviates from its
contract in order to remain useful:
+ * it advances as long as the next value is larger than the given value,
i.e., it is the reverse-order
+ * analog of the forward iterator's behavior.
+ */
+ public static PeekableIntIterator
getReverseBitmapOffsetIterator(ImmutableBitmap bitmapIndex)
{
ImmutableBitmap roaringBitmap = bitmapIndex;
if (!(bitmapIndex instanceof WrappedImmutableRoaringBitmap)) {
diff --git
a/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorHolder.java
b/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorHolder.java
index db5de51a663..d4b79cc95d4 100644
---
a/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorHolder.java
+++
b/processing/src/main/java/org/apache/druid/segment/QueryableIndexCursorHolder.java
@@ -24,6 +24,7 @@ import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
import com.google.common.collect.ImmutableList;
import org.apache.druid.collections.bitmap.BitmapFactory;
+import org.apache.druid.collections.bitmap.ImmutableBitmap;
import org.apache.druid.java.util.common.io.Closer;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.query.BaseQuery;
@@ -51,9 +52,12 @@ import org.apache.druid.segment.data.ReadableOffset;
import org.apache.druid.segment.filter.AndFilter;
import org.apache.druid.segment.historical.HistoricalCursor;
import org.apache.druid.segment.vector.BitmapVectorOffset;
+import org.apache.druid.segment.vector.DescendingBitmapVectorOffset;
+import org.apache.druid.segment.vector.DescendingNoFilterVectorOffset;
import org.apache.druid.segment.vector.FilteredVectorOffset;
import org.apache.druid.segment.vector.NoFilterVectorOffset;
import
org.apache.druid.segment.vector.QueryableIndexVectorColumnSelectorFactory;
+import org.apache.druid.segment.vector.ReverseVectorColumnSelectorFactory;
import org.apache.druid.segment.vector.VectorColumnSelectorFactory;
import org.apache.druid.segment.vector.VectorCursor;
import org.apache.druid.segment.vector.VectorOffset;
@@ -171,8 +175,9 @@ public class QueryableIndexCursorHolder implements
CursorHolder
}
}
- // vector cursors can't iterate backwards yet
- return Cursors.getTimeOrdering(ordering) != Order.DESCENDING;
+ // Descending time order is handled by iterating the underlying offset
back-to-front and reversing the decoded
+ // vectors via ReverseVectorColumnSelectorFactory; see asVectorCursor.
+ return true;
}
@Override
@@ -296,38 +301,65 @@ public class QueryableIndexCursorHolder implements
CursorHolder
endOffset = index.getNumRows();
}
+ // For descending time order, the offset iterates the [startOffset,
endOffset) range back-to-front (one batch at a
+ // time, ascending within each batch), and the column selector factory
returned to the caller is wrapped in a
+ // ReverseVectorColumnSelectorFactory that flips each batch's decoded
values. This keeps the low-level column
+ // readers seeing only ascending offsets.
+ final boolean descending = timeOrder == Order.DESCENDING;
+
// filterBundle will only be null if the filter itself is null, otherwise
check to see if the filter can use
// an index
- final VectorOffset baseOffset =
- filterBundle == null || filterBundle.getIndex() == null
- ? new NoFilterVectorOffset(vectorSize, startOffset, endOffset)
- : new BitmapVectorOffset(vectorSize,
filterBundle.getIndex().getBitmap(), startOffset, endOffset);
+ final VectorOffset baseOffset;
+ if (filterBundle == null || filterBundle.getIndex() == null) {
+ baseOffset =
+ descending
+ ? new DescendingNoFilterVectorOffset(vectorSize, startOffset,
endOffset)
+ : new NoFilterVectorOffset(vectorSize, startOffset, endOffset);
+ } else {
+ final ImmutableBitmap bitmap = filterBundle.getIndex().getBitmap();
+ baseOffset =
+ descending
+ ? new DescendingBitmapVectorOffset(vectorSize, bitmap, startOffset,
endOffset)
+ : new BitmapVectorOffset(vectorSize, bitmap, startOffset, endOffset);
+ }
- // baseColumnSelectorFactory using baseOffset is the column selector for
filtering.
+ // baseColumnSelectorFactory using baseOffset is the column selector for
filtering. Filtering is order-independent,
+ // so it always reads ascending (unreversed) offsets.
final VectorColumnSelectorFactory baseColumnSelectorFactory =
makeVectorColumnSelectorFactoryForOffset(
columnCache,
baseOffset
);
+ final VectorOffset cursorOffset;
+ final VectorColumnSelectorFactory cursorColumnSelectorFactory;
+
// filterBundle will only be null if the filter itself is null, otherwise
check to see if the filter needs to use
// a value matcher
if (filterBundle != null && filterBundle.getMatcherBundle() != null) {
final VectorValueMatcher vectorValueMatcher =
filterBundle.getMatcherBundle()
.vectorMatcher(baseColumnSelectorFactory, baseOffset);
- final VectorOffset filteredOffset = FilteredVectorOffset.create(
+ cursorOffset = FilteredVectorOffset.create(
baseOffset,
vectorValueMatcher
);
- // Now create the cursor and column selector that will be returned to
the caller.
- final VectorColumnSelectorFactory filteredColumnSelectorFactory =
makeVectorColumnSelectorFactoryForOffset(
+ // Now create the column selector that will be returned to the caller.
+ cursorColumnSelectorFactory = makeVectorColumnSelectorFactoryForOffset(
columnCache,
- filteredOffset
+ cursorOffset
);
- return new QueryableIndexVectorCursor(filteredColumnSelectorFactory,
filteredOffset, vectorSize);
} else {
- return new QueryableIndexVectorCursor(baseColumnSelectorFactory,
baseOffset, vectorSize);
+ cursorOffset = baseOffset;
+ cursorColumnSelectorFactory = baseColumnSelectorFactory;
}
+
+ return new QueryableIndexVectorCursor(
+ descending
+ ? new ReverseVectorColumnSelectorFactory(cursorColumnSelectorFactory)
+ : cursorColumnSelectorFactory,
+ cursorOffset,
+ vectorSize
+ );
}
@Override
diff --git
a/processing/src/main/java/org/apache/druid/segment/data/ColumnarDoubles.java
b/processing/src/main/java/org/apache/druid/segment/data/ColumnarDoubles.java
index ef731c36b3e..d0764ba492f 100644
---
a/processing/src/main/java/org/apache/druid/segment/data/ColumnarDoubles.java
+++
b/processing/src/main/java/org/apache/druid/segment/data/ColumnarDoubles.java
@@ -202,10 +202,11 @@ public interface ColumnarDoubles extends Closeable
ColumnarDoubles.this.get(doubleVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- if (offsets[offsets.length - 1] < offsetMark) {
+ final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
+ if (maxOffset < offsetMark) {
nullIterator = nullValueBitmap.peekableIterator();
}
- offsetMark = offsets[offsets.length - 1];
+ offsetMark = maxOffset;
ColumnarDoubles.this.get(doubleVector, offsets,
offset.getCurrentVectorSize());
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/data/ColumnarFloats.java
b/processing/src/main/java/org/apache/druid/segment/data/ColumnarFloats.java
index ba9e99e5429..1276f3294e7 100644
--- a/processing/src/main/java/org/apache/druid/segment/data/ColumnarFloats.java
+++ b/processing/src/main/java/org/apache/druid/segment/data/ColumnarFloats.java
@@ -197,10 +197,11 @@ public interface ColumnarFloats extends Closeable
ColumnarFloats.this.get(floatVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- if (offsets[offsets.length - 1] < offsetMark) {
+ final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
+ if (maxOffset < offsetMark) {
nullIterator = nullValueBitmap.peekableIterator();
}
- offsetMark = offsets[offsets.length - 1];
+ offsetMark = maxOffset;
ColumnarFloats.this.get(floatVector, offsets,
offset.getCurrentVectorSize());
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/data/ColumnarLongs.java
b/processing/src/main/java/org/apache/druid/segment/data/ColumnarLongs.java
index bb029b3c645..0c672195e8b 100644
--- a/processing/src/main/java/org/apache/druid/segment/data/ColumnarLongs.java
+++ b/processing/src/main/java/org/apache/druid/segment/data/ColumnarLongs.java
@@ -208,10 +208,11 @@ public interface ColumnarLongs extends Closeable
ColumnarLongs.this.get(longVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- if (offsets[offsets.length - 1] < offsetMark) {
+ final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
+ if (maxOffset < offsetMark) {
nullIterator = nullValueBitmap.peekableIterator();
}
- offsetMark = offsets[offsets.length - 1];
+ offsetMark = maxOffset;
ColumnarLongs.this.get(longVector, offsets,
offset.getCurrentVectorSize());
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/nested/NestedFieldDictionaryEncodedColumn.java
b/processing/src/main/java/org/apache/druid/segment/nested/NestedFieldDictionaryEncodedColumn.java
index 261da5ddfc7..43a2080d7a6 100644
---
a/processing/src/main/java/org/apache/druid/segment/nested/NestedFieldDictionaryEncodedColumn.java
+++
b/processing/src/main/java/org/apache/druid/segment/nested/NestedFieldDictionaryEncodedColumn.java
@@ -912,10 +912,11 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
longsColumn.get(valueVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- if (offsets[offsets.length - 1] < offsetMark) {
+ final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
+ if (maxOffset < offsetMark) {
nullIterator = nullBitmap.peekableIterator();
}
- offsetMark = offsets[offsets.length - 1];
+ offsetMark = maxOffset;
longsColumn.get(valueVector, offsets,
offset.getCurrentVectorSize());
}
@@ -967,10 +968,11 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
doublesColumn.get(valueVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- if (offsets[offsets.length - 1] < offsetMark) {
+ final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
+ if (maxOffset < offsetMark) {
nullIterator = nullBitmap.peekableIterator();
}
- offsetMark = offsets[offsets.length - 1];
+ offsetMark = maxOffset;
doublesColumn.get(valueVector, offsets,
offset.getCurrentVectorSize());
}
@@ -1026,10 +1028,11 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
column.get(idVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- if (offsets[offsets.length - 1] < offsetMark) {
+ final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
+ if (maxOffset < offsetMark) {
nullIterator = nullBitmap.peekableIterator();
}
- offsetMark = offsets[offsets.length - 1];
+ offsetMark = maxOffset;
column.get(idVector, offsets, offset.getCurrentVectorSize());
}
for (int i = 0; i < offset.getCurrentVectorSize(); i++) {
diff --git
a/processing/src/main/java/org/apache/druid/segment/nested/ScalarDoubleColumn.java
b/processing/src/main/java/org/apache/druid/segment/nested/ScalarDoubleColumn.java
index c6c7137e0cf..4252cf56342 100644
---
a/processing/src/main/java/org/apache/druid/segment/nested/ScalarDoubleColumn.java
+++
b/processing/src/main/java/org/apache/druid/segment/nested/ScalarDoubleColumn.java
@@ -170,10 +170,11 @@ public class ScalarDoubleColumn implements
NestedCommonFormatColumn
valueColumn.get(valueVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- if (offsets[offsets.length - 1] < offsetMark) {
+ final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
+ if (maxOffset < offsetMark) {
nullIterator = nullValueIndex.peekableIterator();
}
- offsetMark = offsets[offsets.length - 1];
+ offsetMark = maxOffset;
valueColumn.get(valueVector, offsets, offset.getCurrentVectorSize());
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/nested/ScalarLongColumn.java
b/processing/src/main/java/org/apache/druid/segment/nested/ScalarLongColumn.java
index 3d7ccd458be..9fa57b7c731 100644
---
a/processing/src/main/java/org/apache/druid/segment/nested/ScalarLongColumn.java
+++
b/processing/src/main/java/org/apache/druid/segment/nested/ScalarLongColumn.java
@@ -171,10 +171,11 @@ public class ScalarLongColumn implements
NestedCommonFormatColumn
valueColumn.get(valueVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- if (offsets[offsets.length - 1] < offsetMark) {
+ final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
+ if (maxOffset < offsetMark) {
nullIterator = nullValueIndex.peekableIterator();
}
- offsetMark = offsets[offsets.length - 1];
+ offsetMark = maxOffset;
valueColumn.get(valueVector, offsets, offset.getCurrentVectorSize());
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/nested/VariantColumn.java
b/processing/src/main/java/org/apache/druid/segment/nested/VariantColumn.java
index 05470edae96..1bbf3ffe26e 100644
---
a/processing/src/main/java/org/apache/druid/segment/nested/VariantColumn.java
+++
b/processing/src/main/java/org/apache/druid/segment/nested/VariantColumn.java
@@ -829,10 +829,11 @@ public class VariantColumn<TStringDictionary extends
Indexed<ByteBuffer>>
encodedValueColumn.get(idVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- if (offsets[offsets.length - 1] < offsetMark) {
+ final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
+ if (maxOffset < offsetMark) {
nullIterator = nullValueBitmap.peekableIterator();
}
- offsetMark = offsets[offsets.length - 1];
+ offsetMark = maxOffset;
encodedValueColumn.get(idVector, offsets,
offset.getCurrentVectorSize());
}
for (int i = 0; i < offset.getCurrentVectorSize(); i++) {
diff --git
a/processing/src/main/java/org/apache/druid/segment/vector/DescendingBitmapVectorOffset.java
b/processing/src/main/java/org/apache/druid/segment/vector/DescendingBitmapVectorOffset.java
new file mode 100644
index 00000000000..fca49fc8d15
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/vector/DescendingBitmapVectorOffset.java
@@ -0,0 +1,154 @@
+/*
+ * 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.segment.vector;
+
+import com.google.common.base.Preconditions;
+import org.apache.druid.collections.bitmap.ImmutableBitmap;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.segment.BitmapOffset;
+import org.roaringbitmap.PeekableIntIterator;
+
+import javax.annotation.Nullable;
+
+/**
+ * Like {@link BitmapVectorOffset}, but in descending order.
+ *
+ * <p>As required by {@link ReadableVectorOffset#getOffsets()}, the offsets
within each batch are in internally
+ * ascending order.
+ *
+ * @see DescendingNoFilterVectorOffset the no-filter version
+ */
+public class DescendingBitmapVectorOffset implements VectorOffset
+{
+ private final ImmutableBitmap bitmap;
+ private final int[] offsets;
+ private final int startOffset;
+ private final int endOffset;
+
+ // Null when [startOffset, endOffset) is empty, in which case this offset is
immediately done.
+ @Nullable
+ private PeekableIntIterator iterator;
+ private int currentVectorSize;
+ private boolean isContiguous;
+
+ public DescendingBitmapVectorOffset(
+ final int vectorSize,
+ final ImmutableBitmap bitmap,
+ final int startOffset,
+ final int endOffset
+ )
+ {
+ this.bitmap = bitmap;
+ this.offsets = new int[vectorSize];
+ this.startOffset = startOffset;
+ this.endOffset = endOffset;
+ reset();
+ }
+
+ @Override
+ public int getId()
+ {
+ Preconditions.checkState(currentVectorSize > 0, "currentVectorSize > 0");
+ return offsets[0];
+ }
+
+ @Override
+ public void advance()
+ {
+ // The iterator yields offsets from highest to lowest, so fill the array
back-to-front (internally-ascending order).
+ int i = offsets.length;
+ while (i > 0 && iterator != null && iterator.hasNext() &&
iterator.peekNext() >= startOffset) {
+ offsets[--i] = iterator.next();
+ }
+
+ currentVectorSize = offsets.length - i;
+
+ if (i > 0 && currentVectorSize > 0) {
+ // Partial batch: move it to the front of the array.
+ System.arraycopy(offsets, i, offsets, 0, currentVectorSize);
+ }
+
+ isContiguous = false;
+ if (currentVectorSize > 1) {
+ final int hiPos = currentVectorSize - 1;
+ isContiguous = offsets[hiPos] - offsets[0] == hiPos;
+ }
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return currentVectorSize == 0;
+ }
+
+ @Override
+ public boolean isContiguous()
+ {
+ return isContiguous;
+ }
+
+ @Override
+ public int getMaxVectorSize()
+ {
+ return offsets.length;
+ }
+
+ @Override
+ public int getCurrentVectorSize()
+ {
+ return currentVectorSize;
+ }
+
+ @Override
+ public int getStartOffset()
+ {
+ if (isContiguous) {
+ return offsets[0];
+ } else {
+ throw DruidException.defensive("Cannot call getStartOffset() on a
non-contiguous offset");
+ }
+ }
+
+ @Override
+ public int[] getOffsets()
+ {
+ if (!isContiguous) {
+ return offsets;
+ } else {
+ throw DruidException.defensive("Cannot call getOffsets() on a contiguous
offset");
+ }
+ }
+
+ @Override
+ public void reset()
+ {
+ currentVectorSize = 0;
+ isContiguous = false;
+
+ if (startOffset < endOffset) {
+ iterator = BitmapOffset.getReverseBitmapOffsetIterator(bitmap);
+ iterator.advanceIfNeeded(endOffset - 1);
+ advance();
+ } else {
+ // Empty range.
+ iterator = null;
+ }
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/segment/vector/DescendingNoFilterVectorOffset.java
b/processing/src/main/java/org/apache/druid/segment/vector/DescendingNoFilterVectorOffset.java
new file mode 100644
index 00000000000..da46d11662a
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/vector/DescendingNoFilterVectorOffset.java
@@ -0,0 +1,107 @@
+/*
+ * 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.segment.vector;
+
+/**
+ * Like {@link NoFilterVectorOffset}, but in descending order.
+ *
+ * <p>As required by {@link ReadableVectorOffset#getOffsets()}, the offsets
within each batch are in internally
+ * ascending order.
+ */
+public class DescendingNoFilterVectorOffset implements VectorOffset
+{
+ private final int maxVectorSize;
+ private final int start;
+ private final int end;
+
+ private int currentStart;
+ private int currentEnd;
+
+ public DescendingNoFilterVectorOffset(final int maxVectorSize, final int
start, final int end)
+ {
+ this.maxVectorSize = maxVectorSize;
+ this.start = start;
+ this.end = end;
+ reset();
+ }
+
+ @Override
+ public int getId()
+ {
+ return currentStart;
+ }
+
+ @Override
+ public void advance()
+ {
+ currentEnd = currentStart;
+ currentStart = computeCurrentStart();
+ }
+
+ @Override
+ public boolean isDone()
+ {
+ return currentEnd <= start;
+ }
+
+ @Override
+ public boolean isContiguous()
+ {
+ return true;
+ }
+
+ @Override
+ public int getMaxVectorSize()
+ {
+ return maxVectorSize;
+ }
+
+ @Override
+ public int getCurrentVectorSize()
+ {
+ return currentEnd - currentStart;
+ }
+
+ @Override
+ public int getStartOffset()
+ {
+ return currentStart;
+ }
+
+ @Override
+ public int[] getOffsets()
+ {
+ throw new UnsupportedOperationException("no filter");
+ }
+
+ @Override
+ public void reset()
+ {
+ currentEnd = end;
+ currentStart = computeCurrentStart();
+ }
+
+ private int computeCurrentStart()
+ {
+ // Align batch boundaries down to a multiple of maxVectorSize, counting
from the beginning of the segment. When
+ // maxVectorSize divides the compressed block size, this prevents batch
boundaries from crossing block boundaries.
+ return Math.max(start, (currentEnd - 1) / maxVectorSize * maxVectorSize);
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/segment/vector/ReadableVectorOffset.java
b/processing/src/main/java/org/apache/druid/segment/vector/ReadableVectorOffset.java
index 35cb0059946..9076e099eca 100644
---
a/processing/src/main/java/org/apache/druid/segment/vector/ReadableVectorOffset.java
+++
b/processing/src/main/java/org/apache/druid/segment/vector/ReadableVectorOffset.java
@@ -47,9 +47,13 @@ public interface ReadableVectorOffset extends
ReadableVectorInspector
/**
* If "isContiguous" is false, this method returns a batch of offsets. The
array may be longer than the number of
- * valid offsets, so callers need to check "getCurrentVectorSize" too.
+ * valid offsets, so callers need to check "getCurrentVectorSize" too.
Entries at or past "getCurrentVectorSize" are
+ * not meaningful and must not be read.
*
- * Throws an exception if "isContiguous" is true.
+ * <p>Offsets in this array are always ascending, even if the cursor is
descending
+ * (see {@link ReverseVectorColumnSelectorFactory}).
+ *
+ * <p>Throws an exception if "isContiguous" is true.
*/
int[] getOffsets();
diff --git
a/processing/src/main/java/org/apache/druid/segment/vector/ReverseVectorColumnSelectorFactory.java
b/processing/src/main/java/org/apache/druid/segment/vector/ReverseVectorColumnSelectorFactory.java
new file mode 100644
index 00000000000..086fe44afdb
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/segment/vector/ReverseVectorColumnSelectorFactory.java
@@ -0,0 +1,434 @@
+/*
+ * 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.segment.vector;
+
+import org.apache.druid.error.NotYetImplemented;
+import org.apache.druid.query.dimension.DimensionSpec;
+import org.apache.druid.query.groupby.DeferExpressionDimensions;
+import
org.apache.druid.query.groupby.epinephelinae.vector.GroupByVectorColumnSelector;
+import org.apache.druid.segment.DimensionDictionarySelector;
+import org.apache.druid.segment.IdLookup;
+import org.apache.druid.segment.column.ColumnCapabilities;
+import org.apache.druid.segment.data.IndexedInts;
+
+import javax.annotation.Nullable;
+import java.nio.ByteBuffer;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Wraps a {@link VectorColumnSelectorFactory} whose underlying offset
iterates batches in descending order (see
+ * {@link DescendingNoFilterVectorOffset} and {@link
DescendingBitmapVectorOffset}). The underlying offset generates
+ * batches in descending order (from the end of the segment to the start), but
each batch is internally in ascending
+ * order. This class reverses each batch internally, to provide an overall
descending order.
+ *
+ * <p>Reversing is done here (on already-decoded vectors) so that low-level
column readers only ever need to return
+ * batches in internally-ascending order. This allows them to be simpler, as
they do not need to have handling for
+ * both ascending and descending order.
+ */
+public class ReverseVectorColumnSelectorFactory implements
VectorColumnSelectorFactory
+{
+ private final VectorColumnSelectorFactory delegate;
+ private final ReadableVectorInspector inspector;
+
+ private final Map<DimensionSpec, SingleValueDimensionVectorSelector>
singleValueDimensionSelectorCache = new HashMap<>();
+ private final Map<DimensionSpec, MultiValueDimensionVectorSelector>
multiValueDimensionSelectorCache = new HashMap<>();
+ private final Map<String, VectorValueSelector> valueSelectorCache = new
HashMap<>();
+ private final Map<String, VectorObjectSelector> objectSelectorCache = new
HashMap<>();
+
+ public ReverseVectorColumnSelectorFactory(final VectorColumnSelectorFactory
delegate)
+ {
+ this.delegate = delegate;
+ this.inspector = delegate.getReadableVectorInspector();
+ }
+
+ @Override
+ public ReadableVectorInspector getReadableVectorInspector()
+ {
+ return inspector;
+ }
+
+ @Override
+ public SingleValueDimensionVectorSelector
makeSingleValueDimensionSelector(final DimensionSpec dimensionSpec)
+ {
+ return singleValueDimensionSelectorCache.computeIfAbsent(
+ dimensionSpec,
+ spec -> new ReverseSingleValueDimensionVectorSelector(
+ delegate.makeSingleValueDimensionSelector(spec),
+ inspector
+ )
+ );
+ }
+
+ @Override
+ public MultiValueDimensionVectorSelector
makeMultiValueDimensionSelector(final DimensionSpec dimensionSpec)
+ {
+ return multiValueDimensionSelectorCache.computeIfAbsent(
+ dimensionSpec,
+ spec -> new
ReverseMultiValueDimensionVectorSelector(delegate.makeMultiValueDimensionSelector(spec),
inspector)
+ );
+ }
+
+ @Override
+ public VectorValueSelector makeValueSelector(final String column)
+ {
+ return valueSelectorCache.computeIfAbsent(
+ column,
+ c -> new ReverseVectorValueSelector(delegate.makeValueSelector(c),
inspector)
+ );
+ }
+
+ @Override
+ public VectorObjectSelector makeObjectSelector(final String column)
+ {
+ return objectSelectorCache.computeIfAbsent(
+ column,
+ c -> new ReverseVectorObjectSelector(delegate.makeObjectSelector(c),
inspector)
+ );
+ }
+
+ @Nullable
+ @Override
+ public ColumnCapabilities getColumnCapabilities(final String column)
+ {
+ return delegate.getColumnCapabilities(column);
+ }
+
+ /**
+ * Reverse n items of src into dst.
+ */
+ private static void reverseInto(final long[] src, final long[] dst, final
int n)
+ {
+ for (int i = 0; i < n; i++) {
+ dst[i] = src[n - 1 - i];
+ }
+ }
+
+ /**
+ * Reverse n items of src into dst.
+ */
+ private static void reverseInto(final float[] src, final float[] dst, final
int n)
+ {
+ for (int i = 0; i < n; i++) {
+ dst[i] = src[n - 1 - i];
+ }
+ }
+
+ /**
+ * Reverse n items of src into dst.
+ */
+ private static void reverseInto(final double[] src, final double[] dst,
final int n)
+ {
+ for (int i = 0; i < n; i++) {
+ dst[i] = src[n - 1 - i];
+ }
+ }
+
+ /**
+ * Reverse n items of src into dst.
+ */
+ private static void reverseInto(final boolean[] src, final boolean[] dst,
final int n)
+ {
+ for (int i = 0; i < n; i++) {
+ dst[i] = src[n - 1 - i];
+ }
+ }
+
+ /**
+ * Reverse n items of src into dst.
+ */
+ private static void reverseInto(final int[] src, final int[] dst, final int
n)
+ {
+ for (int i = 0; i < n; i++) {
+ dst[i] = src[n - 1 - i];
+ }
+ }
+
+ /**
+ * Reverse n items of src into dst.
+ */
+ private static void reverseInto(final Object[] src, final Object[] dst,
final int n)
+ {
+ for (int i = 0; i < n; i++) {
+ dst[i] = src[n - 1 - i];
+ }
+ }
+
+ /**
+ * Base class for the reversing selectors. Each one reverses the valid
entries of a freshly-decoded vector into a
+ * scratch array, caching by vector id so that a given batch is only
reversed once.
+ */
+ private abstract static class ReverseVectorSelector implements
VectorSizeInspector
+ {
+ final ReadableVectorInspector inspector;
+
+ ReverseVectorSelector(final ReadableVectorInspector inspector)
+ {
+ this.inspector = inspector;
+ }
+
+ @Override
+ public int getMaxVectorSize()
+ {
+ return inspector.getMaxVectorSize();
+ }
+
+ @Override
+ public int getCurrentVectorSize()
+ {
+ return inspector.getCurrentVectorSize();
+ }
+ }
+
+ /**
+ * Base class for the reversing dimension selectors, which delegate all the
dictionary lookups.
+ */
+ private abstract static class ReverseDimensionVectorSelector extends
ReverseVectorSelector
+ implements DimensionDictionarySelector
+ {
+ private final DimensionDictionarySelector delegate;
+
+ ReverseDimensionVectorSelector(final DimensionDictionarySelector delegate,
final ReadableVectorInspector inspector)
+ {
+ super(inspector);
+ this.delegate = delegate;
+ }
+
+ @Override
+ public int getValueCardinality()
+ {
+ return delegate.getValueCardinality();
+ }
+
+ @Nullable
+ @Override
+ public String lookupName(final int id)
+ {
+ return delegate.lookupName(id);
+ }
+
+ @Nullable
+ @Override
+ public ByteBuffer lookupNameUtf8(final int id)
+ {
+ return delegate.lookupNameUtf8(id);
+ }
+
+ @Override
+ public boolean supportsLookupNameUtf8()
+ {
+ return delegate.supportsLookupNameUtf8();
+ }
+
+ @Override
+ public boolean nameLookupPossibleInAdvance()
+ {
+ return delegate.nameLookupPossibleInAdvance();
+ }
+
+ @Nullable
+ @Override
+ public IdLookup idLookup()
+ {
+ return delegate.idLookup();
+ }
+ }
+
+ private static class ReverseVectorValueSelector extends
ReverseVectorSelector implements VectorValueSelector
+ {
+ private final VectorValueSelector delegate;
+
+ @Nullable
+ private long[] longs;
+ @Nullable
+ private float[] floats;
+ @Nullable
+ private double[] doubles;
+ @Nullable
+ private boolean[] nulls;
+
+ private int longId = ReadableVectorInspector.NULL_ID;
+ private int floatId = ReadableVectorInspector.NULL_ID;
+ private int doubleId = ReadableVectorInspector.NULL_ID;
+ private int nullId = ReadableVectorInspector.NULL_ID;
+
+ // Null when the current batch has no nulls at all, in which case
getNullVector returns null.
+ @Nullable
+ private boolean[] currentNulls;
+
+ ReverseVectorValueSelector(final VectorValueSelector delegate, final
ReadableVectorInspector inspector)
+ {
+ super(inspector);
+ this.delegate = delegate;
+ }
+
+ @Override
+ public long[] getLongVector()
+ {
+ if (longs == null) {
+ longs = new long[inspector.getMaxVectorSize()];
+ }
+ if (longId != inspector.getId()) {
+ reverseInto(delegate.getLongVector(), longs,
inspector.getCurrentVectorSize());
+ longId = inspector.getId();
+ }
+ return longs;
+ }
+
+ @Override
+ public float[] getFloatVector()
+ {
+ if (floats == null) {
+ floats = new float[inspector.getMaxVectorSize()];
+ }
+ if (floatId != inspector.getId()) {
+ reverseInto(delegate.getFloatVector(), floats,
inspector.getCurrentVectorSize());
+ floatId = inspector.getId();
+ }
+ return floats;
+ }
+
+ @Override
+ public double[] getDoubleVector()
+ {
+ if (doubles == null) {
+ doubles = new double[inspector.getMaxVectorSize()];
+ }
+ if (doubleId != inspector.getId()) {
+ reverseInto(delegate.getDoubleVector(), doubles,
inspector.getCurrentVectorSize());
+ doubleId = inspector.getId();
+ }
+ return doubles;
+ }
+
+ @Nullable
+ @Override
+ public boolean[] getNullVector()
+ {
+ if (nullId != inspector.getId()) {
+ final boolean[] src = delegate.getNullVector();
+ if (src == null) {
+ currentNulls = null;
+ } else {
+ if (nulls == null) {
+ nulls = new boolean[inspector.getMaxVectorSize()];
+ }
+ reverseInto(src, nulls, inspector.getCurrentVectorSize());
+ currentNulls = nulls;
+ }
+ nullId = inspector.getId();
+ }
+ return currentNulls;
+ }
+ }
+
+ @Override
+ public GroupByVectorColumnSelector makeGroupByVectorColumnSelector(
+ final String column,
+ final DeferExpressionDimensions deferExpressionDimensions
+ )
+ {
+ // groupBy does not use descending cursors, so this method is not needed.
+ throw NotYetImplemented.ex(null, "makeGroupByVectorColumnSelector is not
needed for descending cursors");
+ }
+
+ private static class ReverseVectorObjectSelector extends
ReverseVectorSelector implements VectorObjectSelector
+ {
+ private final VectorObjectSelector delegate;
+ private final Object[] objects;
+
+ private int id = ReadableVectorInspector.NULL_ID;
+
+ ReverseVectorObjectSelector(final VectorObjectSelector delegate, final
ReadableVectorInspector inspector)
+ {
+ super(inspector);
+ this.delegate = delegate;
+ this.objects = new Object[inspector.getMaxVectorSize()];
+ }
+
+ @Override
+ public Object[] getObjectVector()
+ {
+ if (id != inspector.getId()) {
+ reverseInto(delegate.getObjectVector(), objects,
inspector.getCurrentVectorSize());
+ id = inspector.getId();
+ }
+ return objects;
+ }
+ }
+
+ private static class ReverseSingleValueDimensionVectorSelector extends
ReverseDimensionVectorSelector
+ implements SingleValueDimensionVectorSelector
+ {
+ private final SingleValueDimensionVectorSelector delegate;
+ private final int[] rows;
+
+ private int id = ReadableVectorInspector.NULL_ID;
+
+ ReverseSingleValueDimensionVectorSelector(
+ final SingleValueDimensionVectorSelector delegate,
+ final ReadableVectorInspector inspector
+ )
+ {
+ super(delegate, inspector);
+ this.delegate = delegate;
+ this.rows = new int[inspector.getMaxVectorSize()];
+ }
+
+ @Override
+ public int[] getRowVector()
+ {
+ if (id != inspector.getId()) {
+ reverseInto(delegate.getRowVector(), rows,
inspector.getCurrentVectorSize());
+ id = inspector.getId();
+ }
+ return rows;
+ }
+ }
+
+ private static class ReverseMultiValueDimensionVectorSelector extends
ReverseDimensionVectorSelector
+ implements MultiValueDimensionVectorSelector
+ {
+ private final MultiValueDimensionVectorSelector delegate;
+ private final IndexedInts[] rows;
+
+ private int id = ReadableVectorInspector.NULL_ID;
+
+ ReverseMultiValueDimensionVectorSelector(
+ final MultiValueDimensionVectorSelector delegate,
+ final ReadableVectorInspector inspector
+ )
+ {
+ super(delegate, inspector);
+ this.delegate = delegate;
+ this.rows = new IndexedInts[inspector.getMaxVectorSize()];
+ }
+
+ @Override
+ public IndexedInts[] getRowVector()
+ {
+ if (id != inspector.getId()) {
+ reverseInto(delegate.getRowVector(), rows,
inspector.getCurrentVectorSize());
+ id = inspector.getId();
+ }
+ return rows;
+ }
+ }
+}
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 2f17b310b67..a204bdcc2f7 100644
--- a/processing/src/test/java/org/apache/druid/query/QueryRunnerTestHelper.java
+++ b/processing/src/test/java/org/apache/druid/query/QueryRunnerTestHelper.java
@@ -28,10 +28,6 @@ import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
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;
-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.io.Closer;
import org.apache.druid.js.JavaScriptConfig;
import org.apache.druid.math.expr.ExprMacroTable;
import org.apache.druid.query.aggregation.AggregatorFactory;
@@ -58,21 +54,15 @@ import org.apache.druid.query.planning.ExecutionVertex;
import org.apache.druid.query.policy.NoopPolicyEnforcer;
import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
import org.apache.druid.query.spec.QuerySegmentSpec;
-import org.apache.druid.query.spec.SpecificSegmentSpec;
import org.apache.druid.query.timeseries.TimeseriesQueryEngine;
import org.apache.druid.query.timeseries.TimeseriesQueryQueryToolChest;
import org.apache.druid.query.timeseries.TimeseriesQueryRunnerFactory;
import org.apache.druid.segment.IncrementalIndexSegment;
import org.apache.druid.segment.QueryableIndexSegment;
-import org.apache.druid.segment.ReferenceCountedSegmentProvider;
import org.apache.druid.segment.Segment;
import org.apache.druid.segment.TestIndex;
import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
-import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
-import org.apache.druid.timeline.TimelineObjectHolder;
-import org.apache.druid.timeline.VersionedIntervalTimeline;
-import org.apache.druid.utils.CloseableUtils;
import org.joda.time.DateTime;
import org.joda.time.Interval;
@@ -567,61 +557,6 @@ public class QueryRunnerTestHelper
return makeQueryRunner(factory, segmentReference.orElseThrow(),
runnerName);
}
- @SuppressWarnings({"rawtypes", "unchecked"})
- public static <T> QueryRunner<T> makeFilteringQueryRunner(
- final VersionedIntervalTimeline<String, DataSegment> timeline,
- final Map<DataSegment, ReferenceCountedSegmentProvider>
referenceProviders,
- final QueryRunnerFactory<T, Query<T>> factory
- )
- {
- final QueryToolChest<T, Query<T>> toolChest = factory.getToolchest();
- return FluentQueryRunner
- .create(
- (queryPlus, responseContext) -> {
- Query<T> query = queryPlus.getQuery();
- List<TimelineObjectHolder> segments = new ArrayList<>();
- for (Interval interval : query.getIntervals()) {
- segments.addAll(timeline.lookup(interval));
- }
- List<Sequence<T>> sequences = new ArrayList<>();
- final Closer closer = Closer.create();
- try {
- for (TimelineObjectHolder<String, DataSegment> holder :
toolChest.filterSegments(
- query,
- segments
- )) {
- final SegmentDescriptor descriptor = new SegmentDescriptor(
- holder.getInterval(),
- holder.getVersion(),
- 0
- );
- final QueryPlus queryPlusRunning = queryPlus.withQuery(
- queryPlus.getQuery().withQuerySegmentSpec(new
SpecificSegmentSpec(descriptor))
- );
- final ReferenceCountedSegmentProvider referenceProvider =
referenceProviders.get(
- holder.getObject().getChunk(0).getObject()
- );
- final QueryRunner<?> runner = factory.createRunner(
-
closer.register(referenceProvider.acquireReference().orElseThrow())
- );
- sequences.add(runner.run(queryPlusRunning, responseContext));
- }
- return Sequences.withBaggage(
- new MergeSequence<>(query.getResultOrdering(),
Sequences.simple(sequences)),
- closer
- );
- }
- catch (Throwable t) {
- throw CloseableUtils.closeAndWrapInCatch(t, closer);
- }
- },
- toolChest
- )
- .applyPreMergeDecoration()
- .mergeResults(true)
- .applyPostMergeDecoration();
- }
-
public static Map<String, Object> of(Object... keyvalues)
{
ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
diff --git
a/processing/src/test/java/org/apache/druid/query/timeboundary/TimeBoundaryQueryRunnerTest.java
b/processing/src/test/java/org/apache/druid/query/timeboundary/TimeBoundaryQueryRunnerTest.java
index 755b325393e..9e9116a2cd3 100644
---
a/processing/src/test/java/org/apache/druid/query/timeboundary/TimeBoundaryQueryRunnerTest.java
+++
b/processing/src/test/java/org/apache/druid/query/timeboundary/TimeBoundaryQueryRunnerTest.java
@@ -22,43 +22,31 @@ package org.apache.druid.query.timeboundary;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
-import com.google.common.io.CharSource;
import org.apache.commons.lang3.StringUtils;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.UOE;
-import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.guava.Sequences;
import org.apache.druid.query.Druids;
import org.apache.druid.query.InlineDataSource;
+import org.apache.druid.query.QueryContexts;
import org.apache.druid.query.QueryPlus;
import org.apache.druid.query.QueryRunner;
-import org.apache.druid.query.QueryRunnerFactory;
import org.apache.druid.query.QueryRunnerTestHelper;
import org.apache.druid.query.Result;
import org.apache.druid.query.TableDataSource;
import org.apache.druid.query.TestQueryRunner;
import org.apache.druid.query.context.ConcurrentResponseContext;
import org.apache.druid.query.context.ResponseContext;
-import org.apache.druid.query.ordering.StringComparators;
+import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.filter.RangeFilter;
+import org.apache.druid.query.filter.SelectorDimFilter;
import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
-import org.apache.druid.segment.IncrementalIndexSegment;
-import org.apache.druid.segment.ReferenceCountedSegmentProvider;
+import org.apache.druid.segment.QueryableIndex;
import org.apache.druid.segment.RowBasedSegment;
-import org.apache.druid.segment.Segment;
-import org.apache.druid.segment.TestHelper;
-import org.apache.druid.segment.TestIndex;
+import org.apache.druid.segment.column.ColumnType;
import org.apache.druid.segment.column.RowSignature;
-import org.apache.druid.segment.incremental.IncrementalIndex;
-import org.apache.druid.segment.incremental.IncrementalIndexSchema;
-import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
import org.apache.druid.testing.InitializedNullHandlingTest;
-import org.apache.druid.timeline.DataSegment;
-import org.apache.druid.timeline.SegmentId;
-import org.apache.druid.timeline.VersionedIntervalTimeline;
-import org.apache.druid.timeline.partition.NoneShardSpec;
-import org.apache.druid.timeline.partition.NumberedShardSpec;
-import org.apache.druid.timeline.partition.SingleElementPartitionChunk;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.joda.time.Interval;
@@ -67,11 +55,12 @@ 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;
import java.util.List;
-import java.util.Map;
/**
*
@@ -91,11 +80,6 @@ public class TimeBoundaryQueryRunnerTest extends
InitializedNullHandlingTest
}
private final TestQueryRunner<Result<TimeBoundaryResultValue>> runner;
- private static final QueryRunnerFactory FACTORY = new
TimeBoundaryQueryRunnerFactory(
- QueryRunnerTestHelper.NOOP_QUERYWATCHER
- );
- private static Segment segment0;
- private static Segment segment1;
public TimeBoundaryQueryRunnerTest(
TestQueryRunner<Result<TimeBoundaryResultValue>> runner
@@ -104,150 +88,49 @@ public class TimeBoundaryQueryRunnerTest extends
InitializedNullHandlingTest
this.runner = runner;
}
- // Adapted from MultiSegmentSelectQueryTest, with modifications to make
filtering meaningful
- public static final String[] V_0112 = {
-
"2011-01-12T01:00:00.000Z\tspot\tbusiness\t1100\t11000.0\t110000\tpreferred\tbpreferred\t100.000000",
-
"2011-01-12T02:00:00.000Z\tspot\tentertainment\t1200\t12000.0\t120000\tpreferred\tepreferred\t100.000000",
-
"2011-01-13T00:00:00.000Z\tspot\tautomotive\t1000\t10000.0\t100000\tpreferred\tapreferred\t100.000000",
-
"2011-01-13T01:00:00.000Z\tspot\tbusiness\t1100\t11000.0\t110000\tpreferred\tbpreferred\t100.000000"
- };
- public static final String[] V_0113 = {
-
"2011-01-14T00:00:00.000Z\tspot\tautomotive\t1000\t10000.0\t100000\tpreferred\tapreferred\t94.874713",
-
"2011-01-14T02:00:00.000Z\tspot\tentertainment\t1200\t12000.0\t120000\tpreferred\tepreferred\t110.087299",
-
"2011-01-15T00:00:00.000Z\tspot\tautomotive\t1000\t10000.0\t100000\tpreferred\tapreferred\t94.874713",
-
"2011-01-15T01:00:00.000Z\tspot\tbusiness\t1100\t11000.0\t110000\tpreferred\tbpreferred\t103.629399",
-
"2011-01-16T00:00:00.000Z\tspot\tautomotive\t1000\t10000.0\t100000\tpreferred\tapreferred\t94.874713",
-
"2011-01-16T01:00:00.000Z\tspot\tbusiness\t1100\t11000.0\t110000\tpreferred\tbpreferred\t103.629399",
-
"2011-01-16T02:00:00.000Z\tspot\tentertainment\t1200\t12000.0\t120000\tpreferred\tepreferred\t110.087299",
-
"2011-01-17T01:00:00.000Z\tspot\tbusiness\t1100\t11000.0\t110000\tpreferred\tbpreferred\t103.629399",
-
"2011-01-17T02:00:00.000Z\tspot\tentertainment\t1200\t12000.0\t120000\tpreferred\tepreferred\t110.087299"
- };
-
- private static IncrementalIndex newIndex(String minTimeStamp)
- {
- return newIndex(minTimeStamp, 10000);
- }
-
- private static IncrementalIndex newIndex(String minTimeStamp, int
maxRowCount)
- {
- final IncrementalIndexSchema schema = new IncrementalIndexSchema.Builder()
- .withMinTimestamp(DateTimes.of(minTimeStamp).getMillis())
- .withQueryGranularity(Granularities.HOUR)
- .withMetrics(TestIndex.METRIC_AGGS)
- .build();
- return new OnheapIncrementalIndex.Builder()
- .setIndexSchema(schema)
- .setMaxRowCount(maxRowCount)
- .build();
- }
-
- private static SegmentId makeIdentifier(IncrementalIndex index, String
version)
- {
- return makeIdentifier(index.getInterval(), version);
- }
-
- private static SegmentId makeIdentifier(Interval interval, String version)
- {
- return SegmentId.of(QueryRunnerTestHelper.DATA_SOURCE, interval, version,
NoneShardSpec.instance());
- }
-
- private QueryRunner getCustomRunner() throws IOException
+ @Test
+ public void testFilteredTimeBoundaryQuery()
{
- CharSource v_0112 = CharSource.wrap(StringUtils.join(V_0112, "\n"));
- CharSource v_0113 = CharSource.wrap(StringUtils.join(V_0113, "\n"));
-
- IncrementalIndex index0 =
TestIndex.loadIncrementalIndexFromTsvCharSource(newIndex("2011-01-12T00:00:00.000Z"),
v_0112);
- IncrementalIndex index1 =
TestIndex.loadIncrementalIndexFromTsvCharSource(newIndex("2011-01-14T00:00:00.000Z"),
v_0113);
-
- segment0 = new IncrementalIndexSegment(index0, makeIdentifier(index0,
"v1"));
- segment1 = new IncrementalIndexSegment(index1, makeIdentifier(index1,
"v1"));
- final DataSegment dataSegment0 = TestHelper.toSimpleDataSegment(segment0,
new NumberedShardSpec(0, 1));
- final DataSegment dataSegment1 = TestHelper.toSimpleDataSegment(segment1,
new NumberedShardSpec(0, 1));
- Map<DataSegment, ReferenceCountedSegmentProvider> referenceProviders =
Map.of(
- dataSegment0, ReferenceCountedSegmentProvider.of(segment0),
- dataSegment1, ReferenceCountedSegmentProvider.of(segment1)
- );
-
- VersionedIntervalTimeline<String, DataSegment> timeline = new
VersionedIntervalTimeline<>(
- StringComparators.LEXICOGRAPHIC);
- timeline.add(
- index0.getInterval(),
- "v1",
- new SingleElementPartitionChunk<>(dataSegment0)
+ // "automotive" rows appear at both ends of the segment, so the boundary
is the boundary of the segment.
+ assertTimeBoundary(
+ new SelectorDimFilter("quality", "automotive", null),
+ null,
+ DateTimes.of("2011-01-12T00:00:00.000Z"),
+ DateTimes.of("2011-04-15T00:00:00.000Z")
);
- timeline.add(
- index1.getInterval(),
- "v1",
- new SingleElementPartitionChunk<>(dataSegment1)
- );
-
- return QueryRunnerTestHelper.makeFilteringQueryRunner(timeline,
referenceProviders, FACTORY);
}
@Test
- @SuppressWarnings("unchecked")
- public void testFilteredTimeBoundaryQuery() throws IOException
+ public void testFilteredTimeBoundaryQueryNarrowerThanSegment()
{
- QueryRunner customRunner = getCustomRunner();
- TimeBoundaryQuery timeBoundaryQuery = Druids.newTimeBoundaryQueryBuilder()
- .dataSource("testing")
- .filters("quality",
"automotive")
- .build();
- Assertions.assertTrue(timeBoundaryQuery.hasFilters());
- List<Result<TimeBoundaryResultValue>> results =
- customRunner.run(QueryPlus.wrap(timeBoundaryQuery)).toList();
-
- Assertions.assertTrue(Iterables.size(results) > 0);
-
- TimeBoundaryResultValue val = results.iterator().next().getValue();
- DateTime minTime = val.getMinTime();
- DateTime maxTime = val.getMaxTime();
-
- Assertions.assertEquals(DateTimes.of("2011-01-13T00:00:00.000Z"), minTime);
- Assertions.assertEquals(DateTimes.of("2011-01-16T00:00:00.000Z"), maxTime);
+ // Only four rows have "index" >= 1700, and they all lie strictly inside
the segment: the earliest is on
+ // 2011-01-30 and the latest is on 2011-03-31.
+ assertTimeBoundary(
+ new RangeFilter("index", ColumnType.DOUBLE, 1700.0, null, false, null,
null),
+ null,
+ DateTimes.of("2011-01-30T00:00:00.000Z"),
+ DateTimes.of("2011-03-31T00:00:00.000Z")
+ );
}
@Test
- @SuppressWarnings("unchecked")
- public void testTimeFilteredTimeBoundaryQuery() throws IOException
+ public void testTimeFilteredTimeBoundaryQuery()
{
- QueryRunner customRunner = getCustomRunner();
- TimeBoundaryQuery timeBoundaryQuery = Druids.newTimeBoundaryQueryBuilder()
- .dataSource("testing")
- .intervals(
- new
MultipleIntervalSegmentSpec(
-
ImmutableList.of(Intervals.of(
-
"2011-01-15T00:00:00.000Z/2011-01-16T00:00:00.000Z"))
- )
- )
- .build();
- List<Result<TimeBoundaryResultValue>> results =
- customRunner.run(QueryPlus.wrap(timeBoundaryQuery)).toList();
-
- Assertions.assertTrue(Iterables.size(results) > 0);
-
- TimeBoundaryResultValue val = results.iterator().next().getValue();
- DateTime minTime = val.getMinTime();
- DateTime maxTime = val.getMaxTime();
-
- Assertions.assertEquals(DateTimes.of("2011-01-15T00:00:00.000Z"), minTime);
- Assertions.assertEquals(DateTimes.of("2011-01-15T01:00:00.000Z"), maxTime);
+ // There are no rows on the edges of the query interval: the earliest row
inside it is on 2011-01-20T01, and the
+ // latest is on 2011-01-22.
+ assertTimeBoundary(
+ null,
+ Intervals.of("2011-01-20T00:00:00.000Z/2011-01-23T00:00:00.000Z"),
+ DateTimes.of("2011-01-20T01:00:00.000Z"),
+ DateTimes.of("2011-01-22T00:00:00.000Z")
+ );
}
@Test
- @SuppressWarnings("unchecked")
- public void testFilteredTimeBoundaryQueryNoMatches() throws IOException
+ public void testFilteredTimeBoundaryQueryNoMatches()
{
- QueryRunner customRunner = getCustomRunner();
- TimeBoundaryQuery timeBoundaryQuery = Druids.newTimeBoundaryQueryBuilder()
- .dataSource("testing")
- .filters("quality", "foobar")
// foobar dimension does not exist
- .build();
- Assertions.assertTrue(timeBoundaryQuery.hasFilters());
- List<Result<TimeBoundaryResultValue>> results =
- customRunner.run(QueryPlus.wrap(timeBoundaryQuery)).toList();
-
- Assertions.assertTrue(Iterables.size(results) == 0);
+ // "foobar" quality does not exist.
+ assertTimeBoundary(new SelectorDimFilter("quality", "foobar", null), null,
null, null);
}
@Test
@@ -435,4 +318,66 @@ public class TimeBoundaryQueryRunnerTest extends
InitializedNullHandlingTest
Assertions.assertFalse(actual.iterator().hasNext());
}
+
+ /**
+ * Run a time boundary query against {@link #runner} for every "bound" and
every vectorization mode that the
+ * runner's segment supports, and verify the min and max time. Null
expectations mean that the query is expected to
+ * return no results at all.
+ */
+ private void assertTimeBoundary(
+ @Nullable final DimFilter filter,
+ @Nullable final Interval interval,
+ @Nullable final DateTime expectedMinTime,
+ @Nullable final DateTime expectedMaxTime
+ )
+ {
+ final List<String> vectorizeValues = new
ArrayList<>(Arrays.asList("false", "true"));
+
+ if (runner.getSegment().as(QueryableIndex.class) != null) {
+ vectorizeValues.add("force");
+ }
+
+ for (final String bound : Arrays.asList(TimeBoundaryQuery.MIN_TIME,
TimeBoundaryQuery.MAX_TIME, null)) {
+ for (final String vectorize : vectorizeValues) {
+ final String message = StringUtils.join(new Object[]{runner.getName(),
bound, vectorize}, ' ');
+ final TimeBoundaryQuery query =
+ Druids.newTimeBoundaryQueryBuilder()
+ .dataSource(QueryRunnerTestHelper.DATA_SOURCE)
+ .filters(filter)
+ .intervals(
+ interval == null
+ ? null
+ : new
MultipleIntervalSegmentSpec(ImmutableList.of(interval))
+ )
+ .bound(bound)
+ .context(
+ ImmutableMap.of(
+ QueryContexts.VECTORIZE_KEY, vectorize,
+ QueryContexts.VECTOR_SIZE_KEY, 7
+ )
+ )
+ .build();
+
+ Assertions.assertEquals(filter != null, query.hasFilters(), message);
+
+ final ResponseContext context =
ConcurrentResponseContext.createEmpty();
+ context.initializeMissingSegments();
+ final List<Result<TimeBoundaryResultValue>> results =
+ runner.run(QueryPlus.wrap(query), context).toList();
+
+ final DateTime expectedMinTimeForBound =
+ TimeBoundaryQuery.MAX_TIME.equals(bound) ? null : expectedMinTime;
+ final DateTime expectedMaxTimeForBound =
+ TimeBoundaryQuery.MIN_TIME.equals(bound) ? null : expectedMaxTime;
+
+ if (expectedMinTimeForBound == null && expectedMaxTimeForBound ==
null) {
+ Assertions.assertEquals(Collections.emptyList(), results, message);
+ } else {
+ final TimeBoundaryResultValue val =
Iterables.getOnlyElement(results).getValue();
+ Assertions.assertEquals(expectedMinTimeForBound, val.getMinTime(),
message);
+ Assertions.assertEquals(expectedMaxTimeForBound, val.getMaxTime(),
message);
+ }
+ }
+ }
+ }
}
diff --git
a/processing/src/test/java/org/apache/druid/query/timeseries/TimeseriesQueryRunnerTest.java
b/processing/src/test/java/org/apache/druid/query/timeseries/TimeseriesQueryRunnerTest.java
index da5d70f9079..f5f6615e53b 100644
---
a/processing/src/test/java/org/apache/druid/query/timeseries/TimeseriesQueryRunnerTest.java
+++
b/processing/src/test/java/org/apache/druid/query/timeseries/TimeseriesQueryRunnerTest.java
@@ -133,9 +133,7 @@ public class TimeseriesQueryRunnerTest extends
InitializedNullHandlingTest
.stream(baseConstructors.spliterator(), false)
.filter(
constructor -> {
- boolean canVectorize =
- QueryRunnerTestHelper.isTestRunnerVectorizable((QueryRunner)
constructor[0])
- && !(boolean) constructor[1] /* descending */;
+ boolean canVectorize =
QueryRunnerTestHelper.isTestRunnerVectorizable((QueryRunner) constructor[0]);
final boolean vectorize = (boolean) constructor[2]; /* vectorize
*/
final boolean useVectorApi = (boolean) constructor[4]; /*
useVectorApi */
if (!vectorize && useVectorApi) {
diff --git
a/processing/src/test/java/org/apache/druid/segment/DescendingVectorCursorDiffTest.java
b/processing/src/test/java/org/apache/druid/segment/DescendingVectorCursorDiffTest.java
new file mode 100644
index 00000000000..fed6d7a0ef1
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/segment/DescendingVectorCursorDiffTest.java
@@ -0,0 +1,306 @@
+/*
+ * 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.segment;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.math.expr.ExprMacroTable;
+import org.apache.druid.query.QueryContext;
+import org.apache.druid.query.QueryContexts;
+import org.apache.druid.query.filter.Filter;
+import org.apache.druid.query.filter.InDimFilter;
+import org.apache.druid.query.filter.NotDimFilter;
+import org.apache.druid.query.filter.NullFilter;
+import org.apache.druid.query.filter.RangeFilter;
+import org.apache.druid.query.filter.SelectorDimFilter;
+import org.apache.druid.segment.column.ColumnHolder;
+import org.apache.druid.segment.column.ColumnType;
+import org.apache.druid.segment.filter.AndFilter;
+import org.apache.druid.segment.incremental.IncrementalIndexSchema;
+import org.apache.druid.segment.shim.ShimCursor;
+import org.apache.druid.segment.virtual.ExpressionVirtualColumn;
+import org.apache.druid.segment.virtual.NestedFieldVirtualColumn;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.joda.time.Interval;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.annotation.Nullable;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Verifies that a time-descending vectorized cursor over a {@link
QueryableIndex} produces exactly the same rows, in
+ * the same order, as the non-vectorized descending cursor.
+ */
+public class DescendingVectorCursorDiffTest extends InitializedNullHandlingTest
+{
+ private static final int[] VECTOR_SIZES = new int[]{1, 2, 3, 7, 512};
+
+ @TempDir
+ public File temporaryFolder;
+
+ @Test
+ public void testStandardColumns()
+ {
+ final VirtualColumns virtualColumns = VirtualColumns.create(
+ new ExpressionVirtualColumn("vc", "index * 2 + 1", ColumnType.DOUBLE,
ExprMacroTable.nil())
+ );
+
+ assertVectorizedMatchesNonVectorized(
+ new QueryableIndexCursorFactory(TestIndex.getMMappedTestIndex()),
+ virtualColumns,
+ ImmutableList.of(
+ ColumnHolder.TIME_COLUMN_NAME,
+ "market", // single-value string
+ "placementish", // multi-value string
+ "partial_null_column", // partially-null string
+ "qualityLong",
+ "longNumericNull", // null-bearing long
+ "floatNumericNull", // null-bearing float
+ "doubleNumericNull", // null-bearing double
+ "index", // double metric
+ "quality_uniques", // complex column
+ "vc" // expression virtual column
+ ),
+ ImmutableMap.of(
+ // Bitmap-indexed filters use DescendingBitmapVectorOffset.
+ "bitmapIndex", new SelectorDimFilter("market", "spot",
null).toFilter(),
+ "bitmapIndexSelective", new InDimFilter("quality",
ImmutableSet.of("automotive")).toFilter(),
+ "nullBitmapIndex", new NotDimFilter(new
NullFilter("longNumericNull", null)).toFilter(),
+
+ // Filters without an index use FilteredVectorOffset on top of a
descending offset.
+ "valueMatcher", new RangeFilter("index", ColumnType.DOUBLE, 100.0,
1000.0, true, true, null),
+ "virtualColumnMatcher", new RangeFilter("vc", ColumnType.DOUBLE,
500.0, null, true, false, null),
+ "bitmapIndexAndValueMatcher", new AndFilter(
+ ImmutableList.of(
+ new SelectorDimFilter("market", "spot", null).toFilter(),
+ new RangeFilter("index", ColumnType.DOUBLE, 100.0, null,
true, false, null)
+ )
+ )
+ ),
+ // Strictly inside the segment interval, which runs from 2011-01-12 to
2011-04-15.
+ Intervals.of("2011-01-13T00:00:00.000Z/2011-03-01T00:00:00.000Z")
+ );
+ }
+
+ @Test
+ public void testNestedColumns()
+ {
+ final VirtualColumns virtualColumns = VirtualColumns.create(
+ new NestedFieldVirtualColumn("nested", "$.x", "nestedX",
ColumnType.LONG),
+ new NestedFieldVirtualColumn("nested", "$.y", "nestedY",
ColumnType.STRING)
+ );
+
+ assertVectorizedMatchesNonVectorized(
+ new QueryableIndexCursorFactory(makeNestedIndex()),
+ virtualColumns,
+ ImmutableList.of(
+ ColumnHolder.TIME_COLUMN_NAME,
+ "str",
+ "lng",
+ "dbl",
+ "arr",
+ "nested",
+ "nestedX",
+ "nestedY"
+ ),
+ ImmutableMap.of(
+ "bitmapIndex", new SelectorDimFilter("str", "s3", null).toFilter(),
+ "valueMatcher", new RangeFilter("lng", ColumnType.LONG, 50L, null,
true, false, null),
+ "nestedField", new SelectorDimFilter("nestedY", "a",
null).toFilter()
+ ),
+ // Rows are one minute apart starting at 2000-01-01, so this clips to
rows [37, 191).
+ Intervals.of("2000-01-01T00:37:00.000Z/2000-01-01T03:11:00.000Z")
+ );
+ }
+
+ /**
+ * Runs the unfiltered case, plus each of the named {@code filters}, at each
of the {@link #VECTOR_SIZES}. Each
+ * combination runs twice: once over the entire segment, and once over
{@code narrowInterval}, which must lie
+ * strictly inside the segment so that the descending offset starts partway
through the segment (nonzero start
+ * offset) and stops before its final row (clipped end offset).
+ */
+ private void assertVectorizedMatchesNonVectorized(
+ final CursorFactory cursorFactory,
+ final VirtualColumns virtualColumns,
+ final List<String> columns,
+ final Map<String, Filter> filters,
+ final Interval narrowInterval
+ )
+ {
+ final Map<String, Filter> allCases = new LinkedHashMap<>();
+ allCases.put("noFilter", null);
+ allCases.putAll(filters);
+
+ for (final Map.Entry<String, Filter> filterEntry : allCases.entrySet()) {
+ for (final int vectorSize : VECTOR_SIZES) {
+ // Row count for Intervals.ETERNITY, used to verify that
"narrowInterval" actually narrows.
+ int fullSegmentRowCount = -1;
+
+ for (final Interval interval : ImmutableList.of(Intervals.ETERNITY,
narrowInterval)) {
+ final String message = StringUtils.format(
+ "interval[%s] filter[%s] vectorSize[%d]",
+ interval,
+ filterEntry.getKey(),
+ vectorSize
+ );
+ final CursorBuildSpec buildSpec =
+ CursorBuildSpec.builder()
+ .setInterval(interval)
+
.setPreferredOrdering(Cursors.descendingTimeOrder())
+ .setVirtualColumns(virtualColumns)
+ .setFilter(filterEntry.getValue())
+ .setQueryContext(
+
QueryContext.of(ImmutableMap.of(QueryContexts.VECTOR_SIZE_KEY, vectorSize))
+ )
+ .build();
+
+ final List<List<Object>> expected;
+ final List<List<Object>> actual;
+
+ try (final CursorHolder holder =
cursorFactory.makeCursorHolder(buildSpec)) {
+ Assertions.assertEquals(Cursors.descendingTimeOrder(),
holder.getOrdering(), message);
+ expected = readRows(holder.asCursor(), columns);
+ }
+
+ try (final CursorHolder holder =
cursorFactory.makeCursorHolder(buildSpec)) {
+ Assertions.assertTrue(holder.canVectorize(), message);
+ actual = readRows(new ShimCursor(holder.asVectorCursor()),
columns);
+ }
+
+ Assertions.assertFalse(expected.isEmpty(), message + ": expected
some rows");
+ Assertions.assertEquals(expected.size(), actual.size(), message + ":
row count");
+ for (int i = 0; i < expected.size(); i++) {
+ Assertions.assertEquals(expected.get(i), actual.get(i), message +
": row " + i);
+ }
+
+ // Sanity check: __time is non-increasing.
+ long previousTime = Long.MAX_VALUE;
+ for (final List<Object> row : actual) {
+ final long time = ((Number) row.get(0)).longValue();
+ Assertions.assertTrue(time <= previousTime, message + ":
descending __time");
+ previousTime = time;
+ }
+
+ if (Intervals.ETERNITY.equals(interval)) {
+ fullSegmentRowCount = actual.size();
+ } else {
+ // Guards against the narrow-interval case silently degrading into
a second full-segment scan.
+ Assertions.assertTrue(
+ actual.size() < fullSegmentRowCount,
+ message + ": narrow interval must return fewer rows than the
full segment"
+ );
+ }
+ }
+ }
+ }
+ }
+
+ private QueryableIndex makeNestedIndex()
+ {
+ final List<String> dimensions = ImmutableList.of("str", "lng", "dbl",
"arr", "nested");
+ final List<InputRow> rows = new ArrayList<>();
+
+ for (int i = 0; i < 300; i++) {
+ final Map<String, Object> event = new HashMap<>();
+ event.put("str", i % 7 == 0 ? null : "s" + (i % 13));
+ event.put("lng", i % 5 == 0 ? null : (long) i);
+ event.put("dbl", i % 3 == 0 ? null : i * 1.5);
+ event.put("arr", i % 11 == 0 ? null : ImmutableList.of((long) i, (long)
(i + 1)));
+ event.put("nested", i % 9 == 0 ? null : ImmutableMap.of("x", i, "y", i %
4 == 0 ? "a" : "b"));
+ rows.add(new MapBasedInputRow(DateTimes.of("2000-01-01").plusMinutes(i),
dimensions, event));
+ }
+
+ return IndexBuilder.create()
+ .tmpDir(temporaryFolder)
+ .schema(
+ IncrementalIndexSchema
+ .builder()
+ .withDimensionsSpec(
+ DimensionsSpec.builder()
+ .setDimensions(
+ dimensions.stream()
+ .map(d -> new
AutoTypeColumnSchema(d, null, null))
+
.collect(Collectors.toList())
+ )
+ .build()
+ )
+ .withRollup(false)
+ .build()
+ )
+ .rows(rows)
+ .buildMMappedIndex();
+ }
+
+ private static List<List<Object>> readRows(final Cursor cursor, final
List<String> columns)
+ {
+ final ColumnSelectorFactory columnSelectorFactory =
cursor.getColumnSelectorFactory();
+ final List<ColumnValueSelector<?>> selectors = new
ArrayList<>(columns.size());
+ for (final String column : columns) {
+ selectors.add(columnSelectorFactory.makeColumnValueSelector(column));
+ }
+
+ final List<List<Object>> rows = new ArrayList<>();
+ while (!cursor.isDone()) {
+ final List<Object> row = new ArrayList<>(columns.size());
+ for (final ColumnValueSelector<?> selector : selectors) {
+ row.add(comparableValue(selector.getObject()));
+ }
+ rows.add(row);
+ cursor.advance();
+ }
+ return rows;
+ }
+
+ /**
+ * Converts a selector value into something with meaningful {@link
Object#equals}.
+ */
+ @Nullable
+ private static Object comparableValue(@Nullable final Object o)
+ {
+ if (o instanceof Object[]) {
+ return comparableList(Arrays.asList((Object[]) o));
+ } else if (o instanceof List) {
+ return comparableList((List<?>) o);
+ } else {
+ return o;
+ }
+ }
+
+ private static List<Object> comparableList(final List<?> list)
+ {
+ return
list.stream().map(DescendingVectorCursorDiffTest::comparableValue).collect(Collectors.toList());
+ }
+}
diff --git a/processing/src/test/java/org/apache/druid/segment/TestHelper.java
b/processing/src/test/java/org/apache/druid/segment/TestHelper.java
index 6b3ef27f854..5cb2883a125 100644
--- a/processing/src/test/java/org/apache/druid/segment/TestHelper.java
+++ b/processing/src/test/java/org/apache/druid/segment/TestHelper.java
@@ -43,9 +43,7 @@ import org.apache.druid.query.topn.TopNResultValue;
import org.apache.druid.segment.column.ColumnConfig;
import org.apache.druid.segment.join.JoinableFactoryWrapper;
import org.apache.druid.segment.writeout.SegmentWriteOutMediumFactory;
-import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.DataSegment.PruneSpecsHolder;
-import org.apache.druid.timeline.partition.ShardSpec;
import org.junit.jupiter.api.Assertions;
import java.io.IOException;
@@ -519,12 +517,4 @@ public class TestHelper
throw new UncheckedIOException(e);
}
}
-
- public static DataSegment toSimpleDataSegment(Segment segment, ShardSpec
shardSpec)
- {
- return DataSegment.builder(segment.getId())
- .shardSpec(shardSpec)
- .size(0)
- .build();
- }
}
diff --git
a/processing/src/test/java/org/apache/druid/segment/vector/DescendingVectorOffsetTest.java
b/processing/src/test/java/org/apache/druid/segment/vector/DescendingVectorOffsetTest.java
new file mode 100644
index 00000000000..c3110e395f6
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/segment/vector/DescendingVectorOffsetTest.java
@@ -0,0 +1,253 @@
+/*
+ * 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.segment.vector;
+
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+import it.unimi.dsi.fastutil.ints.IntList;
+import org.apache.druid.collections.bitmap.ImmutableBitmap;
+import org.apache.druid.collections.bitmap.WrappedImmutableRoaringBitmap;
+import org.apache.druid.error.DruidException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
+
+import java.util.List;
+import java.util.stream.Stream;
+
+/**
+ * Verifies the tail-first, ascending-within-batch behavior of {@link
DescendingNoFilterVectorOffset} and
+ * {@link DescendingBitmapVectorOffset}. Combined with the value reversal done
by
+ * {@link ReverseVectorColumnSelectorFactory}, iterating these offsets yields
rows in fully descending order.
+ */
+public class DescendingVectorOffsetTest
+{
+ private static final int BITMAP_ROWS = 1000;
+
+ private static final List<Integer> NO_FILTER_VECTOR_SIZES = List.of(1, 2, 3,
7, 16);
+ private static final List<Integer> NO_FILTER_ENDS = List.of(0, 1, 5, 16, 17,
100);
+ private static final List<Integer> NO_FILTER_STARTS = List.of(0, 1, 3);
+ private static final List<Integer> BITMAP_VECTOR_SIZES = List.of(1, 2, 7,
16);
+
+ /**
+ * Every bit set, and every third one.
+ */
+ private static final List<Integer> BITMAP_STEPS = List.of(1, 3);
+
+ /**
+ * The whole bitmap, an interior range, a single-row range, the last row,
and two empty ranges.
+ */
+ private static final List<int[]> BITMAP_RANGES = List.of(
+ new int[]{0, BITMAP_ROWS},
+ new int[]{10, 500},
+ new int[]{7, 8},
+ new int[]{999, 1000},
+ new int[]{500, 500},
+ new int[]{0, 0}
+ );
+
+ public static Stream<Arguments> noFilterParameters()
+ {
+ return NO_FILTER_VECTOR_SIZES.stream().flatMap(
+ vectorSize -> NO_FILTER_ENDS.stream().flatMap(
+ end -> NO_FILTER_STARTS.stream()
+ .filter(start -> start <= end)
+ .map(start -> Arguments.of(vectorSize,
start, end))
+ )
+ );
+ }
+
+ public static Stream<Arguments> bitmapParameters()
+ {
+ return BITMAP_STEPS.stream().flatMap(
+ step -> BITMAP_VECTOR_SIZES.stream().flatMap(
+ vectorSize -> BITMAP_RANGES.stream().map(range ->
Arguments.of(step, vectorSize, range[0], range[1]))
+ )
+ );
+ }
+
+ @ParameterizedTest(name = "vectorSize[{0}] range[{1},{2})")
+ @MethodSource("noFilterParameters")
+ public void testNoFilterProducesDescendingOrder(final int vectorSize, final
int start, final int end)
+ {
+ final DescendingNoFilterVectorOffset offset = new
DescendingNoFilterVectorOffset(vectorSize, start, end);
+
+ // Reversing each batch's ascending range and concatenating across batches
must yield [end-1 .. start].
+ final IntList logical = new IntArrayList();
+ while (!offset.isDone()) {
+ Assertions.assertTrue(offset.isContiguous(), "isContiguous");
+ Assertions.assertThrows(UnsupportedOperationException.class,
offset::getOffsets);
+
+ final int startOffset = offset.getStartOffset();
+ final int size = offset.getCurrentVectorSize();
+ Assertions.assertTrue(size > 0 && size <= vectorSize, "0 < size[" + size
+ "] <= vectorSize");
+
+ // Batches are aligned to multiples of vectorSize, so that they do not
straddle compressed block boundaries.
+ // Only the batch that runs into "start" may be unaligned.
+ Assertions.assertTrue(
+ startOffset == start || startOffset % vectorSize == 0,
+ "aligned startOffset[" + startOffset + "]"
+ );
+
+ for (int i = size - 1; i >= 0; i--) {
+ logical.add(startOffset + i);
+ }
+ offset.advance();
+ }
+
+ final IntList expected = new IntArrayList();
+ for (int i = end - 1; i >= start; i--) {
+ expected.add(i);
+ }
+ Assertions.assertEquals(expected, logical);
+ }
+
+ @ParameterizedTest(name = "step[{0}] vectorSize[{1}] range[{2},{3})")
+ @MethodSource("bitmapParameters")
+ public void testBitmapProducesDescendingOrder(
+ final int step,
+ final int vectorSize,
+ final int start,
+ final int end
+ )
+ {
+ final ImmutableBitmap bitmap = makeBitmap(BITMAP_ROWS, step);
+ final DescendingBitmapVectorOffset offset = new
DescendingBitmapVectorOffset(vectorSize, bitmap, start, end);
+
+ final IntList logical = new IntArrayList();
+ while (!offset.isDone()) {
+ final int size = offset.getCurrentVectorSize();
+ Assertions.assertTrue(size > 0 && size <= vectorSize, "0 < size[" + size
+ "] <= vectorSize");
+
+ final int[] batch = currentBatch(offset);
+
+ // Within a batch, offsets are ascending (required by column readers).
+ for (int i = 1; i < size; i++) {
+ Assertions.assertTrue(batch[i] > batch[i - 1], "ascending within
batch");
+ }
+
+ // getId is the smallest offset of the batch.
+ Assertions.assertEquals(batch[0], offset.getId(), "getId");
+
+ for (int i = size - 1; i >= 0; i--) {
+ logical.add(batch[i]);
+ }
+ offset.advance();
+ }
+
+ // Expected: matching set bits in [start, end) in descending order.
+ final IntList expected = new IntArrayList();
+ for (int i = end - 1; i >= start; i--) {
+ if (bitmap.get(i)) {
+ expected.add(i);
+ }
+ }
+ Assertions.assertEquals(expected, logical);
+ }
+
+ @Test
+ public void testNoFilterReset()
+ {
+ final DescendingNoFilterVectorOffset offset = new
DescendingNoFilterVectorOffset(4, 0, 10);
+ final IntList firstPass = drainNoFilter(offset);
+ offset.reset();
+ final IntList secondPass = drainNoFilter(offset);
+ Assertions.assertEquals(firstPass, secondPass);
+ }
+
+ @Test
+ public void testBitmapReset()
+ {
+ final ImmutableBitmap bitmap = makeBitmap(100, 3);
+ final DescendingBitmapVectorOffset offset = new
DescendingBitmapVectorOffset(7, bitmap, 5, 90);
+ final IntList firstPass = drainBitmap(offset);
+ offset.reset();
+ final IntList secondPass = drainBitmap(offset);
+ Assertions.assertEquals(firstPass, secondPass);
+ Assertions.assertFalse(firstPass.isEmpty());
+ }
+
+ @Test
+ public void testBitmapDetectsContiguousBatches()
+ {
+ final DescendingBitmapVectorOffset offset = new
DescendingBitmapVectorOffset(4, makeBitmap(100, 1), 0, 100);
+ Assertions.assertTrue(offset.isContiguous());
+ Assertions.assertEquals(96, offset.getStartOffset());
+ Assertions.assertEquals(4, offset.getCurrentVectorSize());
+ Assertions.assertThrows(DruidException.class, offset::getOffsets);
+ }
+
+ private static ImmutableBitmap makeBitmap(final int rows, final int step)
+ {
+ final MutableRoaringBitmap wrapped = new MutableRoaringBitmap();
+ for (int i = 0; i < rows; i++) {
+ if (i % step == 0) {
+ wrapped.add(i);
+ }
+ }
+ return new
WrappedImmutableRoaringBitmap(wrapped.toImmutableRoaringBitmap());
+ }
+
+ private static int[] currentBatch(final VectorOffset offset)
+ {
+ final int size = offset.getCurrentVectorSize();
+ final int[] batch = new int[size];
+ if (offset.isContiguous()) {
+ Assertions.assertThrows(DruidException.class, offset::getOffsets);
+ final int startOffset = offset.getStartOffset();
+ for (int i = 0; i < size; i++) {
+ batch[i] = startOffset + i;
+ }
+ } else {
+ Assertions.assertThrows(DruidException.class, offset::getStartOffset);
+ System.arraycopy(offset.getOffsets(), 0, batch, 0, size);
+ }
+ return batch;
+ }
+
+ private static IntList drainNoFilter(final DescendingNoFilterVectorOffset
offset)
+ {
+ final IntList logical = new IntArrayList();
+ while (!offset.isDone()) {
+ final int startOffset = offset.getStartOffset();
+ final int size = offset.getCurrentVectorSize();
+ for (int i = size - 1; i >= 0; i--) {
+ logical.add(startOffset + i);
+ }
+ offset.advance();
+ }
+ return logical;
+ }
+
+ private static IntList drainBitmap(final DescendingBitmapVectorOffset offset)
+ {
+ final IntList logical = new IntArrayList();
+ while (!offset.isDone()) {
+ final int[] batch = currentBatch(offset);
+ for (int i = batch.length - 1; i >= 0; i--) {
+ logical.add(batch[i]);
+ }
+ offset.advance();
+ }
+ return logical;
+ }
+}
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java
index 0c21e6a90a6..b1980846207 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteJoinQueryTest.java
@@ -3966,8 +3966,10 @@ public class CalciteJoinQueryTest extends
BaseCalciteQueryTest
@ParameterizedTest(name = "{0}")
public void testTwoSemiJoinsSimultaneously(Map<String, Object> queryContext)
{
- // Cannot vectorize timeBoundary with maxTime (the engine will request
descending order, which cannot vectorize).
- cannotVectorize();
+ if (!isRewriteJoinToFilter(queryContext)) {
+ // Rewriting the joins to filters allows all queries to vectorize.
+ cannotVectorize();
+ }
Map<String, Object> updatedQueryContext = new HashMap<>(queryContext);
updatedQueryContext.put(QueryContexts.TIME_BOUNDARY_PLANNING_KEY, true);
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
index 3c2cff27729..b033aa4ea9a 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteQueryTest.java
@@ -10034,9 +10034,6 @@ public class CalciteQueryTest extends
BaseCalciteQueryTest
@Test
public void testTimeseriesDescending()
{
- // Cannot vectorize due to descending order.
- cannotVectorize();
-
testQuery(
"SELECT gran, SUM(cnt) FROM (\n"
+ " SELECT floor(__time TO month) AS gran,\n"
@@ -12391,9 +12388,6 @@ public class CalciteQueryTest extends
BaseCalciteQueryTest
@Test
public void testPostAggWithTimeseries()
{
- // Cannot vectorize due to descending order.
- cannotVectorize();
-
testQuery(
"SELECT "
+ " FLOOR(__time TO YEAR), "
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeBoundaryQueryTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeBoundaryQueryTest.java
index e555ab65a96..c3c4ba6a575 100644
---
a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeBoundaryQueryTest.java
+++
b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteTimeBoundaryQueryTest.java
@@ -198,7 +198,7 @@ public class CalciteTimeBoundaryQueryTest extends
BaseCalciteQueryTest
@Test
public void testMaxTimeQueryWithJoin()
{
- // Cannot vectorize timeBoundary with maxTime (the engine will request
descending order, which cannot vectorize).
+ // Cannot vectorize join.
cannotVectorize();
HashMap<String, Object> context = new HashMap<>(QUERY_CONTEXT_DEFAULT);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
