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 df81ee9d fix(parquet/internal/encoding): do not leak delta bit width
buffers from decoders (#1394)
df81ee9d is described below
commit df81ee9d30b4fa7c7f69a987ffb712df56da66a4
Author: Matt Topol <[email protected]>
AuthorDate: Wed Oct 21 22:11:10 2026 +0100
fix(parquet/internal/encoding): do not leak delta bit width buffers from
decoders (#1394)
### Rationale for this change
`deltaBitPackDecoder` allocated its per-miniblock bit-width buffer
(`deltaBitWidths`) from the reader's allocator, but decoders have no
`Release`, so the buffer was never returned. This leaks for:
- DELTA_BINARY_PACKED int32/int64 columns, and
- the temporary length decoders that DELTA_LENGTH_BYTE_ARRAY (one) and
DELTA_BYTE_ARRAY (two) create inside `SetData` on every page.
With a checked allocator, reading a 545-row binary column back through
pqarrow left 64 bytes (DELTA_LENGTH_BYTE_ARRAY) and 128 bytes
(DELTA_BYTE_ARRAY) allocated after everything was released, more for
multi-page columns. Found while reviewing #1392.
### What changes are included in this PR?
`deltaBitWidths` is now a reused `[]byte` on the Go heap, grown only
when needed (the same pattern as the decoders' existing
`lengthScratch`). The size bound is unchanged: the header validation
already caps the miniblock count. No API change.
### Are these changes tested?
Yes. `TestDeltaDecodersDoNotRetainAllocations` (in
`delta_bit_packing_validation_test.go`) decodes DELTA_BINARY_PACKED,
DELTA_LENGTH_BYTE_ARRAY and DELTA_BYTE_ARRAY data with a checked
allocator and asserts nothing is left allocated. All three subtests fail
on main (`LEAK of 64 bytes`) and pass with the change.
- `go test ./parquet/internal/encoding ./parquet/file
./parquet/pqarrow`: ok
- `go test -tags noasm ./parquet/internal/encoding` and `GOARCH=386 go
test ./parquet/internal/encoding`: ok
- `go vet ./parquet/internal/encoding` and `gofmt -l`: clean
### Are there any user-facing changes?
No API changes. Readers using a checked or C-backed allocator no longer
leak a small buffer per delta-encoded page.
---
parquet/internal/encoding/delta_bit_packing.go | 18 +++---
.../encoding/delta_bit_packing_validation_test.go | 64 ++++++++++++++++++++++
2 files changed, 74 insertions(+), 8 deletions(-)
diff --git a/parquet/internal/encoding/delta_bit_packing.go
b/parquet/internal/encoding/delta_bit_packing.go
index 3add6947..dba6f6f3 100644
--- a/parquet/internal/encoding/delta_bit_packing.go
+++ b/parquet/internal/encoding/delta_bit_packing.go
@@ -50,7 +50,9 @@ type deltaBitPackDecoder[T int32 | int64] struct {
minDelta int64
miniBlockIdx uint64
- deltaBitWidths *memory.Buffer
+ // deltaBitWidths holds one byte per miniblock. It lives on the Go heap
+ // because decoders have no Release to return allocator memory.
+ deltaBitWidths []byte
deltaBitWidth byte
totalValues uint64
@@ -78,9 +80,6 @@ func (d *deltaBitPackDecoder[T]) SetData(nvalues int, data
[]byte) error {
d.bitdecoder = utils.NewBitReader(bytes.NewReader(d.data))
d.currentBlockVals = 0
d.currentMiniBlockVals = 0
- if d.deltaBitWidths == nil {
- d.deltaBitWidths = memory.NewResizableBuffer(d.mem)
- }
var ok bool
d.blockSize, ok = d.bitdecoder.GetVlqInt()
@@ -144,17 +143,20 @@ func (d *deltaBitPackDecoder[T]) initBlock() error {
}
// ensure we have enough space for our miniblocks to decode the widths
- d.deltaBitWidths.Resize(int(d.miniBlocksPerBlock))
+ if uint64(cap(d.deltaBitWidths)) < d.miniBlocksPerBlock {
+ d.deltaBitWidths = make([]byte, d.miniBlocksPerBlock)
+ }
+ d.deltaBitWidths = d.deltaBitWidths[:d.miniBlocksPerBlock]
var err error
for i := uint64(0); i < d.miniBlocksPerBlock; i++ {
- if d.deltaBitWidths.Bytes()[i], err = d.bitdecoder.ReadByte();
err != nil {
+ if d.deltaBitWidths[i], err = d.bitdecoder.ReadByte(); err !=
nil {
return err
}
}
d.miniBlockIdx = 0
- d.deltaBitWidth = d.deltaBitWidths.Bytes()[0]
+ d.deltaBitWidth = d.deltaBitWidths[0]
d.currentBlockVals = uint32(d.blockSize)
return nil
}
@@ -165,7 +167,7 @@ func (d *deltaBitPackDecoder[T]) unpackNextMini() error {
} else {
d.miniBlockValues = d.miniBlockValues[:0]
}
- d.deltaBitWidth = d.deltaBitWidths.Bytes()[int(d.miniBlockIdx)]
+ d.deltaBitWidth = d.deltaBitWidths[int(d.miniBlockIdx)]
maxBitWidth := byte(64)
if d.Type() == parquet.Types.Int32 {
maxBitWidth = 32
diff --git a/parquet/internal/encoding/delta_bit_packing_validation_test.go
b/parquet/internal/encoding/delta_bit_packing_validation_test.go
index 0f71f804..8122dbc1 100644
--- a/parquet/internal/encoding/delta_bit_packing_validation_test.go
+++ b/parquet/internal/encoding/delta_bit_packing_validation_test.go
@@ -17,6 +17,8 @@
package encoding
import (
+ "bytes"
+ "fmt"
"io"
"testing"
@@ -126,3 +128,65 @@ func TestDeltaBitPackEncoderReleasesSpacedScratch(t
*testing.T) {
enc.PutSpaced([]int32{1, 2, 3, 4}, []byte{0x0f}, 0)
enc.Release()
}
+
+// The delta decoders, including the length decoders that
DELTA_LENGTH_BYTE_ARRAY
+// and DELTA_BYTE_ARRAY create inside SetData, have no Release, so decoding
must
+// not leave anything allocated from the reader's allocator.
+func TestDeltaDecodersDoNotRetainAllocations(t *testing.T) {
+ const n = 545
+ ints := make([]int32, n)
+ strs := make([]parquet.ByteArray, n)
+ for i := range n {
+ ints[i] = int32(i * 7)
+ strs[i] = parquet.ByteArray(fmt.Sprintf("prefix/%06d", i))
+ }
+
+ encode := func(enc TypedEncoder) []byte {
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer buf.Release()
+ enc.Release()
+ return bytes.Clone(buf.Bytes())
+ }
+
+ for _, tc := range []struct {
+ name string
+ typ parquet.Type
+ enc parquet.Encoding
+ data func() []byte
+ decode func(TypedDecoder) (int, error)
+ }{
+ {"delta_binary_packed", parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked,
+ func() []byte {
+ enc := NewEncoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, false, nil,
memory.DefaultAllocator).(Int32Encoder)
+ enc.Put(ints)
+ return encode(enc)
+ },
+ func(d TypedDecoder) (int, error) { return
d.(Int32Decoder).Decode(make([]int32, n)) }},
+ {"delta_length_byte_array", parquet.Types.ByteArray,
parquet.Encodings.DeltaLengthByteArray,
+ func() []byte {
+ enc := NewEncoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaLengthByteArray, false, nil,
memory.DefaultAllocator).(ByteArrayEncoder)
+ enc.Put(strs)
+ return encode(enc)
+ },
+ func(d TypedDecoder) (int, error) { return
d.(ByteArrayDecoder).Decode(make([]parquet.ByteArray, n)) }},
+ {"delta_byte_array", parquet.Types.ByteArray,
parquet.Encodings.DeltaByteArray,
+ func() []byte {
+ enc := NewEncoder(parquet.Types.ByteArray,
parquet.Encodings.DeltaByteArray, false, nil,
memory.DefaultAllocator).(ByteArrayEncoder)
+ enc.Put(strs)
+ return encode(enc)
+ },
+ func(d TypedDecoder) (int, error) { return
d.(ByteArrayDecoder).Decode(make([]parquet.ByteArray, n)) }},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ data := tc.data()
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ dec := NewDecoder(tc.typ, tc.enc, nil, mem)
+ require.NoError(t, dec.SetData(n, data))
+ got, err := tc.decode(dec)
+ require.NoError(t, err)
+ require.Equal(t, n, got)
+ mem.AssertSize(t, 0)
+ })
+ }
+}
