Skip to content

Commit eb1ee2d

Browse files
vinooganeshjulienledem
vinooganesh
and
julienledem
authored
GH-3862: Parquet Java ALP Implementation (#3397)
Implements ALP encoding for FLOAT and DOUBLE types, which converts floating-point values to integers using decimal scaling, then applies Frame of Reference (FOR) encoding and bit-packing for compression. --------- Co-authored-by: Julien Le Dem <[email protected]>
1 parent 2563b37 commit eb1ee2d

27 files changed

Lines changed: 7366 additions & 13 deletions

File tree

‎parquet-benchmarks/pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -95,6 +95,7 @@
9595
<groupId>org.apache.maven.plugins</groupId>
9696
<artifactId>maven-compiler-plugin</artifactId>
9797
<configuration>
98+
<proc>full</proc>
9899
<annotationProcessorPaths>
99100
<path>
100101
<groupId>org.openjdk.jmh</groupId>

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

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
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.ValuesReader;
27+
import org.apache.parquet.column.values.ValuesWriter;
28+
import org.apache.parquet.column.values.alp.AlpValuesReaderForDouble;
29+
import org.apache.parquet.column.values.alp.AlpValuesReaderForFloat;
30+
import org.apache.parquet.column.values.alp.AlpValuesWriter;
31+
import org.apache.parquet.column.values.plain.PlainValuesReader;
32+
import org.apache.parquet.column.values.plain.PlainValuesWriter;
33+
import org.openjdk.jmh.annotations.Benchmark;
34+
import org.openjdk.jmh.annotations.BenchmarkMode;
35+
import org.openjdk.jmh.annotations.Fork;
36+
import org.openjdk.jmh.annotations.Level;
37+
import org.openjdk.jmh.annotations.Measurement;
38+
import org.openjdk.jmh.annotations.Mode;
39+
import org.openjdk.jmh.annotations.OperationsPerInvocation;
40+
import org.openjdk.jmh.annotations.OutputTimeUnit;
41+
import org.openjdk.jmh.annotations.Param;
42+
import org.openjdk.jmh.annotations.Scope;
43+
import org.openjdk.jmh.annotations.Setup;
44+
import org.openjdk.jmh.annotations.State;
45+
import org.openjdk.jmh.annotations.Warmup;
46+
import org.openjdk.jmh.infra.Blackhole;
47+
48+
/**
49+
* Decoding-level micro-benchmarks for ALP, with PLAIN on the same data as a baseline. The data sets
50+
* are those of {@link AlpEncodingBenchmark}.
51+
*
52+
* <p>Each invocation decodes {@value #VALUE_COUNT} values; throughput is reported per-value via
53+
* {@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 AlpDecodingBenchmark {
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({"decimal", "random"})
68+
public String data;
69+
70+
private byte[] alpDoublePage;
71+
private byte[] plainDoublePage;
72+
private byte[] alpFloatPage;
73+
private byte[] plainFloatPage;
74+
75+
@Setup(Level.Trial)
76+
public void setup() throws IOException {
77+
double[] doubles = AlpEncodingBenchmark.doubles(data, VALUE_COUNT);
78+
float[] floats = AlpEncodingBenchmark.floats(data, VALUE_COUNT);
79+
alpDoublePage = page(
80+
new AlpValuesWriter.DoubleAlpValuesWriter(INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator()),
81+
doubles);
82+
plainDoublePage =
83+
page(new PlainValuesWriter(INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator()), doubles);
84+
alpFloatPage = page(
85+
new AlpValuesWriter.FloatAlpValuesWriter(INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator()),
86+
floats);
87+
plainFloatPage = page(new PlainValuesWriter(INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator()), floats);
88+
}
89+
90+
private static byte[] page(ValuesWriter w, double[] values) throws IOException {
91+
for (double v : values) {
92+
w.writeDouble(v);
93+
}
94+
byte[] bytes = w.getBytes().toByteArray();
95+
w.close();
96+
return bytes;
97+
}
98+
99+
private static byte[] page(ValuesWriter w, float[] values) throws IOException {
100+
for (float v : values) {
101+
w.writeFloat(v);
102+
}
103+
byte[] bytes = w.getBytes().toByteArray();
104+
w.close();
105+
return bytes;
106+
}
107+
108+
private static void init(ValuesReader r, byte[] page) throws IOException {
109+
r.initFromPage(VALUE_COUNT, ByteBufferInputStream.wrap(ByteBuffer.wrap(page)));
110+
}
111+
112+
@Benchmark
113+
@OperationsPerInvocation(VALUE_COUNT)
114+
public void decodeDouble(Blackhole bh) throws IOException {
115+
AlpValuesReaderForDouble r = new AlpValuesReaderForDouble();
116+
init(r, alpDoublePage);
117+
for (int i = 0; i < VALUE_COUNT; i++) {
118+
bh.consume(r.readDouble());
119+
}
120+
}
121+
122+
@Benchmark
123+
@OperationsPerInvocation(VALUE_COUNT)
124+
public void decodeDoublePlain(Blackhole bh) throws IOException {
125+
PlainValuesReader.DoublePlainValuesReader r = new PlainValuesReader.DoublePlainValuesReader();
126+
init(r, plainDoublePage);
127+
for (int i = 0; i < VALUE_COUNT; i++) {
128+
bh.consume(r.readDouble());
129+
}
130+
}
131+
132+
@Benchmark
133+
@OperationsPerInvocation(VALUE_COUNT)
134+
public void decodeFloat(Blackhole bh) throws IOException {
135+
AlpValuesReaderForFloat r = new AlpValuesReaderForFloat();
136+
init(r, alpFloatPage);
137+
for (int i = 0; i < VALUE_COUNT; i++) {
138+
bh.consume(r.readFloat());
139+
}
140+
}
141+
142+
@Benchmark
143+
@OperationsPerInvocation(VALUE_COUNT)
144+
public void decodeFloatPlain(Blackhole bh) throws IOException {
145+
PlainValuesReader.FloatPlainValuesReader r = new PlainValuesReader.FloatPlainValuesReader();
146+
init(r, plainFloatPage);
147+
for (int i = 0; i < VALUE_COUNT; i++) {
148+
bh.consume(r.readFloat());
149+
}
150+
}
151+
}

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

Lines changed: 138 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,138 @@
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.ValuesWriter;
26+
import org.apache.parquet.column.values.alp.AlpValuesWriter;
27+
import org.apache.parquet.column.values.plain.PlainValuesWriter;
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+
* Encoding-level micro-benchmarks for ALP, with PLAIN on the same data as a baseline.
44+
*
45+
* <p>{@code decimal} data is what ALP is designed for; {@code random} data is almost all
46+
* exceptions, its worst case.
47+
*
48+
* <p>Each invocation encodes {@value #VALUE_COUNT} values; throughput is reported per-value via
49+
* {@link OperationsPerInvocation}.
50+
*/
51+
@BenchmarkMode(Mode.Throughput)
52+
@OutputTimeUnit(TimeUnit.SECONDS)
53+
@Fork(1)
54+
@Warmup(iterations = 3, time = 1)
55+
@Measurement(iterations = 5, time = 1)
56+
@State(Scope.Thread)
57+
public class AlpEncodingBenchmark {
58+
59+
// Enough values that each run includes both the full parameter search, used until enough
60+
// vectors are sampled, and the preset search that follows, as a column chunk would.
61+
static final int VALUE_COUNT = 1_000_000;
62+
private static final int INIT_SLAB_SIZE = 64 * 1024;
63+
private static final int PAGE_SIZE = 16 * 1024 * 1024;
64+
65+
@Param({"decimal", "random"})
66+
public String data;
67+
68+
private double[] doubleData;
69+
private float[] floatData;
70+
71+
@Setup(Level.Trial)
72+
public void setup() {
73+
doubleData = doubles(data, VALUE_COUNT);
74+
floatData = floats(data, VALUE_COUNT);
75+
}
76+
77+
static double[] doubles(String data, int count) {
78+
Random random = new Random(42);
79+
double[] values = new double[count];
80+
for (int i = 0; i < count; i++) {
81+
values[i] = "decimal".equals(data) ? Math.round(random.nextDouble() * 1e6) / 100.0 : random.nextDouble();
82+
}
83+
return values;
84+
}
85+
86+
static float[] floats(String data, int count) {
87+
Random random = new Random(42);
88+
float[] values = new float[count];
89+
for (int i = 0; i < count; i++) {
90+
values[i] = "decimal".equals(data) ? Math.round(random.nextFloat() * 1e6f) / 100.0f : random.nextFloat();
91+
}
92+
return values;
93+
}
94+
95+
@Benchmark
96+
@OperationsPerInvocation(VALUE_COUNT)
97+
public byte[] encodeDouble() throws IOException {
98+
return encodeDoubles(
99+
new AlpValuesWriter.DoubleAlpValuesWriter(INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator()));
100+
}
101+
102+
@Benchmark
103+
@OperationsPerInvocation(VALUE_COUNT)
104+
public byte[] encodeDoublePlain() throws IOException {
105+
return encodeDoubles(new PlainValuesWriter(INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator()));
106+
}
107+
108+
@Benchmark
109+
@OperationsPerInvocation(VALUE_COUNT)
110+
public byte[] encodeFloat() throws IOException {
111+
return encodeFloats(
112+
new AlpValuesWriter.FloatAlpValuesWriter(INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator()));
113+
}
114+
115+
@Benchmark
116+
@OperationsPerInvocation(VALUE_COUNT)
117+
public byte[] encodeFloatPlain() throws IOException {
118+
return encodeFloats(new PlainValuesWriter(INIT_SLAB_SIZE, PAGE_SIZE, new HeapByteBufferAllocator()));
119+
}
120+
121+
private byte[] encodeDoubles(ValuesWriter w) throws IOException {
122+
for (double v : doubleData) {
123+
w.writeDouble(v);
124+
}
125+
byte[] bytes = w.getBytes().toByteArray();
126+
w.close();
127+
return bytes;
128+
}
129+
130+
private byte[] encodeFloats(ValuesWriter w) throws IOException {
131+
for (float v : floatData) {
132+
w.writeFloat(v);
133+
}
134+
byte[] bytes = w.getBytes().toByteArray();
135+
w.close();
136+
return bytes;
137+
}
138+
}

‎parquet-column/src/main/java/org/apache/parquet/column/ColumnProperty.java‎

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,9 +19,11 @@
1919

2020
package org.apache.parquet.column;
2121

22+
import java.util.Collections;
2223
import java.util.HashMap;
2324
import java.util.Map;
2425
import java.util.Objects;
26+
import java.util.Set;
2527
import org.apache.parquet.hadoop.metadata.ColumnPath;
2628

2729
/**
@@ -62,13 +64,19 @@ private MultipleColumnProperty(T defaultValue, Map<ColumnPath, T> values) {
6264

6365
@Override
6466
public T getValue(ColumnPath columnPath) {
65-
T value = values.get(columnPath);
66-
if (value != null) {
67-
return value;
67+
// containsKey rather than a null check, so a column can be given an explicit null value
68+
// that overrides the default rather than falling back to it.
69+
if (values.containsKey(columnPath)) {
70+
return values.get(columnPath);
6871
}
6972
return getDefaultValue();
7073
}
7174

75+
@Override
76+
Set<ColumnPath> getColumnPaths() {
77+
return values.keySet();
78+
}
79+
7280
@Override
7381
public String toString() {
7482
return Objects.toString(getDefaultValue()) + ' ' + values.toString();
@@ -125,6 +133,14 @@ public static <T> Builder<T> builder(ColumnProperty<T> toCopy) {
125133

126134
public abstract T getValue(ColumnPath columnPath);
127135

136+
/**
137+
* @return the column paths that have an explicit per-column value set (empty when the property
138+
* only carries a default value)
139+
*/
140+
Set<ColumnPath> getColumnPaths() {
141+
return Collections.emptySet();
142+
}
143+
128144
public T getValue(String columnPath) {
129145
return getValue(ColumnPath.fromDotString(columnPath));
130146
}

‎parquet-column/src/main/java/org/apache/parquet/column/Encoding.java‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@
2929
import org.apache.parquet.bytes.BytesUtils;
3030
import org.apache.parquet.column.page.DictionaryPage;
3131
import org.apache.parquet.column.values.ValuesReader;
32+
import org.apache.parquet.column.values.alp.AlpValuesReaderForDouble;
33+
import org.apache.parquet.column.values.alp.AlpValuesReaderForFloat;
3234
import org.apache.parquet.column.values.bitpacking.ByteBitPackingValuesReader;
3335
import org.apache.parquet.column.values.bytestreamsplit.ByteStreamSplitValuesReaderForDouble;
3436
import org.apache.parquet.column.values.bytestreamsplit.ByteStreamSplitValuesReaderForFLBA;
@@ -147,6 +149,27 @@ public ValuesReader getValuesReader(ColumnDescriptor descriptor, ValuesType valu
147149
}
148150
},
149151

152+
/**
153+
* ALP (Adaptive Lossless floating-Point) encoding for FLOAT and DOUBLE types.
154+
*
155+
* @see <a href="https://parquet.apache.org/docs/file-format/data-pages/alpencoding/">ALP specification</a>
156+
* @see <a href="https://parquet.apache.org/blog/2026/09/22/alp-adaptive-lossless-floating-point-encoding-in-apache-parquet/">ALP blog post</a>
157+
*/
158+
ALP {
159+
@Override
160+
public ValuesReader getValuesReader(ColumnDescriptor descriptor, ValuesType valuesType) {
161+
switch (descriptor.getType()) {
162+
case FLOAT:
163+
return new AlpValuesReaderForFloat();
164+
case DOUBLE:
165+
return new AlpValuesReaderForDouble();
166+
default:
167+
throw new ParquetDecodingException(
168+
"ALP encoding is only supported for FLOAT and DOUBLE, not " + descriptor.getType());
169+
}
170+
}
171+
},
172+
150173
/**
151174
* @deprecated This is no longer used, and has been replaced by {@link #RLE}
152175
* which is combination of bit packing and rle

0 commit comments

Comments
 (0)