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 a144c951 perf(parquet): batch DELTA_BYTE_ARRAY encoding (#1179)
a144c951 is described below
commit a144c95147b38ce543d0b5e7fc4489671176995d
Author: Minh Vu <[email protected]>
AuthorDate: Fri Aug 28 22:30:18 2026 +0200
perf(parquet): batch DELTA_BYTE_ARRAY encoding (#1179)
## Summary
- Batch DELTA_BYTE_ARRAY prefix lengths and suffixes in chunks of 256
values.
- Reuse fixed scratch space for DELTA_LENGTH_BYTE_ARRAY lengths.
- Preserve the previous value across batch boundaries and separate Put
calls.
- Add a boundary round-trip test and representative benchmarks.
## Why
DELTA_BYTE_ARRAY was calling the prefix and suffix encoders once per
value. Large pages therefore paid for many small calls and temporary
length slices.
Apache Arrow C++ already uses fixed-size batching for this encoder. This
change follows the same shape while keeping the existing Go encoding
format and last-value behavior.
## Correctness
- The encoded layout is unchanged: prefix lengths first, followed by the
suffix stream.
- The first value still has a zero prefix.
- Empty suffixes still work as before.
- The new test compares one large Put call with Put calls split around a
batch boundary, then decodes and compares every value.
## Benchmark
64K values on an Apple M1 Pro, 1 second per sample, 3 samples:
| Case | Before | After | Change |
| --- | ---: | ---: | ---: |
| prefix-heavy | 2.09 ms/op | 1.75 ms/op | ~16% faster |
| low-prefix | 1.68 ms/op | 1.47 ms/op | ~12% faster |
Allocation counts stayed the same in both cases.
## Checks
- `PARQUET_TEST_DATA=<parquet-testing-data> go test ./parquet/...`
- `go test -race ./parquet/internal/encoding -count=1`
- `git diff --check`
No public API changes.
---
parquet/internal/encoding/delta_byte_array.go | 56 ++++++++-------
.../encoding/delta_byte_array_benchmark_test.go | 78 +++++++++++++++++++++
parquet/internal/encoding/delta_byte_array_test.go | 46 +++++++++++++
.../encoding/delta_byte_array_validation_test.go | 23 +++++++
.../internal/encoding/delta_length_byte_array.go | 18 ++---
.../encoding/delta_length_byte_array_test.go | 80 ++++++++++++++++++++++
6 files changed, 268 insertions(+), 33 deletions(-)
diff --git a/parquet/internal/encoding/delta_byte_array.go
b/parquet/internal/encoding/delta_byte_array.go
index 948de218..86b7d585 100644
--- a/parquet/internal/encoding/delta_byte_array.go
+++ b/parquet/internal/encoding/delta_byte_array.go
@@ -38,9 +38,24 @@ type DeltaByteArrayEncoder struct {
prefixEncoder *DeltaBitPackInt32Encoder
suffixEncoder *DeltaLengthByteArrayEncoder
+ prefixLengths [deltaByteArrayBatchSize]int32
+ suffixes [deltaByteArrayBatchSize]parquet.ByteArray
+
lastVal parquet.ByteArray
}
+const deltaByteArrayBatchSize = 256
+
+func commonPrefixLength(left, right parquet.ByteArray) int {
+ maximum := min(left.Len(), right.Len())
+ for i := 0; i < maximum; i++ {
+ if left[i] != right[i] {
+ return i
+ }
+ }
+ return maximum
+}
+
func (enc *DeltaByteArrayEncoder) EstimatedDataEncodedSize() int64 {
prefixEstimatedSize := int64(0)
if enc.prefixEncoder != nil {
@@ -58,15 +73,15 @@ func (enc *DeltaByteArrayEncoder) initEncoders() {
encoder: newEncoderBase(enc.encoding, nil, enc.mem),
}
enc.suffixEncoder = &DeltaLengthByteArrayEncoder{
- newEncoderBase(enc.encoding, nil, enc.mem),
- &DeltaBitPackInt32Encoder{
+ encoder: newEncoderBase(enc.encoding, nil, enc.mem),
+ lengthEncoder: &DeltaBitPackInt32Encoder{
encoder: newEncoderBase(enc.encoding, nil, enc.mem),
},
}
}
// Type returns the underlying physical type this operates on, in this case
ByteArrays only
-func (DeltaByteArrayEncoder) Type() parquet.Type { return
parquet.Types.ByteArray }
+func (*DeltaByteArrayEncoder) Type() parquet.Type { return
parquet.Types.ByteArray }
// Put writes a slice of ByteArrays to the encoder
func (enc *DeltaByteArrayEncoder) Put(in []parquet.ByteArray) {
@@ -74,39 +89,29 @@ func (enc *DeltaByteArrayEncoder) Put(in
[]parquet.ByteArray) {
return
}
- var suf parquet.ByteArray
if enc.prefixEncoder == nil { // initialize our encoders if we haven't
yet
enc.initEncoders()
- enc.prefixEncoder.Put([]int32{0})
- suf = in[0]
- enc.lastVal = in[0]
- enc.suffixEncoder.Put([]parquet.ByteArray{suf})
- in = in[1:]
}
- // for each value, figure out the common prefix with the previous value
- // and then write the prefix length and the suffix.
- for _, val := range in {
- l1 := enc.lastVal.Len()
- l2 := val.Len()
- j := 0
- for j < l1 && j < l2 {
- if enc.lastVal[j] != val[j] {
- break
- }
- j++
+ lastVal := enc.lastVal
+ for offset := 0; offset < len(in); offset += deltaByteArrayBatchSize {
+ batchSize := min(deltaByteArrayBatchSize, len(in)-offset)
+ for i, val := range in[offset : offset+batchSize] {
+ prefixLength := commonPrefixLength(lastVal, val)
+ lastVal = val
+ enc.prefixLengths[i] = int32(prefixLength)
+ enc.suffixes[i] = val[prefixLength:]
}
- enc.prefixEncoder.Put([]int32{int32(j)})
- suf = val[j:]
- enc.suffixEncoder.Put([]parquet.ByteArray{suf})
- enc.lastVal = val
+ enc.suffixEncoder.Put(enc.suffixes[:batchSize])
+ enc.prefixEncoder.Put(enc.prefixLengths[:batchSize])
+ clear(enc.suffixes[:batchSize])
}
// do the memcpy after the loops to keep a copy of the lastVal
// we do a copy here so that we only copy and keep a reference
// to the suffix, and aren't forcing the *entire* value to stay
// in memory while we have this reference to just the suffix.
- enc.lastVal = append([]byte{}, enc.lastVal...)
+ enc.lastVal = append(enc.lastVal[:0], lastVal...)
}
// PutSpaced is like Put, but assumes the data is already spaced for nulls and
uses the bitmap provided and offset
@@ -143,6 +148,7 @@ func (enc *DeltaByteArrayEncoder) FlushValues() (Buffer,
error) {
ret.ResizeNoShrink(prefixBuf.Len() + suffixBuf.Len())
copy(ret.Bytes(), prefixBuf.Bytes())
copy(ret.Bytes()[prefixBuf.Len():], suffixBuf.Bytes())
+ enc.lastVal = nil
return poolBuffer{ret}, nil
}
diff --git a/parquet/internal/encoding/delta_byte_array_benchmark_test.go
b/parquet/internal/encoding/delta_byte_array_benchmark_test.go
new file mode 100644
index 00000000..37c677b6
--- /dev/null
+++ b/parquet/internal/encoding/delta_byte_array_benchmark_test.go
@@ -0,0 +1,78 @@
+// 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"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+)
+
+func BenchmarkDeltaByteArrayEncoding(b *testing.B) {
+ for _, test := range []struct {
+ name string
+ value func(int) []byte
+ }{
+ {
+ name: "prefix-heavy",
+ value: func(i int) []byte {
+ return []byte(fmt.Sprintf("partition/%06d", i))
+ },
+ },
+ {
+ name: "low-prefix",
+ value: func(i int) []byte {
+ return []byte(fmt.Sprintf("%c/%06d",
byte(i%251), i))
+ },
+ },
+ } {
+ b.Run(test.name, func(b *testing.B) {
+ const nvalues = 64 * 1024
+ values := make([]parquet.ByteArray, nvalues)
+ var inputBytes int64
+ for i := range values {
+ values[i] = test.value(i)
+ inputBytes += int64(values[i].Len())
+ }
+
+ for _, putSize := range []int{1, 8, 32,
deltaByteArrayBatchSize, nvalues} {
+ putSize := putSize
+ b.Run(fmt.Sprintf("put-%d", putSize), func(b
*testing.B) {
+ b.SetBytes(inputBytes)
+ b.ReportAllocs()
+ for b.Loop() {
+ enc :=
NewEncoder(parquet.Types.ByteArray, parquet.Encodings.DeltaByteArray,
+ false, nil,
memory.DefaultAllocator).(ByteArrayEncoder)
+ for offset := 0; offset <
len(values); offset += putSize {
+ end :=
min(offset+putSize, len(values))
+
enc.Put(values[offset:end])
+ }
+ buf, err := enc.FlushValues()
+ if err != nil {
+ enc.Release()
+ b.Fatal(err)
+ }
+ buf.Release()
+ enc.Release()
+ }
+ })
+ }
+ })
+ }
+}
diff --git a/parquet/internal/encoding/delta_byte_array_test.go
b/parquet/internal/encoding/delta_byte_array_test.go
index e3eefe06..8f918a8e 100644
--- a/parquet/internal/encoding/delta_byte_array_test.go
+++ b/parquet/internal/encoding/delta_byte_array_test.go
@@ -23,6 +23,7 @@ import (
"github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/parquet"
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
)
func TestDeltaByteArrayDecoder_SetData(t *testing.T) {
@@ -46,3 +47,48 @@ func TestDeltaByteArrayDecoder_SetData(t *testing.T) {
})
}
}
+
+func TestDeltaByteArrayEncoderPreservesLastValueAcrossBatches(t *testing.T) {
+ values := make([]parquet.ByteArray, deltaByteArrayBatchSize*2+3)
+ for i := range values {
+ values[i] =
parquet.ByteArray(fmt.Sprintf("partition-%03d/value-%03d", i/7, i%7))
+ }
+ values[0] = parquet.ByteArray{}
+ values[deltaByteArrayBatchSize-1] =
parquet.ByteArray("boundary/repeated")
+ values[deltaByteArrayBatchSize] = parquet.ByteArray("boundary/repeated")
+ values[deltaByteArrayBatchSize+1] = parquet.ByteArray("boundary")
+ values[deltaByteArrayBatchSize*2] = parquet.ByteArray("boundary")
+
+ encode := func(batches ...[]parquet.ByteArray) []byte {
+ t.Helper()
+ enc := NewEncoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaByteArray, false, nil,
memory.DefaultAllocator).(ByteArrayEncoder)
+ defer enc.Release()
+ for _, batch := range batches {
+ enc.Put(batch)
+ }
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer buf.Release()
+ return append([]byte(nil), buf.Bytes()...)
+ }
+
+ want := encode(values)
+ for _, split := range []int{0, 1, deltaByteArrayBatchSize - 1,
deltaByteArrayBatchSize,
+ deltaByteArrayBatchSize + 1, deltaByteArrayBatchSize*2 - 1,
deltaByteArrayBatchSize * 2,
+ len(values) - 1, len(values)} {
+ t.Run(fmt.Sprintf("split-%d", split), func(t *testing.T) {
+ got := encode(values[:split], values[split:])
+ require.Equal(t, want, got)
+
+ dec := NewDecoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaByteArray, nil,
memory.DefaultAllocator).(ByteArrayDecoder)
+ require.NoError(t, dec.SetData(len(values), got))
+ out := make([]parquet.ByteArray, len(values))
+ decoded, err := dec.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, len(values), decoded)
+ for i := range values {
+ assert.Equal(t, string(values[i]),
string(out[i]), "value %d", i)
+ }
+ })
+ }
+}
diff --git a/parquet/internal/encoding/delta_byte_array_validation_test.go
b/parquet/internal/encoding/delta_byte_array_validation_test.go
index 0a584835..f4794978 100644
--- a/parquet/internal/encoding/delta_byte_array_validation_test.go
+++ b/parquet/internal/encoding/delta_byte_array_validation_test.go
@@ -64,6 +64,29 @@ func TestDeltaByteArrayDecoderResetsBetweenPages(t
*testing.T) {
require.Error(t, err)
}
+func TestDeltaByteArrayEncoderResetsBetweenPages(t *testing.T) {
+ enc := NewEncoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaByteArray, false, nil,
memory.DefaultAllocator).(ByteArrayEncoder)
+ defer enc.Release()
+
+ enc.Put([]parquet.ByteArray{parquet.ByteArray("prefix/old")})
+ first, err := enc.FlushValues()
+ require.NoError(t, err)
+ first.Release()
+
+ enc.Put([]parquet.ByteArray{parquet.ByteArray("prefix/new")})
+ second, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer second.Release()
+
+ dec := NewDecoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaByteArray, nil,
memory.DefaultAllocator).(ByteArrayDecoder)
+ require.NoError(t, dec.SetData(1, second.Bytes()))
+ out := make([]parquet.ByteArray, 1)
+ decoded, err := dec.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, 1, decoded)
+ require.Equal(t, parquet.ByteArray("prefix/new"), out[0])
+}
+
func TestDeltaByteArrayDecoderRejectsInvalidPrefixes(t *testing.T) {
tests := []struct {
name string
diff --git a/parquet/internal/encoding/delta_length_byte_array.go
b/parquet/internal/encoding/delta_length_byte_array.go
index ffc8096a..30ce53ff 100644
--- a/parquet/internal/encoding/delta_length_byte_array.go
+++ b/parquet/internal/encoding/delta_length_byte_array.go
@@ -38,18 +38,20 @@ type DeltaLengthByteArrayEncoder struct {
encoder
lengthEncoder *DeltaBitPackInt32Encoder
+ lengths [deltaByteArrayBatchSize]int32
}
// Put writes the provided slice of byte arrays to the encoder
func (enc *DeltaLengthByteArrayEncoder) Put(in []parquet.ByteArray) {
- lengths := make([]int32, len(in))
- totalLen := int(0)
- for idx, val := range in {
- lengths[idx] = int32(val.Len())
- totalLen += val.Len()
+ totalLen := 0
+ for i := 0; i < len(in); i += deltaByteArrayBatchSize {
+ batchSize := min(deltaByteArrayBatchSize, len(in)-i)
+ for j, val := range in[i : i+batchSize] {
+ enc.lengths[j] = int32(val.Len())
+ totalLen += val.Len()
+ }
+ enc.lengthEncoder.Put(enc.lengths[:batchSize])
}
-
- enc.lengthEncoder.Put(lengths)
enc.sink.Reserve(totalLen)
for _, val := range in {
enc.sink.UnsafeWrite(val)
@@ -69,7 +71,7 @@ func (enc *DeltaLengthByteArrayEncoder) PutSpaced(in
[]parquet.ByteArray, validB
}
// Type returns the underlying type which is handled by this encoder,
ByteArrays only.
-func (DeltaLengthByteArrayEncoder) Type() parquet.Type {
+func (*DeltaLengthByteArrayEncoder) Type() parquet.Type {
return parquet.Types.ByteArray
}
diff --git a/parquet/internal/encoding/delta_length_byte_array_test.go
b/parquet/internal/encoding/delta_length_byte_array_test.go
new file mode 100644
index 00000000..7ed52d8a
--- /dev/null
+++ b/parquet/internal/encoding/delta_length_byte_array_test.go
@@ -0,0 +1,80 @@
+// 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"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/stretchr/testify/require"
+)
+
+func TestDeltaLengthByteArrayEncoderPreservesBatches(t *testing.T) {
+ for _, nvalues := range []int{255, 256, 257, 511, 512, 513} {
+ nvalues := nvalues
+ t.Run(fmt.Sprintf("%d-values", nvalues), func(t *testing.T) {
+ values := make([]parquet.ByteArray, nvalues)
+ for i := range values {
+ values[i] =
parquet.ByteArray(fmt.Sprintf("value-%d", i))
+ }
+
+ encode := func(batches ...[]parquet.ByteArray) []byte {
+ t.Helper()
+ enc := NewEncoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaLengthByteArray,
+ false, nil,
memory.DefaultAllocator).(ByteArrayEncoder)
+ defer enc.Release()
+ for _, batch := range batches {
+ enc.Put(batch)
+ }
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer buf.Release()
+ return append([]byte(nil), buf.Bytes()...)
+ }
+
+ roundTrip := func(data []byte) {
+ t.Helper()
+ dec := NewDecoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaLengthByteArray,
+ nil,
memory.DefaultAllocator).(ByteArrayDecoder)
+ require.NoError(t, dec.SetData(nvalues, data))
+ out := make([]parquet.ByteArray, nvalues)
+ decoded, err := dec.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, nvalues, decoded)
+ require.Equal(t, values, out)
+ }
+
+ want := encode(values)
+ splits := []int{1, nvalues / 2, nvalues - 1}
+ for _, boundary := range []int{deltaByteArrayBatchSize
- 1, deltaByteArrayBatchSize, deltaByteArrayBatchSize + 1} {
+ if boundary > 0 && boundary < nvalues {
+ splits = append(splits, boundary)
+ }
+ }
+ for _, split := range splits {
+ split := split
+ t.Run(fmt.Sprintf("split-%d", split), func(t
*testing.T) {
+ got := encode(values[:split],
values[split:])
+ require.Equal(t, want, got)
+ roundTrip(got)
+ })
+ }
+ })
+ }
+}
