You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@parquet.apache.org by ga...@apache.org on 2026/10/12 03:10:35 UTC
(parquet-java) branch master updated: GH-3772: Bulk skip in RunLengthBitPackingHybridDecoder / DictionaryValuesReader (#3773)
This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/parquet-java.git
The following commit(s) were added to refs/heads/master by this push:
new 3051aa0b8 GH-3772: Bulk skip in RunLengthBitPackingHybridDecoder / DictionaryValuesReader (#3773)
3051aa0b8 is described below
commit 3051aa0b8caede2acd8fa1896e92ffefe1320857
Author: Bodor Laszlo <bo...@gmail.com>
AuthorDate: Mon Oct 12 05:10:30 2026 +0200
GH-3772: Bulk skip in RunLengthBitPackingHybridDecoder / DictionaryValuesReader (#3773)
### Rationale for this change
`ValuesReader.skip(int n)` defaults to a naive loop of `skip()` calls. For dictionary-encoded columns each `skip()` bottoms out in `RunLengthBitPackingHybridDecoder.readInt()` — a mode switch, array indexing, and a value the caller immediately discards. Any filter-then-skip path (column-index row ranges, hash-join probe filtering, runtime filters) pays this cost per skipped row.
`RleSkipBenchmark`, JMH throughput, JDK 17, 100 k values/op:
| pattern | bitWidth | `readInt()` loop | `skipInts` | speedup |
|---------|---------:|-----------------:|-----------:|--------:|
| rle | 8 | 1.38 B/s | 104.9 B/s | ~76× |
| packed | 8 | 0.89 B/s | 4.12 B/s | ~4.6× |
| mixed | 8 | 0.96 B/s | 6.99 B/s | ~7.2× |
### What changes are included in this PR?
- New `RunLengthBitPackingHybridDecoder.skipInts(int)` — re-uses `readNext()` per run, then advances `currentCount` by `min(n, currentCount)` instead of walking every value through `readInt()`.
- `skip(int)` overrides on `DictionaryValuesReader` and `RunLengthBitPackingHybridValuesReader` delegating to `decoder.skipInts(n)`.
- `parquet-benchmarks / RleSkipBenchmark` JMH benchmark.
`readNext()` and the `ValuesReader.skip(int)` default are unchanged.
### Are these changes tested?
`TestRunLengthBitPackingHybridDecoderSkip` covers RLE-only, PACKED-only, mixed, zero-skip, full-skip, mid-run partial skip, `bitWidth = 0`, and a randomised 4 K-value read/skip interleaving cross-checked against a `readInt()` reference. All `parquet-column` tests pass.
### Are there any user-facing changes?
No. Additive and binary-compatible: no signatures change; `skipInts(n)` is semantically identical to N discarded `readInt()`s. Existing callers of `ValuesReader.skip(int)` pick up the fast path with no code change.
---
.../parquet/benchmarks/RleSkipBenchmark.java | 136 +++++++++++++
.../values/dictionary/DictionaryValuesReader.java | 28 ++-
.../rle/RunLengthBitPackingHybridDecoder.java | 27 +++
.../rle/RunLengthBitPackingHybridValuesReader.java | 9 +
.../column/values/dictionary/TestDictionary.java | 24 +++
.../TestRunLengthBitPackingHybridDecoderSkip.java | 224 +++++++++++++++++++++
6 files changed, 447 insertions(+), 1 deletion(-)
diff --git a/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/RleSkipBenchmark.java b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/RleSkipBenchmark.java
new file mode 100644
index 000000000..b51111e6e
--- /dev/null
+++ b/parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/RleSkipBenchmark.java
@@ -0,0 +1,136 @@
+/*
+ * 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.parquet.benchmarks;
+
+import java.nio.ByteBuffer;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridDecoder;
+import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridEncoder;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OperationsPerInvocation;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+/**
+ * Measures {@link RunLengthBitPackingHybridDecoder#skipInts(int)} vs. the equivalent
+ * discard-via-readInt loop. Both paths decode each run the same way; the win comes from
+ * dropping the per-value mode-switch, array-index arithmetic and method-call overhead of
+ * {@code readInt()} in favour of a single {@code currentCount -= consume} per run.
+ *
+ * <p>Parameters:
+ * <ul>
+ * <li>{@link #bitWidth} -- key width, chosen at 3/8/16 to bracket typical dictionary-index
+ * widths.</li>
+ * <li>{@link #pattern} -- {@code rle} produces long runs of a single value (best case for both
+ * paths); {@code packed} produces mostly distinct values so the encoder emits bit-packed
+ * groups (worst case for the old skip); {@code mixed} interleaves the two.</li>
+ * </ul>
+ *
+ * Each invocation re-wraps a pre-encoded byte buffer and calls {@code skipInts(VALUE_COUNT)} in
+ * the {@code skip} benchmark, versus a {@code VALUE_COUNT}-long {@code readInt()} discard loop
+ * in the {@code readSkip} benchmark.
+ */
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Fork(1)
+@Warmup(iterations = 3, time = 1)
+@Measurement(iterations = 5, time = 1)
+@State(Scope.Thread)
+public class RleSkipBenchmark {
+
+ static final int VALUE_COUNT = 100_000;
+ private static final int INIT_SLAB = 64 * 1024;
+ private static final int PAGE = 4 * 1024 * 1024;
+
+ @Param({"3", "8", "16"})
+ public int bitWidth;
+
+ @Param({"rle", "packed", "mixed"})
+ public String pattern;
+
+ private byte[] encoded;
+
+ @Setup(Level.Trial)
+ public void setup() throws Exception {
+ int mask = bitWidth == 32 ? -1 : ((1 << bitWidth) - 1);
+ RunLengthBitPackingHybridEncoder enc =
+ new RunLengthBitPackingHybridEncoder(bitWidth, INIT_SLAB, PAGE, new DirectByteBufferAllocator());
+ Random r = new Random(42);
+ switch (pattern) {
+ case "rle":
+ // Long stretches of the same value -> the encoder emits RLE runs.
+ for (int i = 0; i < VALUE_COUNT; i++) {
+ enc.writeInt(((i / 500) & 0x1F) & mask);
+ }
+ break;
+ case "packed":
+ // Random values -> the encoder emits bit-packed groups.
+ for (int i = 0; i < VALUE_COUNT; i++) {
+ enc.writeInt(r.nextInt() & mask);
+ }
+ break;
+ case "mixed":
+ // Alternating 250-value RLE blocks and 250-value random blocks.
+ for (int block = 0; block * 250 < VALUE_COUNT; block++) {
+ int val = r.nextInt() & mask;
+ boolean rle = (block & 1) == 0;
+ int limit = Math.min(250, VALUE_COUNT - block * 250);
+ for (int j = 0; j < limit; j++) {
+ enc.writeInt(rle ? val : (r.nextInt() & mask));
+ }
+ }
+ break;
+ default:
+ throw new IllegalArgumentException("unknown pattern: " + pattern);
+ }
+ encoded = enc.toBytes().toByteArray();
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(VALUE_COUNT)
+ public void skip() throws Exception {
+ ByteBufferInputStream in = ByteBufferInputStream.wrap(ByteBuffer.wrap(encoded));
+ RunLengthBitPackingHybridDecoder dec = new RunLengthBitPackingHybridDecoder(bitWidth, in);
+ dec.skipInts(VALUE_COUNT);
+ }
+
+ @Benchmark
+ @OperationsPerInvocation(VALUE_COUNT)
+ public int readSkip() throws Exception {
+ ByteBufferInputStream in = ByteBufferInputStream.wrap(ByteBuffer.wrap(encoded));
+ RunLengthBitPackingHybridDecoder dec = new RunLengthBitPackingHybridDecoder(bitWidth, in);
+ int sink = 0;
+ for (int i = 0; i < VALUE_COUNT; i++) {
+ sink ^= dec.readInt();
+ }
+ return sink;
+ }
+}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/dictionary/DictionaryValuesReader.java b/parquet-column/src/main/java/org/apache/parquet/column/values/dictionary/DictionaryValuesReader.java
index 53fafc55d..8a98d29c7 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/values/dictionary/DictionaryValuesReader.java
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/dictionary/DictionaryValuesReader.java
@@ -35,6 +35,8 @@ import org.slf4j.LoggerFactory;
public class DictionaryValuesReader extends ValuesReader {
private static final Logger LOG = LoggerFactory.getLogger(DictionaryValuesReader.class);
+ private static final String EMPTY_PAGE_MESSAGE = "Attempt to read from empty page";
+
private ByteBufferInputStream in;
private Dictionary dictionary;
@@ -57,7 +59,20 @@ public class DictionaryValuesReader extends ValuesReader {
decoder = new RunLengthBitPackingHybridDecoder(1, in) {
@Override
public int readInt() throws IOException {
- throw new IOException("Attempt to read from empty page");
+ throw new IOException(EMPTY_PAGE_MESSAGE);
+ }
+
+ /**
+ * Skipping goes through the decoder directly instead of {@link #readInt()}, so it needs the
+ * same guard: without it, readNext() would hit the empty stream and throw a raw
+ * IllegalArgumentException that {@link DictionaryValuesReader#skip(int)} does not wrap.
+ * Skipping zero values reads nothing and stays silent.
+ */
+ @Override
+ public void skipInts(int n) throws IOException {
+ if (n > 0) {
+ throw new IOException(EMPTY_PAGE_MESSAGE);
+ }
}
};
}
@@ -125,4 +140,15 @@ public class DictionaryValuesReader extends ValuesReader {
throw new ParquetDecodingException(e);
}
}
+
+ @Override
+ public void skip(int n) {
+ // Bulk-skip dictionary keys without decoding them or looking them up in the dictionary.
+ // See RunLengthBitPackingHybridDecoder#skipInts for the fast-path details.
+ try {
+ decoder.skipInts(n);
+ } catch (IOException e) {
+ throw new ParquetDecodingException(e);
+ }
+ }
}
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridDecoder.java b/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridDecoder.java
index 0a798ef67..2f143d2a0 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridDecoder.java
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridDecoder.java
@@ -80,6 +80,33 @@ public class RunLengthBitPackingHybridDecoder {
return result;
}
+ /**
+ * Skip {@code n} values without returning them. Runs are decoded normally, but the values are
+ * dropped instead of being handed back one-by-one via {@link #readInt()}, which saves the
+ * per-value branch on {@link #mode} and the array-index arithmetic that {@code readInt()} does.
+ *
+ * <p>Intended for callers that filter rows (column-index row ranges, hash-join probe filtering,
+ * etc.) and need to advance past many values on a dictionary-key or level column cheaply.
+ *
+ * @param n number of values to skip; must be non-negative
+ */
+ public void skipInts(int n) throws IOException {
+ Preconditions.checkArgument(n >= 0, "n must be non-negative");
+ while (n > 0) {
+ if (currentCount == 0) {
+ readNext();
+ }
+ int consume = Math.min(n, currentCount);
+ if (mode == MODE.PACKED) {
+ // The run is already unpacked into currentBuffer; skipping must advance the read
+ // position too, otherwise a later readInt() re-reads the values we just dropped.
+ currentBufferPosition += consume;
+ }
+ currentCount -= consume;
+ n -= consume;
+ }
+ }
+
private void readNext() throws IOException {
Preconditions.checkArgument(in.available() > 0, "Reading past RLE/BitPacking stream.");
final int header = BytesUtils.readUnsignedVarInt(in);
diff --git a/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridValuesReader.java b/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridValuesReader.java
index 0bd5a18d2..dd93807bd 100644
--- a/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridValuesReader.java
+++ b/parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridValuesReader.java
@@ -63,4 +63,13 @@ public class RunLengthBitPackingHybridValuesReader extends ValuesReader {
public void skip() {
readInteger();
}
+
+ @Override
+ public void skip(int n) {
+ try {
+ decoder.skipInts(n);
+ } catch (IOException e) {
+ throw new ParquetDecodingException(e);
+ }
+ }
}
diff --git a/parquet-column/src/test/java/org/apache/parquet/column/values/dictionary/TestDictionary.java b/parquet-column/src/test/java/org/apache/parquet/column/values/dictionary/TestDictionary.java
index 13033404c..44f90d901 100644
--- a/parquet-column/src/test/java/org/apache/parquet/column/values/dictionary/TestDictionary.java
+++ b/parquet-column/src/test/java/org/apache/parquet/column/values/dictionary/TestDictionary.java
@@ -25,6 +25,7 @@ import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.DOUBLE;
import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FLOAT;
import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.assertj.core.data.Offset.offset;
import java.io.IOException;
@@ -50,6 +51,7 @@ import org.apache.parquet.column.values.fallback.FallbackValuesWriter;
import org.apache.parquet.column.values.plain.BinaryPlainValuesReader;
import org.apache.parquet.column.values.plain.PlainValuesReader;
import org.apache.parquet.column.values.plain.PlainValuesWriter;
+import org.apache.parquet.io.ParquetDecodingException;
import org.apache.parquet.io.api.Binary;
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
import org.junit.jupiter.api.AfterEach;
@@ -178,6 +180,28 @@ public class TestDictionary {
}
}
+ /**
+ * An empty data page gets a decoder whose reads fail with a {@link ParquetDecodingException}, so
+ * the bulk-skip path has to fail the same way. Without an overridden skipInts it would reach the
+ * RLE decoder's readNext() and surface a raw IllegalArgumentException ("Reading past
+ * RLE/BitPacking stream") instead, which the skip(int) wrapper does not catch.
+ */
+ @Test
+ public void testSkipOnEmptyPage() throws Exception {
+ try (ValuesWriter cw = newPlainBinaryDictionaryValuesWriter(200, 10000)) {
+ writeRepeated(100, cw, "a");
+ cw.getBytes();
+ DictionaryValuesReader cr = initDicReader(cw, BINARY);
+ cr.initFromPage(0, BytesInput.empty().toInputStream());
+
+ assertThatThrownBy(() -> cr.skip(10)).isInstanceOf(ParquetDecodingException.class);
+ assertThatThrownBy(cr::skip).isInstanceOf(ParquetDecodingException.class);
+ // Skipping nothing reads nothing, so it must not throw — same as the pre-bulk-skip
+ // ValuesReader.skip(int) loop, which simply ran zero iterations.
+ cr.skip(0);
+ }
+ }
+
@Test
public void testBinaryDictionaryFallBack() throws IOException {
int slabSize = 100;
diff --git a/parquet-column/src/test/java/org/apache/parquet/column/values/rle/TestRunLengthBitPackingHybridDecoderSkip.java b/parquet-column/src/test/java/org/apache/parquet/column/values/rle/TestRunLengthBitPackingHybridDecoderSkip.java
new file mode 100644
index 000000000..43aa6158a
--- /dev/null
+++ b/parquet-column/src/test/java/org/apache/parquet/column/values/rle/TestRunLengthBitPackingHybridDecoderSkip.java
@@ -0,0 +1,224 @@
+/*
+ * 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.parquet.column.values.rle;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Random;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.DirectByteBufferAllocator;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Parity tests for {@link RunLengthBitPackingHybridDecoder#skipInts(int)}. The invariant
+ * checked in every case is: {@code skipInts(k)} followed by {@code readInt()} yields the same
+ * sequence as {@code k} calls to {@code readInt()} discarded followed by the same
+ * {@code readInt()}. Covers RLE-only streams, PACKED-only streams, mixed streams, run-boundary
+ * splits, skip-to-end, and RLE + PACKED interleaving with rewind via re-decode.
+ */
+public class TestRunLengthBitPackingHybridDecoderSkip {
+
+ private static final DirectByteBufferAllocator ALLOC = new DirectByteBufferAllocator();
+
+ @Test
+ public void skipAcrossRleRuns() throws Exception {
+ // Three big RLE runs. Any skip that lands mid-run or on a run boundary must still return
+ // the correct next value.
+ int bitWidth = 4;
+ int[] values = repeat(0, 500, repeat(7, 500, repeat(3, 500, new int[0])));
+ checkSkipParityAtAllOffsets(bitWidth, values);
+ }
+
+ @Test
+ public void skipAcrossPackedRuns() throws Exception {
+ // Values change every position -> encoder emits bit-packed groups exclusively.
+ int bitWidth = 4;
+ int[] values = new int[512];
+ for (int i = 0; i < values.length; i++) {
+ values[i] = i & 0xF;
+ }
+ checkSkipParityAtAllOffsets(bitWidth, values);
+ }
+
+ @Test
+ public void skipAcrossMixedRuns() throws Exception {
+ // Mix RLE and PACKED sections back-to-back.
+ int bitWidth = 5;
+ int[] values = repeat(9, 200, new int[0]); // RLE
+ // Small PACKED section (differing values)
+ int[] packed = new int[64];
+ for (int i = 0; i < packed.length; i++) {
+ packed[i] = i & 0x1F;
+ }
+ values = concat(values, packed);
+ values = repeat(4, 320, values); // RLE
+ // Larger PACKED section
+ int[] packed2 = new int[256];
+ for (int i = 0; i < packed2.length; i++) {
+ packed2[i] = (i * 3) & 0x1F;
+ }
+ values = concat(values, packed2);
+ values = repeat(31, 100, values); // RLE
+ checkSkipParityAtAllOffsets(bitWidth, values);
+ }
+
+ @Test
+ public void skipZero() throws Exception {
+ int[] values = new int[] {1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
+ RunLengthBitPackingHybridDecoder decoder = decoderFor(4, values);
+ decoder.skipInts(0);
+ for (int v : values) {
+ assertThat(decoder.readInt()).isEqualTo(v);
+ }
+ }
+
+ @Test
+ public void skipEntireStream() throws Exception {
+ int bitWidth = 4;
+ int[] values = new int[1024];
+ Random r = new Random(0xC0FFEE);
+ for (int i = 0; i < values.length; i++) {
+ values[i] = r.nextInt(1 << bitWidth);
+ }
+ RunLengthBitPackingHybridDecoder decoder = decoderFor(bitWidth, values);
+ decoder.skipInts(values.length);
+ // No values remaining; decoder must not be asked to readInt again (matches semantics of
+ // consuming all values normally).
+ }
+
+ @Test
+ public void skipThenReadThenSkipThenRead() throws Exception {
+ // Alternating skip/read pattern that spans multiple runs including partial runs.
+ int bitWidth = 6;
+ int[] values = new int[4096];
+ Random r = new Random(42);
+ for (int i = 0; i < values.length; i++) {
+ // Bias to produce both long RLE runs and PACKED regions.
+ values[i] = (i % 100 < 40) ? 17 : (r.nextInt(1 << bitWidth));
+ }
+ RunLengthBitPackingHybridDecoder decoder = decoderFor(bitWidth, values);
+ int pos = 0;
+ Random ops = new Random(0xABC);
+ while (pos < values.length) {
+ int remaining = values.length - pos;
+ int op = ops.nextInt(2);
+ if (op == 0) {
+ int skip = ops.nextInt(Math.min(200, remaining + 1));
+ decoder.skipInts(skip);
+ pos += skip;
+ } else {
+ int reads = ops.nextInt(Math.min(200, remaining) + 1);
+ for (int i = 0; i < reads; i++, pos++) {
+ assertThat(decoder.readInt()).isEqualTo(values[pos]);
+ }
+ }
+ }
+ }
+
+ @Test
+ public void skipWithBitWidthZero() throws Exception {
+ // bitWidth 0 -> a single RLE run of zeros. skip must not attempt any input reads.
+ int[] values = new int[500];
+ RunLengthBitPackingHybridDecoder decoder = decoderFor(0, values);
+ decoder.skipInts(300);
+ for (int i = 300; i < values.length; i++) {
+ assertThat(decoder.readInt()).isEqualTo(0);
+ }
+ }
+
+ @Test
+ public void skipPartialThenReadRest() throws Exception {
+ // Ensure the first N reads after a partial-run skip return the correct tail values.
+ int bitWidth = 3;
+ int[] values = new int[0];
+ // A PACKED run followed by an RLE run.
+ int[] packed = new int[64];
+ for (int i = 0; i < packed.length; i++) packed[i] = i & 0x7;
+ values = concat(values, packed);
+ values = repeat(5, 200, values);
+
+ // Skip inside the PACKED run.
+ RunLengthBitPackingHybridDecoder decoder = decoderFor(bitWidth, values);
+ decoder.skipInts(37);
+ for (int i = 37; i < values.length; i++) {
+ assertThat(decoder.readInt())
+ .as("value at index %d after skipInts(37)", i)
+ .isEqualTo(values[i]);
+ }
+ }
+
+ private static int[] repeat(int val, int count, int[] tail) {
+ int[] head = new int[count];
+ Arrays.fill(head, val);
+ return concat(head, tail);
+ }
+
+ private static int[] concat(int[] a, int[] b) {
+ int[] out = new int[a.length + b.length];
+ System.arraycopy(a, 0, out, 0, a.length);
+ System.arraycopy(b, 0, out, a.length, b.length);
+ return out;
+ }
+
+ private static RunLengthBitPackingHybridDecoder decoderFor(int bitWidth, int[] values) throws Exception {
+ RunLengthBitPackingHybridEncoder encoder = new RunLengthBitPackingHybridEncoder(bitWidth, 1024, 1 << 20, ALLOC);
+ for (int v : values) {
+ encoder.writeInt(v);
+ }
+ ByteBuffer buf = encoder.toBytes().toByteBuffer();
+ ByteBufferInputStream in = ByteBufferInputStream.wrap(buf);
+ return new RunLengthBitPackingHybridDecoder(bitWidth, in);
+ }
+
+ /**
+ * For a well-chosen set of skip offsets, assert that (a) skipInts(k) then readInt() equals
+ * values[k], and (b) after skipInts(k) subsequent reads walk values[k..end] correctly.
+ */
+ private static void checkSkipParityAtAllOffsets(int bitWidth, int[] values) throws Exception {
+ int[] offsets = new int[] {
+ 0,
+ 1,
+ 7,
+ 8,
+ 63,
+ 64,
+ 65,
+ values.length / 4,
+ values.length / 3,
+ values.length / 2,
+ (values.length * 2) / 3,
+ values.length - 1
+ };
+ for (int off : offsets) {
+ if (off < 0 || off >= values.length) {
+ continue;
+ }
+ RunLengthBitPackingHybridDecoder decoder = decoderFor(bitWidth, values);
+ decoder.skipInts(off);
+ // Read remainder and compare.
+ for (int i = off; i < values.length; i++) {
+ assertThat(decoder.readInt())
+ .as("bitWidth=%d, skipped=%d, index=%d", bitWidth, off, i)
+ .isEqualTo(values[i]);
+ }
+ }
+ }
+}