This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new f5331867 perf(parquet/internal/encoding): fast-path constant delta
bitpacked blocks (#1396)
f5331867 is described below
commit f53318674137ad9b0f7701a28ae8d00bccca20f4
Author: Minh Vu <[email protected]>
AuthorDate: Thu Oct 22 19:49:41 2026 +0200
perf(parquet/internal/encoding): fast-path constant delta bitpacked blocks
(#1396)
### What
Speed up DELTA_BINARY_PACKED encoding for constant and sparse deltas.
### Why
Zero-width miniblocks produce no payload, but the encoder still does
per-value work.
### Implementation
- Encode uniform blocks without a payload.
- Skip writes for zero-width miniblocks in mixed blocks.
### Benchmark
65,536 values, Go 1.26.3, Apple M1 Pro, median of 5 runs.
| Metric | Before | After | Change |
|---|---:|---:|---:|
| int32 constant | 416.6 µs/op | 197.8 µs/op | 2.1x faster |
| int32 sparse | 420.2 µs/op | 312.1 µs/op | 1.35x faster |
| int64 constant | 411.5 µs/op | 193.1 µs/op | 2.1x faster |
| int64 sparse | 420.7 µs/op | 314.7 µs/op | 1.34x faster |
---------
Co-authored-by: Matt Topol <[email protected]>
---
parquet/internal/encoding/delta_bit_packing.go | 41 +++--
.../encoding/delta_bit_packing_block_bench_test.go | 81 ++++++++++
.../encoding/delta_bit_packing_fast_path_test.go | 174 +++++++++++++++++++++
3 files changed, 285 insertions(+), 11 deletions(-)
diff --git a/parquet/internal/encoding/delta_bit_packing.go
b/parquet/internal/encoding/delta_bit_packing.go
index dba6f6f3..884b115f 100644
--- a/parquet/internal/encoding/delta_bit_packing.go
+++ b/parquet/internal/encoding/delta_bit_packing.go
@@ -382,12 +382,28 @@ func (enc *deltaBitPackEncoder[T]) flushBlock() {
return
}
- // determine the minimum delta value
- minDelta := int64(math.MaxInt64)
- for _, delta := range enc.deltas {
+ // Find the minimum delta and check whether the block is constant in one
+ // pass. A separate equality scan hurts mixed blocks with a late
outlier.
+ firstDelta := enc.deltas[0]
+ minDelta := firstDelta
+ allSame := true
+ for _, delta := range enc.deltas[1:] {
if delta < minDelta {
minDelta = delta
}
+ if delta != firstDelta {
+ allSame = false
+ }
+ }
+ if allSame {
+ // Identical deltas need only the minimum and zero widths, no
payload.
+ enc.bitWriter.WriteZigZagVlqInt(firstDelta)
+ offset, _ := enc.bitWriter.SkipBytes(int(enc.numMiniBlocks))
+ // Explicitly clear the reserved widths for a reused buffer.
+ var widths [defaultNumMiniBlocks]byte
+ enc.bitWriter.WriteAt(widths[:], int64(offset))
+ enc.deltas = enc.deltas[:0]
+ return
}
enc.bitWriter.WriteZigZagVlqInt(minDelta)
@@ -412,17 +428,20 @@ func (enc *deltaBitPackEncoder[T]) flushBlock() {
// write out the bit width we used into the bytes we reserved
earlier
enc.bitWriter.WriteAt([]byte{byte(width)}, int64(offset+i))
- // write out our deltas
- for _, val := range enc.deltas[start : start+int(n)] {
- enc.bitWriter.WriteValue(uint64(val-minDelta), width)
+ // A zero-width miniblock has no packed payload. Avoid per-value
+ // writes for these miniblocks even if the full block is not
constant.
+ if width != 0 {
+ for _, val := range enc.deltas[start : start+int(n)] {
+ enc.bitWriter.WriteValue(uint64(val-minDelta),
width)
+ }
+
+ // pad the last block if n < miniBlockSize
+ for pad := n; pad < int64(enc.miniBlockSize); pad++ {
+ enc.bitWriter.WriteValue(0, width)
+ }
}
valuesToWrite -= n
-
- // pad the last block if n < miniBlockSize
- for ; n < int64(enc.miniBlockSize); n++ {
- enc.bitWriter.WriteValue(0, width)
- }
}
enc.deltas = enc.deltas[:0]
}
diff --git a/parquet/internal/encoding/delta_bit_packing_block_bench_test.go
b/parquet/internal/encoding/delta_bit_packing_block_bench_test.go
new file mode 100644
index 00000000..47ee2ba2
--- /dev/null
+++ b/parquet/internal/encoding/delta_bit_packing_block_bench_test.go
@@ -0,0 +1,81 @@
+// 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 encoding
+
+import (
+ "math/rand"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+)
+
+func benchmarkDeltaBitPackBlockFastPath[T int32 | int64](b *testing.B, typ
parquet.Type, bytesPerValue int) {
+ const length = 64 * 1024
+
+ for _, pattern := range []string{"constant", "sparse", "last-outlier",
"alternating", "random"} {
+ b.Run(pattern, func(b *testing.B) {
+ values := make([]T, length)
+ rng := rand.New(rand.NewSource(42))
+ for i := 1; i < length; i++ {
+ switch pattern {
+ case "constant":
+ values[i] = values[i-1] + T(3)
+ case "sparse":
+ step := T(3)
+ if i%128 == 90 {
+ step = 7
+ }
+ values[i] = values[i-1] + step
+ case "last-outlier":
+ step := T(3)
+ // Delta indices 128, 256, ... end each
full block.
+ if i%128 == 0 {
+ step = -7
+ }
+ values[i] = values[i-1] + step
+ case "alternating":
+ values[i] = values[i-1] + T(i%2+1)
+ case "random":
+ values[i] = T(rng.Uint64())
+ }
+ }
+
+ enc := NewEncoder(typ,
parquet.Encodings.DeltaBinaryPacked, false, nil,
memory.DefaultAllocator).(*deltaBitPackEncoder[T])
+ defer enc.Release()
+
+ b.SetBytes(int64(length * bytesPerValue))
+ b.ReportAllocs()
+ for b.Loop() {
+ enc.Put(values)
+ buf, err := enc.FlushValues()
+ if err != nil {
+ b.Fatal(err)
+ }
+ buf.Release()
+ }
+ })
+ }
+}
+
+func BenchmarkDeltaBitPackBlockFastPathInt32(b *testing.B) {
+ benchmarkDeltaBitPackBlockFastPath[int32](b, parquet.Types.Int32, 4)
+}
+
+func BenchmarkDeltaBitPackBlockFastPathInt64(b *testing.B) {
+ benchmarkDeltaBitPackBlockFastPath[int64](b, parquet.Types.Int64, 8)
+}
diff --git a/parquet/internal/encoding/delta_bit_packing_fast_path_test.go
b/parquet/internal/encoding/delta_bit_packing_fast_path_test.go
new file mode 100644
index 00000000..e3da8d34
--- /dev/null
+++ b/parquet/internal/encoding/delta_bit_packing_fast_path_test.go
@@ -0,0 +1,174 @@
+// 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 encoding
+
+import (
+ "fmt"
+ "math"
+ "math/rand"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/stretchr/testify/require"
+)
+
+func TestDeltaBitPackConstantBlockWireFormat(t *testing.T) {
+ t.Run("full positive block", func(t *testing.T) {
+ values := make([]int32, 129)
+ for i := range values {
+ values[i] = 10 + int32(i)*2
+ }
+
+ enc := NewEncoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, false, nil,
memory.DefaultAllocator).(*deltaBitPackEncoder[int32])
+ defer enc.Release()
+ enc.Put(values)
+
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer buf.Release()
+
+ // Block size 128, 4 miniblocks, 129 values, first value 10,
+ // minimum delta 2, then four zero bit widths and no payload.
+ require.Equal(t, []byte{0x80, 0x01, 0x04, 0x81, 0x01, 0x14,
0x04, 0, 0, 0, 0}, buf.Bytes())
+ })
+
+ t.Run("partial negative block", func(t *testing.T) {
+ values := []int64{5, 4, 3, 2, 1, 0, -1}
+ enc := NewEncoder(parquet.Types.Int64,
parquet.Encodings.DeltaBinaryPacked, false, nil,
memory.DefaultAllocator).(*deltaBitPackEncoder[int64])
+ defer enc.Release()
+ enc.Put(values)
+
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer buf.Release()
+
+ // The trailing miniblocks are also written as zero-width.
+ require.Equal(t, []byte{0x80, 0x01, 0x04, 0x07, 0x0a, 0x01, 0,
0, 0, 0}, buf.Bytes())
+ })
+}
+
+func testDeltaBitPackBlockRoundTrip[T int32 | int64](t *testing.T, typ
parquet.Type) {
+ t.Helper()
+
+ encode := func(parts ...[]T) []byte {
+ t.Helper()
+ enc := NewEncoder(typ, parquet.Encodings.DeltaBinaryPacked,
false, nil, memory.DefaultAllocator).(*deltaBitPackEncoder[T])
+ defer enc.Release()
+ for _, part := range parts {
+ enc.Put(part)
+ }
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer buf.Release()
+ return append([]byte(nil), buf.Bytes()...)
+ }
+
+ verify := func(t *testing.T, values []T) {
+ t.Helper()
+ n := len(values)
+ wire := encode(values)
+ split := max(1, n/2)
+ require.Equal(t, wire, encode(values[:1], values[1:split],
values[split:]))
+
+ dec := NewDecoder(typ, parquet.Encodings.DeltaBinaryPacked,
nil, memory.DefaultAllocator).(*deltaBitPackDecoder[T])
+ require.NoError(t, dec.SetData(n, wire))
+ out := make([]T, n)
+ got, err := dec.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, n, got)
+ require.Equal(t, values, out)
+ }
+
+ for _, n := range []int{1, 2, 31, 32, 33, 127, 128, 129, 130, 255, 257,
1001} {
+ for _, pattern := range []string{"constant", "zero", "sparse",
"alternating", "random", "overflow"} {
+ t.Run(fmt.Sprintf("%s/%d", pattern, n), func(t
*testing.T) {
+ values := make([]T, n)
+ rng := rand.New(rand.NewSource(int64(n)))
+ switch pattern {
+ case "overflow":
+ raw := int64(math.MaxInt32 - 2)
+ if typ == parquet.Types.Int64 {
+ raw = int64(math.MaxInt64 - 2)
+ }
+ values[0] = T(raw)
+ default:
+ values[0] = T(11)
+ }
+
+ for i := 1; i < n; i++ {
+ switch pattern {
+ case "constant":
+ values[i] = values[i-1] + T(3)
+ case "zero":
+ values[i] = values[i-1]
+ case "sparse":
+ step := T(3)
+ if i%128 == 90 {
+ step = 7
+ }
+ values[i] = values[i-1] + step
+ case "alternating":
+ values[i] = values[i-1] +
T(i%2+1)
+ case "random":
+ values[i] = T(rng.Uint64())
+ case "overflow":
+ values[i] = values[i-1] + T(7)
+ }
+ }
+
+ verify(t, values)
+ })
+ }
+ }
+ // The last delta in a block must be included in constant-block
detection.
+ // The 3-value higher case is [0, 3, 10]; 129 values fill one delta
block.
+ for _, n := range []int{3, 129, 130, 257} {
+ for _, outlier := range []struct {
+ name string
+ step T
+ }{
+ {name: "higher", step: 7},
+ {name: "lower", step: -7},
+ } {
+ t.Run(fmt.Sprintf("last-delta-%s/%d", outlier.name, n),
func(t *testing.T) {
+ values := make([]T, n)
+ outlierIndex := n - 1
+ if n == 130 {
+ // The outlier ends the first block,
with another delta after it.
+ outlierIndex = 128
+ }
+ for i := 1; i < n; i++ {
+ step := T(3)
+ if i == outlierIndex {
+ step = outlier.step
+ }
+ values[i] = values[i-1] + step
+ }
+ verify(t, values)
+ })
+ }
+ }
+}
+
+func TestDeltaBitPackConstantAndMixedBlocksInt32(t *testing.T) {
+ testDeltaBitPackBlockRoundTrip[int32](t, parquet.Types.Int32)
+}
+
+func TestDeltaBitPackConstantAndMixedBlocksInt64(t *testing.T) {
+ testDeltaBitPackBlockRoundTrip[int64](t, parquet.Types.Int64)
+}
