Skip to content

Commit 640d28e

Browse files
iemejiaiemejia
committed
GH-3530: Optimize RLE hybrid encoder/decoder scalar hot-path performance
Targeted optimizations to the Run-Length Encoding / Bit-Packing hybrid codec (RunLengthBitPackingHybridDecoder and Encoder), focused on the scalar read/write hot path used by the default Parquet column reader. Decoder optimizations: 1. InputStream to ByteBuffer: Replace InputStream/DataInputStream with direct ByteBuffer (LITTLE_ENDIAN) access, enabling buffer.get(), getShort(), getInt() intrinsics. Add ByteBuffer overloads for BytesUtils.readUnsignedVarInt() and readIntLittleEndianPaddedOnBitWidth(). Remove checked IOException from readInt(), simplifying all call sites (DictionaryValuesReader, ColumnReaderBase, ValuesReader wrapper). 2. Buffer reuse: Allocate the int[] packed values buffer once per decoder and grow lazily, instead of allocating fresh arrays on every PACKED run. Unpack directly from the main ByteBuffer (zero-copy common case); use a reusable zero-padded ByteBuffer for end-of-data edge cases. 3. unpack32Values fast path: Batch 4 groups (32 values) into a single unpack32Values call instead of looping unpack8Values, symmetric to the encoder change. Falls back to unpack8Values for residual groups. Encoder optimizations: 4. pack32Values fast path: Buffer four 8-value groups (32 values) then pack via pack32Values instead of four separate pack8Values calls, reducing per-group overhead and enabling the packer's optimized 32-value code path. flushBitPackedValues()/flushBitPackedValuesIfFull() deduplication to avoid repeating the flush logic. Adapted consumers to ByteBuffer-based decoder: - DictionaryValuesReader: Uses in.slice() for ByteBuffer; removed all try/catch IOException wrappers from read methods. - ColumnReaderBase.newRLEIterator: Uses bytes.toByteBuffer() directly; RLEIntIterator.nextInt() no longer wraps IOException. - RunLengthBitPackingHybridValuesReader: Uses stream.slice() for ByteBuffer constructor; readInteger() no longer wraps IOException. JMH benchmarks: - RleEncodingBenchmark: scalar boolean encoding via ValuesWriter across 6 data patterns (ALL_TRUE, ALL_FALSE, ALTERNATING, RANDOM, MOSTLY_TRUE_99, MOSTLY_FALSE_99). - RleDecodingBenchmark: scalar boolean decoding via ValuesReader, same 6 patterns. - RleDictionaryIndexDecodingBenchmark: scalar encode/decode via encoder/decoder directly, plus ValuesReader wrapper decode for dictionary index pages. Parameterized by bit width (1, 4, 8, 10, 16) and data pattern (SEQUENTIAL, RANDOM, LOW_CARDINALITY, CONSTANT) to exercise pure RLE, pure packed, and mixed paths across packing densities. All 610 parquet-column and parquet-common tests pass.
1 parent 390b90d commit 640d28e

14 files changed

Lines changed: 1110 additions & 88 deletions

