Skip to content

Commit 5179bd9

Browse files
abstractdogabstractdog
committed
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.
1 parent a1d8829 commit 5179bd9

5 files changed

Lines changed: 402 additions & 0 deletions

File tree

‎parquet-benchmarks/src/main/java/org/apache/parquet/benchmarks/RleSkipBenchmark.java‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.apache.parquet.benchmarks;
20+
21+
import java.nio.ByteBuffer;
22+
import java.util.Random;
23+
import java.util.concurrent.TimeUnit;
24+
import org.apache.parquet.bytes.ByteBufferInputStream;
25+
import org.apache.parquet.bytes.DirectByteBufferAllocator;
26+
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridDecoder;
27+
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridEncoder;
28+
import org.openjdk.jmh.annotations.Benchmark;
29+
import org.openjdk.jmh.annotations.BenchmarkMode;
30+
import org.openjdk.jmh.annotations.Fork;
31+
import org.openjdk.jmh.annotations.Level;
32+
import org.openjdk.jmh.annotations.Measurement;
33+
import org.openjdk.jmh.annotations.Mode;
34+
import org.openjdk.jmh.annotations.OperationsPerInvocation;
35+
import org.openjdk.jmh.annotations.OutputTimeUnit;
36+
import org.openjdk.jmh.annotations.Param;
37+
import org.openjdk.jmh.annotations.Scope;
38+
import org.openjdk.jmh.annotations.Setup;
39+
import org.openjdk.jmh.annotations.State;
40+
import org.openjdk.jmh.annotations.Warmup;
41+
42+
/**
43+
* Measures {@link RunLengthBitPackingHybridDecoder#skipInts(int)} vs. the equivalent
44+
* discard-via-readInt loop. Both paths decode each run the same way; the win comes from
45+
* dropping the per-value mode-switch, array-index arithmetic and method-call overhead of
46+
* {@code readInt()} in favour of a single {@code currentCount -= consume} per run.
47+
*
48+
* <p>Parameters:
49+
* <ul>
50+
* <li>{@link #bitWidth} -- key width, chosen at 3/8/16 to bracket typical dictionary-index
51+
* widths.</li>
52+
* <li>{@link #pattern} -- {@code rle} produces long runs of a single value (best case for both
53+
* paths); {@code packed} produces mostly distinct values so the encoder emits bit-packed
54+
* groups (worst case for the old skip); {@code mixed} interleaves the two.</li>
55+
* </ul>
56+
*
57+
* Each invocation re-wraps a pre-encoded byte buffer and calls {@code skipInts(VALUE_COUNT)} in
58+
* the {@code skip} benchmark, versus a {@code VALUE_COUNT}-long {@code readInt()} discard loop
59+
* in the {@code readSkip} benchmark.
60+
*/
61+
@BenchmarkMode(Mode.Throughput)
62+
@OutputTimeUnit(TimeUnit.SECONDS)
63+
@Fork(1)
64+
@Warmup(iterations = 3, time = 1)
65+
@Measurement(iterations = 5, time = 1)
66+
@State(Scope.Thread)
67+
public class RleSkipBenchmark {
68+
69+
static final int VALUE_COUNT = 100_000;
70+
private static final int INIT_SLAB = 64 * 1024;
71+
private static final int PAGE = 4 * 1024 * 1024;
72+
73+
@Param({"3", "8", "16"})
74+
public int bitWidth;
75+
76+
@Param({"rle", "packed", "mixed"})
77+
public String pattern;
78+
79+
private byte[] encoded;
80+
81+
@Setup(Level.Trial)
82+
public void setup() throws Exception {
83+
int mask = bitWidth == 32 ? -1 : ((1 << bitWidth) - 1);
84+
RunLengthBitPackingHybridEncoder enc =
85+
new RunLengthBitPackingHybridEncoder(bitWidth, INIT_SLAB, PAGE, new DirectByteBufferAllocator());
86+
Random r = new Random(42);
87+
switch (pattern) {
88+
case "rle":
89+
// Long stretches of the same value -> the encoder emits RLE runs.
90+
for (int i = 0; i < VALUE_COUNT; i++) {
91+
enc.writeInt(((i / 500) & 0x1F) & mask);
92+
}
93+
break;
94+
case "packed":
95+
// Random values -> the encoder emits bit-packed groups.
96+
for (int i = 0; i < VALUE_COUNT; i++) {
97+
enc.writeInt(r.nextInt() & mask);
98+
}
99+
break;
100+
case "mixed":
101+
// Alternating 250-value RLE blocks and 250-value random blocks.
102+
for (int block = 0; block * 250 < VALUE_COUNT; block++) {
103+
int val = r.nextInt() & mask;
104+
boolean rle = (block & 1) == 0;
105+
int limit = Math.min(250, VALUE_COUNT - block * 250);
106+
for (int j = 0; j < limit; j++) {
107+
enc.writeInt(rle ? val : (r.nextInt() & mask));
108+
}
109+
}
110+
break;
111+
default:
112+
throw new IllegalArgumentException("unknown pattern: " + pattern);
113+
}
114+
encoded = enc.toBytes().toByteArray();
115+
}
116+
117+
@Benchmark
118+
@OperationsPerInvocation(VALUE_COUNT)
119+
public void skip() throws Exception {
120+
ByteBufferInputStream in = ByteBufferInputStream.wrap(ByteBuffer.wrap(encoded));
121+
RunLengthBitPackingHybridDecoder dec = new RunLengthBitPackingHybridDecoder(bitWidth, in);
122+
dec.skipInts(VALUE_COUNT);
123+
}
124+
125+
@Benchmark
126+
@OperationsPerInvocation(VALUE_COUNT)
127+
public int readSkip() throws Exception {
128+
ByteBufferInputStream in = ByteBufferInputStream.wrap(ByteBuffer.wrap(encoded));
129+
RunLengthBitPackingHybridDecoder dec = new RunLengthBitPackingHybridDecoder(bitWidth, in);
130+
int sink = 0;
131+
for (int i = 0; i < VALUE_COUNT; i++) {
132+
sink ^= dec.readInt();
133+
}
134+
return sink;
135+
}
136+
}

