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 9318bbee1a2 perf: Faster advanceIfNeeded for Roaring iterators.
(#20129)
9318bbee1a2 is described below
commit 9318bbee1a26e959b2526ac7cf7045dc906147f2
Author: Gian Merlino <[email protected]>
AuthorDate: Fri Aug 28 20:29:10 2026 -0700
perf: Faster advanceIfNeeded for Roaring iterators. (#20129)
This patch adds SeekableRoaringIntIterator and uses it instead of
ImmutableRoaringBitmap#getIntIterator in WrappedImmutableRoaringBitmap.
The seekable iterator performs better than the builtin iterator,
because it uses PointableRoaringArray#advanceUntil to advance containers,
which is capable of binary search.
This patch also updates various code paths that reset bitmap iterators
in vectorized code paths to get them all using common logic: they should
reset the iterator if the new start is prior to the old end, rather than
comparing start-to-start or end-to-end.
---
.../druid/benchmark/query/TimeseriesBenchmark.java | 145 ++---
.../bitmap/SeekableRoaringIntIterator.java | 179 ++++++
.../bitmap/WrappedImmutableRoaringBitmap.java | 2 +-
.../apache/druid/segment/data/ColumnarDoubles.java | 8 +-
.../apache/druid/segment/data/ColumnarFloats.java | 8 +-
.../apache/druid/segment/data/ColumnarLongs.java | 8 +-
.../org/apache/druid/segment/filter/OrFilter.java | 24 +-
.../nested/NestedFieldDictionaryEncodedColumn.java | 43 +-
.../druid/segment/nested/ScalarDoubleColumn.java | 11 +-
.../druid/segment/nested/ScalarLongColumn.java | 9 +-
.../apache/druid/segment/nested/VariantColumn.java | 15 +-
.../buffer/DruidRoaringBufferAccess.java | 38 ++
.../bitmap/SeekableRoaringIntIteratorTest.java | 597 +++++++++++++++++++++
.../segment/filter/OrFilterVectorMatcherTest.java | 133 +++++
14 files changed, 1089 insertions(+), 131 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 54bfca56f86..3a8c1b8c78e 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
@@ -19,12 +19,9 @@
package org.apache.druid.benchmark.query;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.druid.jackson.DefaultObjectMapper;
-import org.apache.druid.java.util.common.FileUtils;
import org.apache.druid.java.util.common.Intervals;
import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.granularity.Granularity;
import org.apache.druid.java.util.common.guava.Sequence;
import org.apache.druid.java.util.common.logger.Logger;
@@ -48,6 +45,7 @@ import
org.apache.druid.query.aggregation.hyperloglog.HyperUniquesSerde;
import org.apache.druid.query.context.ResponseContext;
import org.apache.druid.query.filter.BoundDimFilter;
import org.apache.druid.query.filter.DimFilter;
+import org.apache.druid.query.filter.OrDimFilter;
import org.apache.druid.query.filter.SelectorDimFilter;
import org.apache.druid.query.ordering.StringComparators;
import org.apache.druid.query.spec.MultipleIntervalSegmentSpec;
@@ -58,23 +56,17 @@ import
org.apache.druid.query.timeseries.TimeseriesQueryQueryToolChest;
import org.apache.druid.query.timeseries.TimeseriesQueryRunnerFactory;
import org.apache.druid.query.timeseries.TimeseriesResultValue;
import org.apache.druid.segment.IncrementalIndexSegment;
-import org.apache.druid.segment.IndexIO;
-import org.apache.druid.segment.IndexMergerV9;
-import org.apache.druid.segment.IndexSpec;
import org.apache.druid.segment.QueryableIndex;
import org.apache.druid.segment.QueryableIndexSegment;
-import org.apache.druid.segment.column.ColumnConfig;
import org.apache.druid.segment.column.ColumnHolder;
-import org.apache.druid.segment.generator.DataGenerator;
import org.apache.druid.segment.generator.GeneratorBasicSchemas;
import org.apache.druid.segment.generator.GeneratorSchemaInfo;
-import org.apache.druid.segment.incremental.AppendableIndexSpec;
+import org.apache.druid.segment.generator.SegmentGenerator;
import org.apache.druid.segment.incremental.IncrementalIndex;
-import org.apache.druid.segment.incremental.IncrementalIndexCreator;
-import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
import org.apache.druid.segment.serde.ComplexMetrics;
-import
org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
+import org.apache.druid.timeline.DataSegment;
import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.partition.LinearShardSpec;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
@@ -89,7 +81,6 @@ import org.openjdk.jmh.annotations.TearDown;
import org.openjdk.jmh.annotations.Warmup;
import org.openjdk.jmh.infra.Blackhole;
-import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
@@ -118,7 +109,13 @@ public class TimeseriesBenchmark
@Param({"750000"})
private int rowsPerSegment;
- @Param({"basic.A", "basic.timeFilterNumeric",
"basic.timeFilterAlphanumeric", "basic.timeFilterByInterval"})
+ @Param({
+ "basic.A",
+ "basic.timeFilterNumeric",
+ "basic.timeFilterAlphanumeric",
+ "basic.timeFilterByInterval",
+ "basic.orFilterPartialIndex"
+ })
private String schemaAndQuery;
@Param({"true", "false"})
@@ -131,28 +128,11 @@ public class TimeseriesBenchmark
private String queryGranularity;
private static final Logger log = new Logger(TimeseriesBenchmark.class);
- private static final int RNG_SEED = 9999;
- private static final IndexMergerV9 INDEX_MERGER_V9;
- private static final IndexIO INDEX_IO;
- public static final ObjectMapper JSON_MAPPER;
- private AppendableIndexSpec appendableIndexSpec;
- private DataGenerator generator;
private QueryRunnerFactory factory;
private GeneratorSchemaInfo schemaInfo;
private TimeseriesQuery query;
- static {
- JSON_MAPPER = new DefaultObjectMapper();
- INDEX_IO = new IndexIO(
- JSON_MAPPER,
- new ColumnConfig()
- {
- }
- );
- INDEX_MERGER_V9 = new IndexMergerV9(JSON_MAPPER, INDEX_IO,
OffHeapMemorySegmentWriteOutMediumFactory.instance());
- }
-
private static final Map<String, Map<String, TimeseriesQuery>>
SCHEMA_QUERY_MAP = new LinkedHashMap<>();
private void setupQueries()
@@ -243,6 +223,33 @@ public class TimeseriesBenchmark
basicQueries.put("timeFilterByInterval", timeFilterQuery);
}
+ {
+ // One clause has a bitmap index, the other is a numeric column and has
none. OrFilter therefore unions the
+ // indexable clauses into a partial index and converts that into a value
matcher, which walks a bitmap iterator
+ // alongside the cursor offset. See
OrFilter#convertIndexToVectorValueMatcher.
+ QuerySegmentSpec intervalSpec = new
MultipleIntervalSegmentSpec(Collections.singletonList(basicSchema.getDataInterval()));
+
+ List<AggregatorFactory> queryAggs = new ArrayList<>();
+ queryAggs.add(new LongSumAggregatorFactory("sumLongSequential",
"sumLongSequential"));
+
+ DimFilter orFilter = new OrDimFilter(
+ new SelectorDimFilter("dimSequential", "311", null),
+ new BoundDimFilter("maxLongUniform", "100", null, false, false,
null, null, StringComparators.NUMERIC)
+ );
+
+ TimeseriesQuery orFilterQuery =
+ Druids.newTimeseriesQueryBuilder()
+ .dataSource("blah")
+ .granularity(Granularity.fromString(queryGranularity))
+ .intervals(intervalSpec)
+ .filters(orFilter)
+ .aggregators(queryAggs)
+ .descending(descending)
+ .context(Map.of("vectorize", vectorize))
+ .build();
+
+ basicQueries.put("orFilterPartialIndex", orFilterQuery);
+ }
SCHEMA_QUERY_MAP.put("basic", basicQueries);
@@ -268,13 +275,6 @@ public class TimeseriesBenchmark
schemaInfo = GeneratorBasicSchemas.SCHEMA_MAP.get(schemaName);
query = SCHEMA_QUERY_MAP.get(schemaName).get(queryName);
- generator = new DataGenerator(
- schemaInfo.getColumnSchemas(),
- RNG_SEED,
- schemaInfo.getDataInterval(),
- rowsPerSegment
- );
-
factory = new TimeseriesQueryRunnerFactory(
new TimeseriesQueryQueryToolChest(),
new TimeseriesQueryEngine(),
@@ -288,27 +288,31 @@ public class TimeseriesBenchmark
@State(Scope.Benchmark)
public static class IncrementalIndexState
{
- @Param({"onheap"})
- private String indexType;
+ private SegmentGenerator segmentGenerator;
IncrementalIndex incIndex;
@Setup
- public void setup(TimeseriesBenchmark global) throws
JsonProcessingException
+ public void setup(TimeseriesBenchmark global)
{
- // Creates an AppendableIndexSpec that corresponds to the indexType
parametrization.
- // It is used in {@code global.makeIncIndex()} to instanciate an
incremental-index of the specified type.
- global.appendableIndexSpec =
IncrementalIndexCreator.parseIndexType(indexType);
- incIndex = global.makeIncIndex();
- global.generator.addToIndex(incIndex, global.rowsPerSegment);
+ segmentGenerator = new SegmentGenerator();
+ incIndex = segmentGenerator.generateIncrementalIndex(
+ global.makeDataSegment(0),
+ global.schemaInfo,
+ Granularities.NONE,
+ global.rowsPerSegment
+ );
}
@TearDown
- public void tearDown()
+ public void tearDown() throws IOException
{
if (incIndex != null) {
incIndex.close();
}
+ if (segmentGenerator != null) {
+ segmentGenerator.close();
+ }
}
}
@@ -322,57 +326,54 @@ public class TimeseriesBenchmark
private int numSegments;
private ExecutorService executorService;
- private File qIndexesDir;
+ private SegmentGenerator segmentGenerator;
private List<QueryableIndex> qIndexes;
@Setup
- public void setup(TimeseriesBenchmark global) throws IOException
+ public void setup(TimeseriesBenchmark global)
{
- global.appendableIndexSpec = new OnheapIncrementalIndex.Spec();
-
executorService = Execs.multiThreaded(numSegments,
"TimeseriesThreadPool");
- qIndexesDir = FileUtils.createTempDir();
+ segmentGenerator = new SegmentGenerator();
qIndexes = new ArrayList<>();
for (int i = 0; i < numSegments; i++) {
log.info("Generating rows for segment " + i);
- IncrementalIndex incIndex = global.makeIncIndex();
- global.generator.reset(RNG_SEED + i).addToIndex(incIndex,
global.rowsPerSegment);
-
- File indexFile = INDEX_MERGER_V9.persist(
- incIndex,
- new File(qIndexesDir, String.valueOf(i)),
- IndexSpec.getDefault(),
- null
+ qIndexes.add(
+ segmentGenerator.generate(
+ global.makeDataSegment(i),
+ global.schemaInfo,
+ Granularities.NONE,
+ global.rowsPerSegment
+ )
);
- incIndex.close();
-
- qIndexes.add(INDEX_IO.loadIndex(indexFile));
}
}
@TearDown
- public void tearDown()
+ public void tearDown() throws IOException
{
for (QueryableIndex index : qIndexes) {
if (index != null) {
index.close();
}
}
- if (qIndexesDir != null) {
- qIndexesDir.delete();
+ if (segmentGenerator != null) {
+ segmentGenerator.close();
}
}
}
- private IncrementalIndex makeIncIndex()
+ private DataSegment makeDataSegment(final int segmentNumber)
{
- return appendableIndexSpec.builder()
- .setSimpleTestingIndexSchema(schemaInfo.getAggsArray())
- .setMaxRowCount(rowsPerSegment)
- .build();
+ return DataSegment.builder()
+ .dataSource("blah")
+ .interval(schemaInfo.getDataInterval())
+ .version("1")
+ .shardSpec(new LinearShardSpec(segmentNumber))
+ .size(0)
+ .build();
}
private static <T> List<T> runQuery(QueryRunnerFactory factory, QueryRunner
runner, Query<T> query)
diff --git
a/processing/src/main/java/org/apache/druid/collections/bitmap/SeekableRoaringIntIterator.java
b/processing/src/main/java/org/apache/druid/collections/bitmap/SeekableRoaringIntIterator.java
new file mode 100644
index 00000000000..cb46f629801
--- /dev/null
+++
b/processing/src/main/java/org/apache/druid/collections/bitmap/SeekableRoaringIntIterator.java
@@ -0,0 +1,179 @@
+/*
+ * 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.collections.bitmap;
+
+import org.roaringbitmap.PeekableCharIterator;
+import org.roaringbitmap.PeekableIntIterator;
+import org.roaringbitmap.buffer.DruidRoaringBufferAccess;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.roaringbitmap.buffer.MappeableContainer;
+import org.roaringbitmap.buffer.PointableRoaringArray;
+
+import java.util.NoSuchElementException;
+
+/**
+ * A {@link PeekableIntIterator} over an {@link ImmutableRoaringBitmap} that
repositions by finding the next container
+ * keys with {@link PointableRoaringArray#advanceUntil(char, int)}, which can
binary search when appropriate.
+ * The builtin iterator from {@link ImmutableRoaringBitmap#getIntIterator()}
is slower since it advances containers
+ * one at a time.
+ */
+public final class SeekableRoaringIntIterator implements PeekableIntIterator
+{
+ private final PointableRoaringArray highLowContainer;
+ private final int size;
+
+ /**
+ * Index of the container that {@link #iter} is reading.
+ */
+ private int index;
+
+ /**
+ * Key of the container at {@link #index}, shifted into the high bits.
+ */
+ private int shiftedKey;
+
+ /**
+ * Iterator over the (short) values of the current container.
+ */
+ private PeekableCharIterator iter;
+
+ /**
+ * Whether {@link #index} is in range. While true, {@code iter.hasNext()} is
also true.
+ */
+ private boolean ok;
+
+ /**
+ * The container behind {@link #index}, retained so that rewinding within it
does not have to materialize it again.
+ */
+ private MappeableContainer container;
+ private int containerIndex = -1;
+
+ public SeekableRoaringIntIterator(final ImmutableRoaringBitmap bitmap)
+ {
+ this.highLowContainer = DruidRoaringBufferAccess.highLowContainer(bitmap);
+ this.size = highLowContainer.size();
+ setContainer(0);
+ }
+
+ /**
+ * Positions this iterator so that {@link #peekNext()} returns the smallest
value >= target, or {@link #hasNext()}
+ * returns false if the bitmap holds no such value. Unlike {@link
#advanceIfNeeded}, target may be below the
+ * current position.
+ */
+ public void seek(final int target)
+ {
+ final int targetKey = target >>> 16;
+
+ if (ok && (shiftedKey >>> 16) == targetKey) {
+ if (iter.hasNext() && (iter.peekNext() & 0xFFFF) > (target & 0xFFFF)) {
+ // Attempting to seek within a container to a value earlier than the
current iter position. Need to rewrap.
+ setContainer(index);
+ }
+ } else if (ok && targetKey > (shiftedKey >>> 16)) {
+ // Moving forwards one or more container(s).
+ setContainer(highLowContainer.advanceUntil((char) targetKey, index));
+ } else {
+ // Moving backwards one or more container(s).
+ final int found = highLowContainer.getContainerIndex((char) targetKey);
+ // A miss returns -(insertionPoint + 1), and the insertion point is the
first container above targetKey.
+ setContainer(found >= 0 ? found : -found - 1);
+ }
+
+ if (ok && (shiftedKey >>> 16) == targetKey) {
+ iter.advanceIfNeeded((char) target);
+ if (!iter.hasNext()) {
+ setContainer(index + 1);
+ }
+ }
+ }
+
+ @Override
+ public void advanceIfNeeded(final int minval)
+ {
+ // Forward only: return early if minval is earlier than current iteration
state.
+ if (!ok || Integer.compareUnsigned(peekNext(), minval) >= 0) {
+ return;
+ }
+ seek(minval);
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ return ok;
+ }
+
+ @Override
+ public int next()
+ {
+ if (!ok) {
+ throw new NoSuchElementException();
+ }
+ final int x = iter.nextAsInt() | shiftedKey;
+ if (!iter.hasNext()) {
+ setContainer(index + 1);
+ }
+ return x;
+ }
+
+ @Override
+ public int peekNext()
+ {
+ if (!ok) {
+ throw new NoSuchElementException();
+ }
+ return iter.peekNext() | shiftedKey;
+ }
+
+ @Override
+ public SeekableRoaringIntIterator clone()
+ {
+ try {
+ final SeekableRoaringIntIterator cloned = (SeekableRoaringIntIterator)
super.clone();
+ if (iter != null) {
+ cloned.iter = iter.clone();
+ }
+ return cloned;
+ }
+ catch (CloneNotSupportedException e) {
+ throw new AssertionError(e);
+ }
+ }
+
+ /**
+ * Seeks to a new container, by index, and resets {@link #iter}.
+ */
+ private void setContainer(final int newIndex)
+ {
+ index = newIndex;
+ if (newIndex >= 0 && newIndex < size) {
+ if (newIndex != containerIndex) {
+ container = highLowContainer.getContainerAtIndex(newIndex);
+ containerIndex = newIndex;
+ shiftedKey = (highLowContainer.getKeyAtIndex(newIndex)) << 16;
+ }
+ iter = container.getCharIterator();
+ ok = iter.hasNext();
+ } else {
+ iter = null;
+ ok = false;
+ }
+ }
+}
diff --git
a/processing/src/main/java/org/apache/druid/collections/bitmap/WrappedImmutableRoaringBitmap.java
b/processing/src/main/java/org/apache/druid/collections/bitmap/WrappedImmutableRoaringBitmap.java
index 97f7b65cfe2..6448f955a00 100644
---
a/processing/src/main/java/org/apache/druid/collections/bitmap/WrappedImmutableRoaringBitmap.java
+++
b/processing/src/main/java/org/apache/druid/collections/bitmap/WrappedImmutableRoaringBitmap.java
@@ -81,7 +81,7 @@ public class WrappedImmutableRoaringBitmap implements
ImmutableBitmap
@Override
public PeekableIntIterator peekableIterator()
{
- return bitmap.getIntIterator();
+ return new SeekableRoaringIntIterator(bitmap);
}
@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 d0764ba492f..6a738091ec2 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
@@ -162,6 +162,9 @@ public interface ColumnarDoubles extends Closeable
private int id = ReadableVectorInspector.NULL_ID;
private PeekableIntIterator nullIterator =
nullValueBitmap.peekableIterator();
+ /**
+ * One past the highest row id of the previous batch, or -1 before the
first batch.
+ */
private int offsetMark = -1;
@Nullable
@@ -202,11 +205,10 @@ public interface ColumnarDoubles extends Closeable
ColumnarDoubles.this.get(doubleVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
- if (maxOffset < offsetMark) {
+ if (offsets[0] < offsetMark) {
nullIterator = nullValueBitmap.peekableIterator();
}
- offsetMark = maxOffset;
+ offsetMark = offsets[offset.getCurrentVectorSize() - 1] + 1;
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 1276f3294e7..539933aa1b4 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
@@ -157,6 +157,9 @@ public interface ColumnarFloats extends Closeable
private int id = ReadableVectorInspector.NULL_ID;
private PeekableIntIterator nullIterator =
nullValueBitmap.peekableIterator();
+ /**
+ * One past the highest row id of the previous batch, or -1 before the
first batch.
+ */
private int offsetMark = -1;
@Nullable
@@ -197,11 +200,10 @@ public interface ColumnarFloats extends Closeable
ColumnarFloats.this.get(floatVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
- if (maxOffset < offsetMark) {
+ if (offsets[0] < offsetMark) {
nullIterator = nullValueBitmap.peekableIterator();
}
- offsetMark = maxOffset;
+ offsetMark = offsets[offset.getCurrentVectorSize() - 1] + 1;
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 0c672195e8b..6351bed74ec 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
@@ -168,6 +168,9 @@ public interface ColumnarLongs extends Closeable
private int id = ReadableVectorInspector.NULL_ID;
private PeekableIntIterator nullIterator =
nullValueBitmap.peekableIterator();
+ /**
+ * One past the highest row id of the previous batch, or -1 before the
first batch.
+ */
private int offsetMark = -1;
@Nullable
@@ -208,11 +211,10 @@ public interface ColumnarLongs extends Closeable
ColumnarLongs.this.get(longVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
- if (maxOffset < offsetMark) {
+ if (offsets[0] < offsetMark) {
nullIterator = nullValueBitmap.peekableIterator();
}
- offsetMark = maxOffset;
+ offsetMark = offsets[offset.getCurrentVectorSize() - 1] + 1;
ColumnarLongs.this.get(longVector, offsets,
offset.getCurrentVectorSize());
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/filter/OrFilter.java
b/processing/src/main/java/org/apache/druid/segment/filter/OrFilter.java
index 53e37036336..9d633e7e041 100644
--- a/processing/src/main/java/org/apache/druid/segment/filter/OrFilter.java
+++ b/processing/src/main/java/org/apache/druid/segment/filter/OrFilter.java
@@ -372,7 +372,7 @@ public class OrFilter implements BooleanFilter
}
}
- private static VectorValueMatcher convertIndexToVectorValueMatcher(
+ static VectorValueMatcher convertIndexToVectorValueMatcher(
final ReadableVectorOffset vectorOffset,
final ImmutableBitmap bitmap
)
@@ -386,7 +386,10 @@ public class OrFilter implements BooleanFilter
{
final VectorMatch match = VectorMatch.wrap(new
int[vectorOffset.getMaxVectorSize()]);
int iterOffset = -1;
- int previousStartOffset = -1;
+ /**
+ * One past the highest row id of the previous batch.
+ */
+ int previousEndOffset = 0;
PeekableIntIterator iterator = initialIterator;
@Override
@@ -396,12 +399,12 @@ public class OrFilter implements BooleanFilter
if (vectorOffset.isContiguous()) {
final int startOffset = vectorOffset.getStartOffset();
- // check if the cursor was reset, and reset iterator if so
- if (startOffset <= previousStartOffset) {
+ // check if the cursor moved backwards, and reset iterator if so
+ if (startOffset < previousEndOffset) {
iterOffset = -1;
iterator = bitmap.peekableIterator();
}
- previousStartOffset = startOffset;
+ previousEndOffset = startOffset + getCurrentVectorSize();
int numRows = 0;
for (int i = 0; i < mask.getSelectionSize(); i++) {
final int maskNum = mask.getSelection()[i];
@@ -418,11 +421,14 @@ public class OrFilter implements BooleanFilter
return match;
} else {
final int[] currentOffsets = vectorOffset.getOffsets();
- if (getCurrentVectorSize() > 0 && currentOffsets[0] <=
previousStartOffset) {
- iterOffset = -1;
- iterator = bitmap.peekableIterator();
+ if (getCurrentVectorSize() > 0) {
+ // check if the cursor moved backwards, and reset iterator if so
+ if (currentOffsets[0] < previousEndOffset) {
+ iterOffset = -1;
+ iterator = bitmap.peekableIterator();
+ }
+ previousEndOffset = currentOffsets[getCurrentVectorSize() - 1] + 1;
}
- previousStartOffset = currentOffsets[0];
int numRows = 0;
for (int i = 0; i < mask.getSelectionSize(); i++) {
final int maskNum = mask.getSelection()[i];
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 43a2080d7a6..5acf77be664 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
@@ -879,8 +879,10 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
private boolean[] nullVector = null;
private int id = ReadableVectorInspector.NULL_ID;
- @Nullable
private PeekableIntIterator nullIterator =
nullBitmap.peekableIterator();
+ /**
+ * One past the highest row id of the previous batch, or -1 before
the first batch.
+ */
private int offsetMark = -1;
@Override
@@ -912,17 +914,14 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
longsColumn.get(valueVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
- if (maxOffset < offsetMark) {
+ if (offsets[0] < offsetMark) {
nullIterator = nullBitmap.peekableIterator();
}
- offsetMark = maxOffset;
+ offsetMark = offsets[offset.getCurrentVectorSize() - 1] + 1;
longsColumn.get(valueVector, offsets,
offset.getCurrentVectorSize());
}
- if (nullIterator != null) {
- nullVector = VectorSelectorUtils.populateNullVector(nullVector,
offset, nullIterator);
- }
+ nullVector = VectorSelectorUtils.populateNullVector(nullVector,
offset, nullIterator);
id = offset.getId();
}
@@ -935,8 +934,10 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
private boolean[] nullVector = null;
private int id = ReadableVectorInspector.NULL_ID;
- @Nullable
- private PeekableIntIterator nullIterator = nullBitmap != null ?
nullBitmap.peekableIterator() : null;
+ private PeekableIntIterator nullIterator =
nullBitmap.peekableIterator();
+ /**
+ * One past the highest row id of the previous batch, or -1 before
the first batch.
+ */
private int offsetMark = -1;
@Override
@@ -968,17 +969,14 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
doublesColumn.get(valueVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
- if (maxOffset < offsetMark) {
+ if (offsets[0] < offsetMark) {
nullIterator = nullBitmap.peekableIterator();
}
- offsetMark = maxOffset;
+ offsetMark = offsets[offset.getCurrentVectorSize() - 1] + 1;
doublesColumn.get(valueVector, offsets,
offset.getCurrentVectorSize());
}
- if (nullIterator != null) {
- nullVector = VectorSelectorUtils.populateNullVector(nullVector,
offset, nullIterator);
- }
+ nullVector = VectorSelectorUtils.populateNullVector(nullVector,
offset, nullIterator);
id = offset.getId();
}
@@ -995,8 +993,10 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
private boolean[] nullVector = null;
private int id = ReadableVectorInspector.NULL_ID;
- @Nullable
- private PeekableIntIterator nullIterator = nullBitmap != null ?
nullBitmap.peekableIterator() : null;
+ private PeekableIntIterator nullIterator =
nullBitmap.peekableIterator();
+ /**
+ * One past the highest row id of the previous batch, or -1 before the
first batch.
+ */
private int offsetMark = -1;
@Override
@@ -1028,11 +1028,10 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
column.get(idVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
- if (maxOffset < offsetMark) {
+ if (offsets[0] < offsetMark) {
nullIterator = nullBitmap.peekableIterator();
}
- offsetMark = maxOffset;
+ offsetMark = offsets[offset.getCurrentVectorSize() - 1] + 1;
column.get(idVector, offsets, offset.getCurrentVectorSize());
}
for (int i = 0; i < offset.getCurrentVectorSize(); i++) {
@@ -1047,9 +1046,7 @@ public class
NestedFieldDictionaryEncodedColumn<TStringDictionary extends Indexe
}
}
- if (nullIterator != null) {
- nullVector = VectorSelectorUtils.populateNullVector(nullVector,
offset, nullIterator);
- }
+ nullVector = VectorSelectorUtils.populateNullVector(nullVector,
offset, nullIterator);
id = offset.getId();
}
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 4252cf56342..acc8171f768 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
@@ -137,8 +137,10 @@ public class ScalarDoubleColumn implements
NestedCommonFormatColumn
private boolean[] nullVector = null;
private int id = ReadableVectorInspector.NULL_ID;
- @Nullable
- private PeekableIntIterator nullIterator = nullValueIndex != null ?
nullValueIndex.peekableIterator() : null;
+ private PeekableIntIterator nullIterator =
nullValueIndex.peekableIterator();
+ /**
+ * One past the highest row id of the previous batch, or -1 before the
first batch.
+ */
private int offsetMark = -1;
@Override
@@ -170,11 +172,10 @@ public class ScalarDoubleColumn implements
NestedCommonFormatColumn
valueColumn.get(valueVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
- if (maxOffset < offsetMark) {
+ if (offsets[0] < offsetMark) {
nullIterator = nullValueIndex.peekableIterator();
}
- offsetMark = maxOffset;
+ offsetMark = offsets[offset.getCurrentVectorSize() - 1] + 1;
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 9fa57b7c731..2ebd4f8bc7b 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
@@ -138,8 +138,10 @@ public class ScalarLongColumn implements
NestedCommonFormatColumn
private boolean[] nullVector = null;
private int id = ReadableVectorInspector.NULL_ID;
- @Nullable
private PeekableIntIterator nullIterator =
nullValueIndex.peekableIterator();
+ /**
+ * One past the highest row id of the previous batch, or -1 before the
first batch.
+ */
private int offsetMark = -1;
@Override
@@ -171,11 +173,10 @@ public class ScalarLongColumn implements
NestedCommonFormatColumn
valueColumn.get(valueVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
- if (maxOffset < offsetMark) {
+ if (offsets[0] < offsetMark) {
nullIterator = nullValueIndex.peekableIterator();
}
- offsetMark = maxOffset;
+ offsetMark = offsets[offset.getCurrentVectorSize() - 1] + 1;
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 1bbf3ffe26e..b6e1532d719 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
@@ -797,8 +797,10 @@ public class VariantColumn<TStringDictionary extends
Indexed<ByteBuffer>>
private boolean[] nullVector = null;
private int id = ReadableVectorInspector.NULL_ID;
- @Nullable
- private PeekableIntIterator nullIterator = nullValueBitmap != null ?
nullValueBitmap.peekableIterator() : null;
+ private PeekableIntIterator nullIterator =
nullValueBitmap.peekableIterator();
+ /**
+ * One past the highest row id of the previous batch, or -1 before the
first batch.
+ */
private int offsetMark = -1;
@Override
public double[] getDoubleVector()
@@ -829,11 +831,10 @@ public class VariantColumn<TStringDictionary extends
Indexed<ByteBuffer>>
encodedValueColumn.get(idVector, offset.getStartOffset(),
offset.getCurrentVectorSize());
} else {
final int[] offsets = offset.getOffsets();
- final int maxOffset = offsets[offset.getCurrentVectorSize() - 1];
- if (maxOffset < offsetMark) {
+ if (offsets[0] < offsetMark) {
nullIterator = nullValueBitmap.peekableIterator();
}
- offsetMark = maxOffset;
+ offsetMark = offsets[offset.getCurrentVectorSize() - 1] + 1;
encodedValueColumn.get(idVector, offsets,
offset.getCurrentVectorSize());
}
for (int i = 0; i < offset.getCurrentVectorSize(); i++) {
@@ -847,9 +848,7 @@ public class VariantColumn<TStringDictionary extends
Indexed<ByteBuffer>>
}
}
- if (nullIterator != null) {
- nullVector = VectorSelectorUtils.populateNullVector(nullVector,
offset, nullIterator);
- }
+ nullVector = VectorSelectorUtils.populateNullVector(nullVector,
offset, nullIterator);
id = offset.getId();
}
diff --git
a/processing/src/main/java/org/roaringbitmap/buffer/DruidRoaringBufferAccess.java
b/processing/src/main/java/org/roaringbitmap/buffer/DruidRoaringBufferAccess.java
new file mode 100644
index 00000000000..b26753a5c4c
--- /dev/null
+++
b/processing/src/main/java/org/roaringbitmap/buffer/DruidRoaringBufferAccess.java
@@ -0,0 +1,38 @@
+/*
+ * 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.
+ */
+
+//CHECKSTYLE.OFF: PackageName - Must be in RoaringBitmap to reach
ImmutableRoaringBitmap.highLowContainer
+
+package org.roaringbitmap.buffer;
+
+/**
+ * Exposes {@link ImmutableRoaringBitmap#highLowContainer}, which is
package-private. Every method on
+ * {@link PointableRoaringArray} is public; only the field holding it is not.
+ */
+public final class DruidRoaringBufferAccess
+{
+ private DruidRoaringBufferAccess()
+ {
+ }
+
+ public static PointableRoaringArray highLowContainer(final
ImmutableRoaringBitmap bitmap)
+ {
+ return bitmap.highLowContainer;
+ }
+}
diff --git
a/processing/src/test/java/org/apache/druid/collections/bitmap/SeekableRoaringIntIteratorTest.java
b/processing/src/test/java/org/apache/druid/collections/bitmap/SeekableRoaringIntIteratorTest.java
new file mode 100644
index 00000000000..293bd42b16d
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/collections/bitmap/SeekableRoaringIntIteratorTest.java
@@ -0,0 +1,597 @@
+/*
+ * 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.collections.bitmap;
+
+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.IntIterator;
+import org.roaringbitmap.PeekableIntIterator;
+import org.roaringbitmap.buffer.DruidRoaringBufferAccess;
+import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
+import org.roaringbitmap.buffer.MappeableArrayContainer;
+import org.roaringbitmap.buffer.MappeableBitmapContainer;
+import org.roaringbitmap.buffer.MappeableRunContainer;
+import org.roaringbitmap.buffer.MutableRoaringBitmap;
+import org.roaringbitmap.buffer.PointableRoaringArray;
+
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.NoSuchElementException;
+import java.util.Random;
+import java.util.Set;
+import java.util.stream.Stream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Checks {@link SeekableRoaringIntIterator} against {@link
ImmutableRoaringBitmap#getIntIterator()}.
+ *
+ * <p>Bitmap values are unsigned, so this test carries them around as longs in
{@code [0, 0xFFFFFFFF]} and uses
+ * {@link #NONE} for "the iterator is exhausted".
+ */
+public class SeekableRoaringIntIteratorTest
+{
+ /**
+ * Stands in for "no value", since every int is a legal bitmap value.
+ */
+ private static final long NONE = -1L;
+ private static final long UNSIGNED_LIMIT = 1L << 32;
+ private static final long RANDOM_SEED = 1234;
+
+ public static Stream<Arguments> bitmaps()
+ {
+ final List<Arguments> cases = new ArrayList<>();
+ final Random random = new Random(RANDOM_SEED);
+
+ // Sparse bitmaps. Should end up as array containers.
+ final MutableRoaringBitmap sparse = new MutableRoaringBitmap();
+ for (int i = 0; i < 3_000_000; i += 1 + random.nextInt(4000)) {
+ sparse.add(i);
+ }
+ cases.add(Arguments.of("sparse", toBufferBackedBitmap(sparse)));
+
+ // Dense bitmaps. Should end up as bitmap containers.
+ final MutableRoaringBitmap dense = new MutableRoaringBitmap();
+ for (int i = 0; i < 1_000_000; i++) {
+ if (random.nextDouble() < 0.6) {
+ dense.add(i);
+ }
+ }
+ cases.add(Arguments.of("dense", toBufferBackedBitmap(dense)));
+
+ // Runs, with whole containers missing in between. Should end up as run
containers.
+ final MutableRoaringBitmap runs = new MutableRoaringBitmap();
+ runs.add(200_000L, 400_000L);
+ runs.add(2_000_000L, 2_300_000L);
+ runs.add(9_000_000L, 9_010_000L);
+ cases.add(Arguments.of("runs", toBufferBackedBitmap(runs)));
+
+ // A single value far from the origin.
+ final MutableRoaringBitmap lonely = new MutableRoaringBitmap();
+ lonely.add(15_000_000);
+ cases.add(Arguments.of("lonely", toBufferBackedBitmap(lonely)));
+
+ // Backed by a MutableRoaringArray rather than ByteBuffer, which is what
+ // WrappedRoaringBitmap.toImmutableBitmap() hands out.
+ final MutableRoaringBitmap heap = new MutableRoaringBitmap();
+ heap.add(50L, 70L);
+ heap.add(300_000L, 305_000L);
+ heap.add(5_000_000L, 5_000_100L);
+ heap.runOptimize();
+ cases.add(Arguments.of("heap", heap.toImmutableRoaringBitmap()));
+
+ // All three kinds of distributions in one bitmap.
+ final MutableRoaringBitmap mixed = new MutableRoaringBitmap();
+ for (int i = 0; i < 100; i++) {
+ mixed.add(i * 37);
+ }
+ for (int i = 0; i < 65536; i++) {
+ if (random.nextDouble() < 0.5) {
+ mixed.add(65536 + i);
+ }
+ }
+ mixed.add(2 * 65536L, 3 * 65536L);
+ mixed.add(4 * 65536 + 11);
+ mixed.add(4 * 65536 + 65535);
+ cases.add(Arguments.of("mixed", toBufferBackedBitmap(mixed)));
+
+ // Consecutive container keys.
+ final MutableRoaringBitmap adjacent = new MutableRoaringBitmap();
+ for (int key = 0; key < 10; key++) {
+ for (int i = 0; i < 20; i++) {
+ adjacent.add(key * 65536 + i * 3000);
+ }
+ }
+ cases.add(Arguments.of("adjacent", toBufferBackedBitmap(adjacent)));
+
+ // The first and last value of several containers, so seeks land exactly
on container edges.
+ final MutableRoaringBitmap boundaries = new MutableRoaringBitmap();
+ for (int key = 0; key < 5; key++) {
+ boundaries.add(key * 65536);
+ boundaries.add(key * 65536 + 65535);
+ }
+ cases.add(Arguments.of("boundaries", toBufferBackedBitmap(boundaries)));
+
+ // Values above Integer.MAX_VALUE, where the container key has its high
bit set and hs is negative.
+ final MutableRoaringBitmap high = new MutableRoaringBitmap();
+ high.add(1);
+ high.add(0x7FFFFFFF);
+ high.add(0x80000000);
+ high.add(0x80000001);
+ high.add(0xC0000000L, 0xC0010000L);
+ high.add(0xFFFF0000L, 0xFFFF0010L);
+ high.add(-1); // 0xFFFFFFFF, the largest unsigned value
+ cases.add(Arguments.of("high", toBufferBackedBitmap(high)));
+
+ return cases.stream();
+ }
+
+ @Test
+ public void testFixturesCoverEveryContainerKind()
+ {
+ // Verify we really do have all three kinds of containers in the test
bitmaps.
+ final Set<String> kinds = new HashSet<>();
+ bitmaps().forEach(args -> {
+ final ImmutableRoaringBitmap bitmap = (ImmutableRoaringBitmap)
args.get()[1];
+ final PointableRoaringArray containers =
DruidRoaringBufferAccess.highLowContainer(bitmap);
+ for (int i = 0; i < containers.size(); i++) {
+
kinds.add(containers.getContainerAtIndex(i).getClass().getSimpleName());
+ }
+ });
+
+ assertTrue(kinds.contains(MappeableArrayContainer.class.getSimpleName()),
"array containers, got " + kinds);
+ assertTrue(kinds.contains(MappeableBitmapContainer.class.getSimpleName()),
"bitmap containers, got " + kinds);
+ assertTrue(kinds.contains(MappeableRunContainer.class.getSimpleName()),
"run containers, got " + kinds);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testSequentialIteration(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ final IntIterator reference = bitmap.getIntIterator();
+ long n = 0;
+ while (reference.hasNext()) {
+ assertTrue(iterator.hasNext(), "ran out early at " + n);
+ assertEquals(reference.next(), iterator.next(), "value " + n);
+ n++;
+ }
+ assertFalse(iterator.hasNext(), "extra values after " + n);
+ assertThrows(NoSuchElementException.class, iterator::next);
+ assertThrows(NoSuchElementException.class, iterator::peekNext);
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testPeekNextDoesNotAdvance(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ final IntIterator reference = bitmap.getIntIterator();
+ while (reference.hasNext()) {
+ final int value = reference.next();
+ assertEquals(value, iterator.peekNext());
+ assertEquals(value, iterator.peekNext());
+ assertEquals(value, iterator.next());
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testRandomSeeksInBothDirections(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ final Random random = new Random(5678);
+ final long span = span(bitmap);
+ for (int trial = 0; trial < 200_000; trial++) {
+ final int target = (int) random.nextLong(span);
+ iterator.seek(target);
+ assertEquals(expected(bitmap, target), peek(iterator), "seek(" +
Integer.toUnsignedString(target) + ")");
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testAscendingSeeks(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ final long span = span(bitmap);
+ final long step = Math.max(1, span / 20_000);
+ for (long target = 0; target < span; target += step) {
+ iterator.seek((int) target);
+ assertEquals(expected(bitmap, (int) target), peek(iterator), "seek(" +
target + ")");
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testDescendingSeeks(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ final long span = span(bitmap);
+ final long step = Math.max(1, span / 20_000);
+ for (long target = span - 1; target >= 0; target -= step) {
+ iterator.seek((int) target);
+ assertEquals(expected(bitmap, (int) target), peek(iterator), "seek(" +
target + ")");
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testSeekIsIdempotent(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ final Random random = new Random(2468);
+ final long span = span(bitmap);
+ for (int trial = 0; trial < 20_000; trial++) {
+ final int target = (int) random.nextLong(span);
+ iterator.seek(target);
+ final long first = peek(iterator);
+ iterator.seek(target);
+ assertEquals(first, peek(iterator), "second seek(" +
Integer.toUnsignedString(target) + ")");
+ assertEquals(expected(bitmap, target), first, "seek(" +
Integer.toUnsignedString(target) + ")");
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testInterleavedNextAndAdvanceIfNeeded(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ final PeekableIntIterator reference = bitmap.getIntIterator();
+ final Random random = new Random(4321);
+ long cursor = 0;
+ while (reference.hasNext() && iterator.hasNext()) {
+ if (random.nextBoolean()) {
+ assertEquals(reference.next(), iterator.next(), "next at " + cursor);
+ } else {
+ // advanceIfNeeded is forward only, so never ask for less than where
the iterator already is.
+ cursor = Math.max(cursor, peek(iterator)) + random.nextInt(20000);
+ if (cursor >= UNSIGNED_LIMIT) {
+ break;
+ }
+ reference.advanceIfNeeded((int) cursor);
+ iterator.advanceIfNeeded((int) cursor);
+ assertEquals(peek(reference), peek(iterator), "advanceIfNeeded(" +
cursor + ")");
+ }
+ }
+ assertEquals(reference.hasNext(), iterator.hasNext());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testRandomOperationsMatchReference(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final Random random = new Random(13579);
+ final long span = span(bitmap);
+ SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ PeekableIntIterator reference = bitmap.getIntIterator();
+
+ for (int op = 0; op < 100_000; op++) {
+ final int choice = random.nextInt(10);
+ final String what;
+ if (choice < 4) {
+ what = "next";
+ assertEquals(reference.hasNext(), iterator.hasNext(), "hasNext before
" + what + " at op " + op);
+ if (reference.hasNext()) {
+ assertEquals(reference.next(), iterator.next(), "next at op " + op);
+ } else {
+ assertThrows(NoSuchElementException.class, iterator::next, "next at
op " + op);
+ }
+ } else if (choice < 7) {
+ // Forward only, per the advanceIfNeeded contract. An exhausted
iterator accepts anything, since it must
+ // stay exhausted either way.
+ final long from = Math.max(0, peek(iterator));
+ final long minval = from + random.nextLong(span - from);
+ what = "advanceIfNeeded(" + minval + ")";
+ reference.advanceIfNeeded((int) minval);
+ iterator.advanceIfNeeded((int) minval);
+ } else if (choice < 9) {
+ final long target = random.nextLong(span);
+ what = "seek(" + target + ")";
+ iterator.seek((int) target);
+ reference = bitmap.getIntIterator();
+ reference.advanceIfNeeded((int) target);
+ } else {
+ what = "clone";
+ iterator = iterator.clone();
+ reference = reference.clone();
+ }
+ assertEquals(peek(reference), peek(iterator), what + " at op " + op);
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testCloneIsIndependent(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator original = new
SeekableRoaringIntIterator(bitmap);
+ for (int i = 0; i < 100 && original.hasNext(); i++) {
+ original.next();
+ }
+ final SeekableRoaringIntIterator copy = original.clone();
+ assertEquals(original.hasNext(), copy.hasNext());
+ while (original.hasNext()) {
+ assertTrue(copy.hasNext());
+ assertEquals(original.next(), copy.next());
+ }
+ assertFalse(copy.hasNext());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testCloneDoesNotShareCursor(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator original = new
SeekableRoaringIntIterator(bitmap);
+ final long first = peek(original);
+
+ // Draining the copy leaves the original untouched.
+ final SeekableRoaringIntIterator copy = original.clone();
+ while (copy.hasNext()) {
+ copy.next();
+ }
+ assertFalse(copy.hasNext());
+ assertEquals(first, peek(original));
+
+ // And seeking the original leaves an earlier copy untouched.
+ final SeekableRoaringIntIterator pinned = original.clone();
+ final int last = bitmap.last();
+ original.seek(last);
+ assertEquals(Integer.toUnsignedLong(last), peek(original));
+ assertEquals(first, peek(pinned));
+
+ // A clone of an exhausted iterator is exhausted, and can be brought back
on its own.
+ original.next();
+ assertFalse(original.hasNext());
+ final SeekableRoaringIntIterator exhausted = original.clone();
+ assertFalse(exhausted.hasNext());
+ assertThrows(NoSuchElementException.class, exhausted::next);
+ exhausted.seek(0);
+ assertEquals(first, peek(exhausted));
+ assertFalse(original.hasNext());
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testAdvanceIfNeededDoesNotMoveBackwards(final String name, final
ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ final Random random = new Random(97531);
+
+ for (int trial = 0; trial < 20_000 && iterator.hasNext(); trial++) {
+ final long position = peek(iterator);
+ iterator.advanceIfNeeded((int) random.nextLong(position + 1));
+ assertEquals(position, peek(iterator), "backwards advanceIfNeeded from "
+ position);
+ iterator.next();
+ }
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("bitmaps")
+ public void testAdvanceIfNeededOnExhaustedIteratorIsANoOp(final String name,
final ImmutableRoaringBitmap bitmap)
+ {
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ final int first = iterator.peekNext();
+ while (iterator.hasNext()) {
+ iterator.next();
+ }
+
+ final long span = span(bitmap);
+ final Random random = new Random(11223);
+ for (int trial = 0; trial < 1000; trial++) {
+ iterator.advanceIfNeeded((int) random.nextLong(span));
+ assertFalse(iterator.hasNext(), "still exhausted");
+ }
+ iterator.advanceIfNeeded(0);
+ assertFalse(iterator.hasNext(), "still exhausted");
+
+ // seek() is not bound by the forward-only rule, so it does come back.
+ iterator.seek(0);
+ assertEquals(first, iterator.peekNext());
+ }
+
+ @Test
+ public void testAdvanceIntoGapLandsAtStartOfNextRange()
+ {
+ final MutableRoaringBitmap mutable = new MutableRoaringBitmap();
+ mutable.add(2_000_000L, 2_200_000L);
+ mutable.add(4_000_000L, 4_300_000L);
+ final ImmutableRoaringBitmap bitmap = toBufferBackedBitmap(mutable);
+
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+
+ assertEquals(2_000_000, iterator.next());
+
+ iterator.advanceIfNeeded(2_100_000);
+ assertEquals(2_100_000, iterator.next());
+
+ assertFalse(bitmap.contains(2_300_000));
+ iterator.advanceIfNeeded(2_300_000);
+ assertEquals(4_000_000, iterator.peekNext());
+
+ iterator.advanceIfNeeded(4_000_000);
+ assertEquals(4_000_000, iterator.next());
+ }
+
+ @Test
+ public void testSeekIntoWholeContainerGap()
+ {
+ // Targets in gaps that span whole containers, reached from both
directions.
+ final MutableRoaringBitmap mutable = new MutableRoaringBitmap();
+ mutable.add(5);
+ mutable.add(10 * 65536 + 7);
+ mutable.add(50 * 65536 + 9);
+ final ImmutableRoaringBitmap bitmap = toBufferBackedBitmap(mutable);
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+
+ final int[] targets = {
+ 0,
+ 5,
+ 6,
+ 65536,
+ 3 * 65536,
+ 10 * 65536,
+ 10 * 65536 + 7,
+ 10 * 65536 + 8,
+ 11 * 65536,
+ 49 * 65536,
+ 50 * 65536 + 9,
+ 50 * 65536 + 10,
+ 60 * 65536
+ };
+
+ for (final int target : targets) {
+ iterator.seek(target);
+ assertEquals(expected(bitmap, target), peek(iterator), "ascending seek("
+ target + ")");
+ }
+ for (int i = targets.length - 1; i >= 0; i--) {
+ iterator.seek(targets[i]);
+ assertEquals(expected(bitmap, targets[i]), peek(iterator), "descending
seek(" + targets[i] + ")");
+ }
+ }
+
+ @Test
+ public void testEmptyBitmap()
+ {
+ final ImmutableRoaringBitmap bitmap = toBufferBackedBitmap(new
MutableRoaringBitmap());
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+ assertFalse(iterator.hasNext());
+ iterator.seek(0);
+ assertFalse(iterator.hasNext());
+ iterator.seek(1_000_000);
+ assertFalse(iterator.hasNext());
+ iterator.advanceIfNeeded(0);
+ assertFalse(iterator.hasNext());
+ assertThrows(NoSuchElementException.class, iterator::next);
+ assertThrows(NoSuchElementException.class, iterator::peekNext);
+ assertFalse(iterator.clone().hasNext());
+ }
+
+ @Test
+ public void testSeekPastEndThenBack()
+ {
+ final MutableRoaringBitmap mutable = new MutableRoaringBitmap();
+ mutable.add(100L, 200L);
+ final ImmutableRoaringBitmap bitmap = toBufferBackedBitmap(mutable);
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+
+ iterator.seek(1_000_000);
+ assertFalse(iterator.hasNext());
+
+ iterator.seek(150);
+ assertEquals(150, iterator.peekNext());
+
+ iterator.seek(0);
+ assertEquals(100, iterator.peekNext());
+ }
+
+ @Test
+ public void testUnsignedValuesAboveIntegerMaxValue()
+ {
+ final MutableRoaringBitmap mutable = new MutableRoaringBitmap();
+ mutable.add(1);
+ mutable.add(0x7FFFFFFF);
+ mutable.add(0x80000000);
+ mutable.add(0xFFFFFFF0L, 0x100000000L);
+ final ImmutableRoaringBitmap bitmap = toBufferBackedBitmap(mutable);
+ final SeekableRoaringIntIterator iterator = new
SeekableRoaringIntIterator(bitmap);
+
+ assertEquals(1, iterator.next());
+ assertEquals(0x7FFFFFFF, iterator.next());
+ assertEquals(0x80000000, iterator.next());
+ assertEquals(0xFFFFFFF0, iterator.next());
+
+ // Forwards across the sign boundary.
+ iterator.seek(0x7FFFFFFF);
+ assertEquals(0x7FFFFFFF, iterator.peekNext());
+ iterator.advanceIfNeeded(0x80000000);
+ assertEquals(0x80000000, iterator.peekNext());
+
+ // Backwards across it too, which advanceIfNeeded would refuse.
+ iterator.seek(0);
+ assertEquals(1, iterator.peekNext());
+
+ // The very top of the range.
+ iterator.seek(0xFFFFFFFF);
+ assertEquals(0xFFFFFFFF, iterator.peekNext());
+ assertEquals(0xFFFFFFFF, iterator.next());
+ assertFalse(iterator.hasNext());
+
+ // A gap between the sign boundary and the top containers.
+ iterator.seek(0x90000000);
+ assertEquals(0xFFFFFFF0, iterator.peekNext());
+ }
+
+ @Test
+ public void testWrappedImmutableRoaringBitmapHandsOutSeekableIterator()
+ {
+ final MutableRoaringBitmap mutable = new MutableRoaringBitmap();
+ mutable.add(1L, 100L);
+ final ImmutableBitmap wrapped = new
WrappedImmutableRoaringBitmap(toBufferBackedBitmap(mutable));
+ assertInstanceOf(SeekableRoaringIntIterator.class,
wrapped.peekableIterator());
+ }
+
+ /**
+ * Serializes through a ByteBuffer so the bitmap is backed by an
ImmutableRoaringArray.
+ */
+ private static ImmutableRoaringBitmap toBufferBackedBitmap(final
MutableRoaringBitmap mutable)
+ {
+ mutable.runOptimize();
+ final ByteBuffer buffer =
ByteBuffer.allocate(mutable.serializedSizeInBytes())
+ .order(ByteOrder.LITTLE_ENDIAN);
+ mutable.serialize(buffer);
+ buffer.flip();
+ return new ImmutableRoaringBitmap(buffer);
+ }
+
+ /**
+ * Get the expected result for seeking an iterator to {@code target} and
then peeking at the next value.
+ */
+ private static long expected(final ImmutableRoaringBitmap bitmap, final int
target)
+ {
+ final PeekableIntIterator reference = bitmap.getIntIterator();
+ reference.advanceIfNeeded(target);
+ return peek(reference);
+ }
+
+ /**
+ * Peek at the next value, and return it as an unsigned int, or return
{@link #NONE} if it is exhausted.
+ */
+ private static long peek(final PeekableIntIterator iterator)
+ {
+ return iterator.hasNext() ? Integer.toUnsignedLong(iterator.peekNext()) :
NONE;
+ }
+
+ /**
+ * One past the highest value in the bitmap.
+ */
+ private static long span(final ImmutableRoaringBitmap bitmap)
+ {
+ return Integer.toUnsignedLong(bitmap.last()) + 1;
+ }
+}
diff --git
a/processing/src/test/java/org/apache/druid/segment/filter/OrFilterVectorMatcherTest.java
b/processing/src/test/java/org/apache/druid/segment/filter/OrFilterVectorMatcherTest.java
new file mode 100644
index 00000000000..27f78e7acb8
--- /dev/null
+++
b/processing/src/test/java/org/apache/druid/segment/filter/OrFilterVectorMatcherTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.filter;
+
+import org.apache.druid.collections.bitmap.MutableBitmap;
+import org.apache.druid.collections.bitmap.RoaringBitmapFactory;
+import org.apache.druid.query.filter.vector.ReadableVectorMatch;
+import org.apache.druid.query.filter.vector.VectorMatch;
+import org.apache.druid.query.filter.vector.VectorValueMatcher;
+import org.apache.druid.segment.vector.ReadableVectorOffset;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Stream;
+
+public class OrFilterVectorMatcherTest
+{
+ private static final int[] INDEXED_ROWS = new int[]{3, 10, 13, 21};
+
+ public static Stream<Arguments> batches()
+ {
+ return Stream.of(
+ Arguments.of("ascending", new int[][]{{1, 2, 3}, {10, 11, 12}, {13,
20}}, List.of(3, 10, 13)),
+ Arguments.of("descendingRanges", new int[][]{{20, 21}, {10, 11}, {1,
3}}, List.of(21, 10, 3)),
+ Arguments.of("overlappingRanges", new int[][]{{1, 10}, {2, 3, 11},
{12, 13}}, List.of(10, 3, 13))
+ );
+ }
+
+ @ParameterizedTest(name = "{0}")
+ @MethodSource("batches")
+ public void testMatchesIndexedRows(String name, int[][] batches,
List<Integer> expected)
+ {
+ final MutableBitmap mutableBitmap =
RoaringBitmapFactory.INSTANCE.makeEmptyMutableBitmap();
+ for (final int row : INDEXED_ROWS) {
+ mutableBitmap.add(row);
+ }
+
+ final BatchedVectorOffset offset = new BatchedVectorOffset(batches);
+ final VectorValueMatcher matcher =
OrFilter.convertIndexToVectorValueMatcher(
+ offset,
+ RoaringBitmapFactory.INSTANCE.makeImmutableBitmap(mutableBitmap)
+ );
+
+ final List<Integer> matched = new ArrayList<>();
+ for (; offset.batchIndex < batches.length; offset.batchIndex++) {
+ final ReadableVectorMatch match =
matcher.match(VectorMatch.allTrue(offset.getCurrentVectorSize()), false);
+ for (int i = 0; i < match.getSelectionSize(); i++) {
+ matched.add(offset.getOffsets()[match.getSelection()[i]]);
+ }
+ }
+
+ Assertions.assertEquals(expected, matched);
+ }
+
+ /**
+ * A {@link ReadableVectorOffset} that hands back a predetermined list of
batches. Each batch must be internally
+ * ascending, but batches are otherwise unconstrained.
+ */
+ private static class BatchedVectorOffset implements ReadableVectorOffset
+ {
+ private final int[][] batches;
+ private final int maxVectorSize;
+
+ private int batchIndex = 0;
+
+ BatchedVectorOffset(final int[][] batches)
+ {
+ this.batches = batches;
+ int max = 0;
+ for (final int[] batch : batches) {
+ max = Math.max(max, batch.length);
+ }
+ this.maxVectorSize = max;
+ }
+
+ @Override
+ public boolean isContiguous()
+ {
+ return false;
+ }
+
+ @Override
+ public int getStartOffset()
+ {
+ throw new UnsupportedOperationException("not contiguous");
+ }
+
+ @Override
+ public int[] getOffsets()
+ {
+ return batches[batchIndex];
+ }
+
+ @Override
+ public int getId()
+ {
+ return batchIndex;
+ }
+
+ @Override
+ public int getMaxVectorSize()
+ {
+ return maxVectorSize;
+ }
+
+ @Override
+ public int getCurrentVectorSize()
+ {
+ return batches[batchIndex].length;
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