File tree

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

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
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.io.IOException;
22+
import java.nio.ByteBuffer;
23+
import java.util.concurrent.TimeUnit;
24+
import org.apache.parquet.bytes.ByteBufferInputStream;
25+
import org.apache.parquet.bytes.HeapByteBufferAllocator;
26+
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridValuesReader;
27+
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridValuesWriter;
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+
import org.openjdk.jmh.infra.Blackhole;
42+
43+
/**
44+
* Decoding-level micro-benchmarks for the RLE/bit-packing hybrid encoding used
45+
* for {@code BOOLEAN} values in Parquet data pages V2.
46+
* Encoding benchmarks live in {@link RleEncodingBenchmark}.
47+
*
48+
* <p>The {@code dataPattern} parameter exercises RLE's best cases (ALL_TRUE,
49+
* ALL_FALSE), worst case (ALTERNATING), and realistic distributions (RANDOM,
50+
* MOSTLY_TRUE_99, MOSTLY_FALSE_99).
51+
*
52+
* <p>Each invocation decodes {@value #VALUE_COUNT} values; throughput is
53+
* reported per-value via {@link OperationsPerInvocation}.
54+
*/
55+
@BenchmarkMode(Mode.Throughput)
56+
@OutputTimeUnit(TimeUnit.SECONDS)
57+
@Fork(1)
58+
@Warmup(iterations = 3, time = 1)
59+
@Measurement(iterations = 5, time = 1)
60+
@State(Scope.Thread)
61+
public class RleDecodingBenchmark {
62+
63+
static final int VALUE_COUNT = 100_000;
64+
private static final int INIT_SLAB_SIZE = 64 * 1024;
65+
private static final int PAGE_SIZE = 4 * 1024 * 1024;
66+
67+
@Param({"ALL_TRUE", "ALL_FALSE", "ALTERNATING", "RANDOM", "MOSTLY_TRUE_99", "MOSTLY_FALSE_99"})
68+
public String dataPattern;
69+
70+
/** RLE-encoded bytes with 4-byte LE length prefix (ValuesReader format). */
71+
private byte[] encodedWithLengthPrefix;
72+
73+
@Setup(Level.Trial)
74+
public void setup() throws IOException {
75+
boolean[] data = RleEncodingBenchmark.generateData(dataPattern);
76+
77+
// Encode using the scalar ValuesWriter path
78+
RunLengthBitPackingHybridValuesWriter w =
79+
new RunLengthBitPackingHybridValuesWriter(1, INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator());
80+
for (boolean v : data) {
81+
w.writeBoolean(v);
82+
}
83+
encodedWithLengthPrefix = w.getBytes().toByteArray();
84+
w.close();
85+
}
86+
87+
// ---- Scalar decode via ValuesReader ----
88+
89+
@Benchmark
90+
@OperationsPerInvocation(VALUE_COUNT)
91+
public void decodeBoolean(Blackhole bh) throws IOException {
92+
RunLengthBitPackingHybridValuesReader r = new RunLengthBitPackingHybridValuesReader(1);
93+
r.initFromPage(VALUE_COUNT, ByteBufferInputStream.wrap(ByteBuffer.wrap(encodedWithLengthPrefix)));
94+
for (int i = 0; i < VALUE_COUNT; i++) {
95+
bh.consume(r.readBoolean());
96+
}
97+
}
98+
}

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

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,166 @@
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.io.IOException;
22+
import java.nio.ByteBuffer;
23+
import java.nio.ByteOrder;
24+
import java.util.concurrent.TimeUnit;
25+
import org.apache.parquet.bytes.ByteBufferInputStream;
26+
import org.apache.parquet.bytes.HeapByteBufferAllocator;
27+
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridDecoder;
28+
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridEncoder;
29+
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridValuesReader;
30+
import org.openjdk.jmh.annotations.Benchmark;
31+
import org.openjdk.jmh.annotations.BenchmarkMode;
32+
import org.openjdk.jmh.annotations.Fork;
33+
import org.openjdk.jmh.annotations.Level;
34+
import org.openjdk.jmh.annotations.Measurement;
35+
import org.openjdk.jmh.annotations.Mode;
36+
import org.openjdk.jmh.annotations.OperationsPerInvocation;
37+
import org.openjdk.jmh.annotations.OutputTimeUnit;
38+
import org.openjdk.jmh.annotations.Param;
39+
import org.openjdk.jmh.annotations.Scope;
40+
import org.openjdk.jmh.annotations.Setup;
41+
import org.openjdk.jmh.annotations.State;
42+
import org.openjdk.jmh.annotations.Warmup;
43+
import org.openjdk.jmh.infra.Blackhole;
44+
45+
/**
46+
* Encoding and decoding micro-benchmarks for synthetic dictionary-id pages using
47+
* {@link RunLengthBitPackingHybridEncoder} and {@link RunLengthBitPackingHybridDecoder}.
48+
* This isolates the RLE/bit-packing hybrid codec paths and is intentionally
49+
* separate from full INT32/INT64 value encode/decode path benchmarks.
50+
*
51+
* <p>The encode benchmark measures the RLE encoder's {@code pack32Values} fast path
52+
* and bit-packing throughput. The decode benchmark measures the corresponding
53+
* {@code unpack32Values} fast path and RLE run expansion.
54+
*
55+
* <p>The {@code bitWidth} parameter exercises different packing densities (1-bit to
56+
* 16-bit), and the {@code indexPattern} parameter exercises pure RLE (CONSTANT),
57+
* mixed (LOW_CARDINALITY), and pure bit-packing (SEQUENTIAL, RANDOM) paths.
58+
*
59+
* <p>Per-invocation overhead (encoder/decoder construction and {@link ByteBufferInputStream}
60+
* wrapping) is amortized over {@value #VALUE_COUNT} reads via
61+
* {@link OperationsPerInvocation}.
62+
*/
63+
@BenchmarkMode(Mode.Throughput)
64+
@OutputTimeUnit(TimeUnit.SECONDS)
65+
@Fork(1)
66+
@Warmup(iterations = 3, time = 1)
67+
@Measurement(iterations = 5, time = 1)
68+
@State(Scope.Thread)
69+
public class RleDictionaryIndexDecodingBenchmark {
70+
71+
static final int VALUE_COUNT = 100_000;
72+
private static final int INIT_SLAB_SIZE = 64 * 1024;
73+
private static final int PAGE_SIZE = 1024 * 1024;
74+
75+
@Param({"1", "4", "8", "10", "16"})
76+
public int bitWidth;
77+
78+
@Param({"SEQUENTIAL", "RANDOM", "LOW_CARDINALITY", "CONSTANT"})
79+
public String indexPattern;
80+
81+
/** Raw RLE-encoded bytes (no length prefix). */
82+
private byte[] encoded;
83+
84+
private int[] ids;
85+
86+
/** RLE-encoded bytes with 4-byte LE length prefix (ValuesReader format). */
87+
private byte[] encodedWithLengthPrefix;
88+
89+
@Setup(Level.Trial)
90+
public void setup() throws IOException {
91+
int maxId = 1 << bitWidth;
92+
ids = generateDictionaryIds(maxId);
93+
try (RunLengthBitPackingHybridEncoder encoder = new RunLengthBitPackingHybridEncoder(
94+
bitWidth, INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator())) {
95+
for (int id : ids) {
96+
encoder.writeInt(id);
97+
}
98+
encoded = encoder.toBytes().toByteArray();
99+
}
100+
101+
// Prepend 4-byte LE length for ValuesReader.initFromPage() format
102+
encodedWithLengthPrefix = new byte[4 + encoded.length];
103+
ByteBuffer.wrap(encodedWithLengthPrefix).order(ByteOrder.LITTLE_ENDIAN).putInt(encoded.length);
104+
System.arraycopy(encoded, 0, encodedWithLengthPrefix, 4, encoded.length);
105+
}
106+
107+
private int[] generateDictionaryIds(int maxId) {
108+
switch (indexPattern) {
109+
case "SEQUENTIAL":
110+
int[] sequential = new int[VALUE_COUNT];
111+
for (int i = 0; i < VALUE_COUNT; i++) {
112+
sequential[i] = i % maxId;
113+
}
114+
return sequential;
115+
case "RANDOM":
116+
return TestDataFactory.generateLowCardinalityInts(VALUE_COUNT, maxId, TestDataFactory.DEFAULT_SEED);
117+
case "LOW_CARDINALITY":
118+
int distinct = Math.min(TestDataFactory.LOW_CARDINALITY_DISTINCT, maxId);
119+
return TestDataFactory.generateLowCardinalityInts(VALUE_COUNT, distinct, TestDataFactory.DEFAULT_SEED);
120+
case "CONSTANT":
121+
int[] constant = new int[VALUE_COUNT];
122+
java.util.Arrays.fill(constant, 0);
123+
return constant;
124+
default:
125+
throw new IllegalArgumentException("Unknown index pattern: " + indexPattern);
126+
}
127+
}
128+
129+
// ---- Scalar encode via encoder ----
130+
131+
@Benchmark
132+
@OperationsPerInvocation(VALUE_COUNT)
133+
public byte[] encodeDictionaryIds() throws IOException {
134+
try (RunLengthBitPackingHybridEncoder encoder = new RunLengthBitPackingHybridEncoder(
135+
bitWidth, INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator())) {
136+
for (int id : ids) {
137+
encoder.writeInt(id);
138+
}
139+
return encoder.toBytes().toByteArray();
140+
}
141+
}
142+
143+
// ---- Scalar decode via decoder ----
144+
145+
@Benchmark
146+
@OperationsPerInvocation(VALUE_COUNT)
147+
public void decodeDictionaryIds(Blackhole bh) {
148+
RunLengthBitPackingHybridDecoder decoder =
149+
new RunLengthBitPackingHybridDecoder(bitWidth, ByteBuffer.wrap(encoded));
150+
for (int i = 0; i < VALUE_COUNT; i++) {
151+
bh.consume(decoder.readInt());
152+
}
153+
}
154+
155+
// ---- Scalar decode via ValuesReader wrapper ----
156+
157+
@Benchmark
158+
@OperationsPerInvocation(VALUE_COUNT)
159+
public void decodeValuesReader(Blackhole bh) throws IOException {
160+
RunLengthBitPackingHybridValuesReader reader = new RunLengthBitPackingHybridValuesReader(bitWidth);
161+
reader.initFromPage(VALUE_COUNT, ByteBufferInputStream.wrap(ByteBuffer.wrap(encodedWithLengthPrefix)));
162+
for (int i = 0; i < VALUE_COUNT; i++) {
163+
bh.consume(reader.readInteger());
164+
}
165+
}
166+
}

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

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
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.io.IOException;
22+
import java.util.Random;
23+
import java.util.concurrent.TimeUnit;
24+
import org.apache.parquet.bytes.HeapByteBufferAllocator;
25+
import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridValuesWriter;
26+
import org.openjdk.jmh.annotations.Benchmark;
27+
import org.openjdk.jmh.annotations.BenchmarkMode;
28+
import org.openjdk.jmh.annotations.Fork;
29+
import org.openjdk.jmh.annotations.Level;
30+
import org.openjdk.jmh.annotations.Measurement;
31+
import org.openjdk.jmh.annotations.Mode;
32+
import org.openjdk.jmh.annotations.OperationsPerInvocation;
33+
import org.openjdk.jmh.annotations.OutputTimeUnit;
34+
import org.openjdk.jmh.annotations.Param;
35+
import org.openjdk.jmh.annotations.Scope;
36+
import org.openjdk.jmh.annotations.Setup;
37+
import org.openjdk.jmh.annotations.State;
38+
import org.openjdk.jmh.annotations.Warmup;
39+
40+
/**
41+
* Encoding-level micro-benchmarks for the RLE/bit-packing hybrid encoding used
42+
* for {@code BOOLEAN} values in Parquet data pages V2.
43+
* Decoding benchmarks live in {@link RleDecodingBenchmark}.
44+
*
45+
* <p>The {@code dataPattern} parameter exercises RLE's best cases (ALL_TRUE,
46+
* ALL_FALSE), worst case (ALTERNATING), and realistic distributions (RANDOM,
47+
* MOSTLY_TRUE_99, MOSTLY_FALSE_99).
48+
*
49+
* <p>Each invocation encodes {@value #VALUE_COUNT} values; throughput is
50+
* reported per-value via {@link OperationsPerInvocation}.
51+
*/
52+
@BenchmarkMode(Mode.Throughput)
53+
@OutputTimeUnit(TimeUnit.SECONDS)
54+
@Fork(1)
55+
@Warmup(iterations = 3, time = 1)
56+
@Measurement(iterations = 5, time = 1)
57+
@State(Scope.Thread)
58+
public class RleEncodingBenchmark {
59+
60+
static final int VALUE_COUNT = 100_000;
61+
private static final int INIT_SLAB_SIZE = 64 * 1024;
62+
private static final int PAGE_SIZE = 4 * 1024 * 1024;
63+
64+
@Param({"ALL_TRUE", "ALL_FALSE", "ALTERNATING", "RANDOM", "MOSTLY_TRUE_99", "MOSTLY_FALSE_99"})
65+
public String dataPattern;
66+
67+
private boolean[] data;
68+
69+
@Setup(Level.Trial)
70+
public void setup() {
71+
data = generateData(dataPattern);
72+
}
73+
74+
static boolean[] generateData(String pattern) {
75+
boolean[] d = new boolean[VALUE_COUNT];
76+
Random rng = new Random(42);
77+
switch (pattern) {
78+
case "ALL_TRUE":
79+
for (int i = 0; i < VALUE_COUNT; i++) d[i] = true;
80+
break;
81+
case "ALL_FALSE":
82+
// already false
83+
break;
84+
case "ALTERNATING":
85+
for (int i = 0; i < VALUE_COUNT; i++) d[i] = (i & 1) == 0;
86+
break;
87+
case "RANDOM":
88+
for (int i = 0; i < VALUE_COUNT; i++) d[i] = rng.nextBoolean();
89+
break;
90+
case "MOSTLY_TRUE_99":
91+
for (int i = 0; i < VALUE_COUNT; i++) d[i] = rng.nextInt(100) != 0;
92+
break;
93+
case "MOSTLY_FALSE_99":
94+
for (int i = 0; i < VALUE_COUNT; i++) d[i] = rng.nextInt(100) == 0;
95+
break;
96+
default:
97+
throw new IllegalArgumentException("Unknown pattern: " + pattern);
98+
}
99+
return d;
100+
}
101+
102+
// ---- Scalar encode via ValuesWriter ----
103+
104+
@Benchmark
105+
@OperationsPerInvocation(VALUE_COUNT)
106+
public byte[] encodeBoolean() throws IOException {
107+
RunLengthBitPackingHybridValuesWriter w =
108+
new RunLengthBitPackingHybridValuesWriter(1, INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator());
109+
for (boolean v : data) {
110+
w.writeBoolean(v);
111+
}
112+
byte[] bytes = w.getBytes().toByteArray();
113+
w.close();
114+
return bytes;
115+
}
116+
}

0 commit comments

Comments
 (0)