From f3246b5f5119e5315ae4ad152b7b0f73f41f4454 Mon Sep 17 00:00:00 2001 From: Laszlo Bodor Date: Fri, 4 Sep 2026 18:06:50 +0200 Subject: [PATCH] GH-3772: Bulk skip in RunLengthBitPackingHybridDecoder / DictionaryValuesReader Add skipInts(int n) to RunLengthBitPackingHybridDecoder. It re-uses the existing readNext() to load each run and then advances currentCount by min(n, currentCount) instead of returning values one-by-one via readInt(). Runs are decoded the same way as before -- the win is dropping the per-value mode switch, array-index arithmetic and method-call overhead that readInt() pays for each value the caller is going to throw away. Propagate to the two ValuesReader wrappers (DictionaryValuesReader, RunLengthBitPackingHybridValuesReader) so callers going through the public ValuesReader.skip(int) contract get the fast path. Motivation: dictionary-encoded columns are ubiquitous in production Parquet, and the default ValuesReader.skip(int) is a naive loop over skip() -- which for dict columns is a RunLengthBitPackingHybridDecoder readInt(). Filter-driven read paths (column-index row ranges, Hive ProbeDecode, arbitrary row-skip) pay that cost per skipped row even when the values are being thrown away. Bench (parquet-benchmarks / RleSkipBenchmark, thrpt, 1 fork, 3x1s warmup, 5x1s measure, 100k values/op, ops/s of individual values): pattern=rle bitWidth=8: 1.38 B/s -> 103.2 B/s (~75x) pattern=packed bitWidth=8: 0.89 B/s -> 4.12 B/s (~4.6x) pattern=mixed bitWidth=8: 0.96 B/s -> 6.85 B/s (~7.1x) RLE runs dominate because a whole run is consumed with a single currentCount decrement; bit-packed runs still get fully unpacked, so the gain there is just the readInt() overhead avoided per value. Tests: TestRunLengthBitPackingHybridDecoderSkip covers RLE-only, PACKED-only, mixed, zero-skip, full-skip, partial-run skip, bitWidth=0, and randomized skip/read alternation across 4K values. All 700 parquet-column tests still pass. --- .../parquet/benchmarks/RleSkipBenchmark.java | 136 +++++++++++ .../dictionary/DictionaryValuesReader.java | 11 + .../rle/RunLengthBitPackingHybridDecoder.java | 22 ++ ...RunLengthBitPackingHybridValuesReader.java | 9 + ...tRunLengthBitPackingHybridDecoderSkip.java | 224 ++++++++++++++++++ 5 files changed, 402 insertions(+) create mode 100644 parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/RleSkipBenchmark.java create mode 100644 parquet-column/src/test/java/org/apache/parquet/column/values/rle/TestRunLengthBitPackingHybridDecoderSkip.java 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 0000000000..b51111e6ef --- /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. + * + *

Parameters: + *

+ * + * 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 53fafc55dc..f95e90de6c 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 @@ -125,4 +125,15 @@ public void skip() { 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 e55b276b29..c0897faa3b 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 @@ -77,6 +77,28 @@ public int readInt() throws IOException { 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. + * + *

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); + 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 0bd5a18d2b..dd93807bd9 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 boolean readBoolean() { 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/rle/TestRunLengthBitPackingHybridDecoderSkip.java b/parquet-column/src/test/java/org/apache/parquet/column/values/rle/TestRunLengthBitPackingHybridDecoderSkip.java new file mode 100644 index 0000000000..43aa6158af --- /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]); + } + } + } +}