‎parquet-column/src/main/java/org/apache/parquet/column/values/dictionary/DictionaryValuesReader.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -125,4 +125,15 @@ public void skip() {
125125
throw new ParquetDecodingException(e);
126126
}
127127
}
128+
129+
@Override
130+
public void skip(int n) {
131+
// Bulk-skip dictionary keys without decoding them or looking them up in the dictionary.
132+
// See RunLengthBitPackingHybridDecoder#skipInts for the fast-path details.
133+
try {
134+
decoder.skipInts(n);
135+
} catch (IOException e) {
136+
throw new ParquetDecodingException(e);
137+
}
138+
}
128139
}

‎parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridDecoder.java‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,28 @@ public int readInt() throws IOException {
7777
return result;
7878
}
7979

80+
/**
81+
* Skip {@code n} values without returning them. Runs are decoded normally, but the values are
82+
* dropped instead of being handed back one-by-one via {@link #readInt()}, which saves the
83+
* per-value branch on {@link #mode} and the array-index arithmetic that {@code readInt()} does.
84+
*
85+
* <p>Intended for callers that filter rows (column-index row ranges, hash-join probe filtering,
86+
* etc.) and need to advance past many values on a dictionary-key or level column cheaply.
87+
*
88+
* @param n number of values to skip; must be non-negative
89+
*/
90+
public void skipInts(int n) throws IOException {
91+
Preconditions.checkArgument(n >= 0, "n must be non-negative");
92+
while (n > 0) {
93+
if (currentCount == 0) {
94+
readNext();
95+
}
96+
int consume = Math.min(n, currentCount);
97+
currentCount -= consume;
98+
n -= consume;
99+
}
100+
}
101+
80102
private void readNext() throws IOException {
81103
Preconditions.checkArgument(in.available() > 0, "Reading past RLE/BitPacking stream.");
82104
final int header = BytesUtils.readUnsignedVarInt(in);

‎parquet-column/src/main/java/org/apache/parquet/column/values/rle/RunLengthBitPackingHybridValuesReader.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,4 +63,13 @@ public boolean readBoolean() {
6363
public void skip() {
6464
readInteger();
6565
}
66+
67+
@Override
68+
public void skip(int n) {
69+
try {
70+
decoder.skipInts(n);
71+
} catch (IOException e) {
72+
throw new ParquetDecodingException(e);
73+
}
74+
}
6675
}

0 commit comments

Comments
 (0)