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 a7a6c19b perf(parquet): reuse DELTA decoder scratch (#1322)
a7a6c19b is described below
commit a7a6c19b6b8db92e3d01463356e69f26ac569b33
Author: Minh Vu <[email protected]>
AuthorDate: Mon Sep 21 19:11:36 2026 +0200
perf(parquet): reuse DELTA decoder scratch (#1322)
**What**
- Reuse DELTA length and prefix scratch buffers across decoder pages.
- Add tests for scratch reuse and page-to-page correctness.
**Why**
- The column reader caches decoders by encoding.
- Each page currently allocates fresh length and prefix arrays even when
the decoder already has enough capacity.
**Implementation**
- Keep stable backing buffers for decoded lengths and prefix lengths.
- Expose sliced views to the existing Decode and Discard paths.
- Keep the existing page-size behavior when a larger page needs more
capacity.
- The existing repeated `SetData + Decode` benchmark now shows 18 to 15
allocations/op and about 512 KiB less allocation for a full 65,536-value
prefix-heavy page.
Tests:
- `go test ./parquet/internal/encoding -count=1`
- `go test ./parquet/file -run
'^(TestWithEOFReader|TestInvalidHeaders|TestInvalidFooter|TestIncompleteMetadata|TestDeltaLengthByteArrayPackingWithNulls|TestDeltaBinaryPackedMultipleBatches|TestPageStreaming.*|TestPrimitiveReader|TestFullSeekRow|TestSkipEmptyRepeatedRows)$'
-count=1`
- `go test -race ./parquet/internal/encoding -run
'TestDelta(ByteArrayDecoderReusesPageScratch|LengthByteArrayDecoderReusesLengthScratch|ByteArrayDecoderKeepsResultsAcrossPages|ByteArrayDecoderRejectsInvalidPrefixes)$'
-count=1`
- `go vet ./parquet/internal/encoding`
---
parquet/internal/encoding/delta_byte_array.go | 9 ++++-
.../encoding/delta_byte_array_decode_test.go | 36 ++++++++++++++++++
.../internal/encoding/delta_length_byte_array.go | 13 +++++--
.../encoding/delta_length_byte_array_test.go | 44 ++++++++++++++++++++++
4 files changed, 98 insertions(+), 4 deletions(-)
diff --git a/parquet/internal/encoding/delta_byte_array.go
b/parquet/internal/encoding/delta_byte_array.go
index b2ed5bf5..13428aca 100644
--- a/parquet/internal/encoding/delta_byte_array.go
+++ b/parquet/internal/encoding/delta_byte_array.go
@@ -163,6 +163,7 @@ type DeltaByteArrayDecoder struct {
*DeltaLengthByteArrayDecoder
prefixLengths []int32
+ prefixScratch []int32
lastVal parquet.ByteArray
}
@@ -189,7 +190,13 @@ func (d *DeltaByteArrayDecoder) SetData(nvalues int, data
[]byte) error {
return fmt.Errorf("parquet: delta prefix count %d exceeds value
count %d", prefixLenDec.totalValues, nvalues)
}
- d.prefixLengths = make([]int32, prefixLenDec.ValuesLeft())
+ prefixCount := prefixLenDec.ValuesLeft()
+ if cap(d.prefixScratch) < prefixCount {
+ d.prefixScratch = make([]int32, prefixCount)
+ } else {
+ d.prefixScratch = d.prefixScratch[:prefixCount]
+ }
+ d.prefixLengths = d.prefixScratch
// decode all the prefix lengths first so we know how many bytes it
took to get the
// prefix lengths for nvalues
decoded, err := prefixLenDec.Decode(d.prefixLengths)
diff --git a/parquet/internal/encoding/delta_byte_array_decode_test.go
b/parquet/internal/encoding/delta_byte_array_decode_test.go
index 1a5ea5f2..588e6b41 100644
--- a/parquet/internal/encoding/delta_byte_array_decode_test.go
+++ b/parquet/internal/encoding/delta_byte_array_decode_test.go
@@ -151,6 +151,42 @@ func TestDeltaByteArrayDecoderKeepsResultsAcrossPages(t
*testing.T) {
requireDecodedStrings(t, secondOut, secondValues)
}
+func TestDeltaByteArrayDecoderReusesPageScratch(t *testing.T) {
+ firstValues := []string{
+ "partition/000/value/000", "partition/000/value/001",
+ "partition/001/value/000", "partition/001/value/001",
+ }
+ secondValues := []string{"partition/100/value/000",
"partition/100/value/001"}
+ dec := NewDecoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaByteArray,
+ nil, memory.DefaultAllocator).(*DeltaByteArrayDecoder)
+
+ firstData := encodeDeltaByteArrayPage(t, firstValues)
+ require.NoError(t, dec.SetData(len(firstValues), firstData))
+ lengthStart := &dec.lengthScratch[0]
+ prefixStart := &dec.prefixScratch[0]
+ lengthCap := cap(dec.lengthScratch)
+ prefixCap := cap(dec.prefixScratch)
+
+ firstOut := make([]parquet.ByteArray, len(firstValues))
+ decoded, err := dec.Decode(firstOut)
+ require.NoError(t, err)
+ require.Equal(t, len(firstValues), decoded)
+ requireDecodedStrings(t, firstOut, firstValues)
+
+ secondData := encodeDeltaByteArrayPage(t, secondValues)
+ require.NoError(t, dec.SetData(len(secondValues), secondData))
+ require.Equal(t, lengthCap, cap(dec.lengthScratch))
+ require.Equal(t, prefixCap, cap(dec.prefixScratch))
+ require.Same(t, lengthStart, &dec.lengthScratch[0])
+ require.Same(t, prefixStart, &dec.prefixScratch[0])
+
+ secondOut := make([]parquet.ByteArray, len(secondValues))
+ decoded, err = dec.Decode(secondOut)
+ require.NoError(t, err)
+ require.Equal(t, len(secondValues), decoded)
+ requireDecodedStrings(t, secondOut, secondValues)
+}
+
func TestDeltaByteArrayDecoderDecodeSpaced(t *testing.T) {
values := []string{"a/000", "a/001", "b/000", "b/001"}
data := encodeDeltaByteArrayPage(t, values)
diff --git a/parquet/internal/encoding/delta_length_byte_array.go
b/parquet/internal/encoding/delta_length_byte_array.go
index 4b74ff1d..c06a3bb3 100644
--- a/parquet/internal/encoding/delta_length_byte_array.go
+++ b/parquet/internal/encoding/delta_length_byte_array.go
@@ -105,8 +105,9 @@ func (enc *DeltaLengthByteArrayEncoder) FlushValues()
(Buffer, error) {
type DeltaLengthByteArrayDecoder struct {
decoder
- mem memory.Allocator
- lengths []int32
+ mem memory.Allocator
+ lengths []int32
+ lengthScratch []int32
}
// Type returns the underlying type which is handled by this encoder,
ByteArrays only.
@@ -130,7 +131,13 @@ func (d *DeltaLengthByteArrayDecoder) SetData(nvalues int,
data []byte) error {
if dec.totalValues > uint64(nvalues) {
return fmt.Errorf("parquet: delta length count %d exceeds value
count %d", dec.totalValues, nvalues)
}
- d.lengths = make([]int32, dec.totalValues)
+ lengthCount := int(dec.totalValues)
+ if cap(d.lengthScratch) < lengthCount {
+ d.lengthScratch = make([]int32, lengthCount)
+ } else {
+ d.lengthScratch = d.lengthScratch[:lengthCount]
+ }
+ d.lengths = d.lengthScratch
decoded, err := dec.Decode(d.lengths)
if err != nil {
return err
diff --git a/parquet/internal/encoding/delta_length_byte_array_test.go
b/parquet/internal/encoding/delta_length_byte_array_test.go
index 7ed52d8a..dec0e200 100644
--- a/parquet/internal/encoding/delta_length_byte_array_test.go
+++ b/parquet/internal/encoding/delta_length_byte_array_test.go
@@ -78,3 +78,47 @@ func TestDeltaLengthByteArrayEncoderPreservesBatches(t
*testing.T) {
})
}
}
+
+func TestDeltaLengthByteArrayDecoderReusesLengthScratch(t *testing.T) {
+ firstValues := []parquet.ByteArray{
+ parquet.ByteArray("partition/000/value/000"),
+ parquet.ByteArray("partition/000/value/001"),
+ parquet.ByteArray("partition/001/value/000"),
+ }
+ secondValues :=
[]parquet.ByteArray{parquet.ByteArray("partition/100/value/000")}
+ encode := func(values []parquet.ByteArray) []byte {
+ t.Helper()
+ enc := NewEncoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaLengthByteArray,
+ false, nil, memory.DefaultAllocator).(ByteArrayEncoder)
+ defer enc.Release()
+ enc.Put(values)
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer buf.Release()
+ return append([]byte(nil), buf.Bytes()...)
+ }
+
+ dec := NewDecoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaLengthByteArray,
+ nil, memory.DefaultAllocator).(*DeltaLengthByteArrayDecoder)
+ firstData := encode(firstValues)
+ require.NoError(t, dec.SetData(len(firstValues), firstData))
+ lengthStart := &dec.lengthScratch[0]
+ lengthCap := cap(dec.lengthScratch)
+
+ firstOut := make([]parquet.ByteArray, len(firstValues))
+ decoded, err := dec.Decode(firstOut)
+ require.NoError(t, err)
+ require.Equal(t, len(firstValues), decoded)
+ require.Equal(t, firstValues, firstOut)
+
+ secondData := encode(secondValues)
+ require.NoError(t, dec.SetData(len(secondValues), secondData))
+ require.Equal(t, lengthCap, cap(dec.lengthScratch))
+ require.Same(t, lengthStart, &dec.lengthScratch[0])
+
+ secondOut := make([]parquet.ByteArray, len(secondValues))
+ decoded, err = dec.Decode(secondOut)
+ require.NoError(t, err)
+ require.Equal(t, len(secondValues), decoded)
+ require.Equal(t, secondValues, secondOut)
+}
