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 4c374d65 fix(parquet/internal/encoding): reset delta bit pack encoder
state after flush (#1400)
4c374d65 is described below
commit 4c374d650b9a02fe2b0690f60ec5a4c1f692670a
Author: Matt Topol <[email protected]>
AuthorDate: Wed Oct 21 21:11:27 2026 +0100
fix(parquet/internal/encoding): reset delta bit pack encoder state after
flush (#1400)
### Rationale for this change
`deltaBitPackEncoder.FlushValues` hands the sink's bytes to the caller
but left the bit writer (and the first/current value) in place. The next
`Put` replaces the writer, but a page whose values are all null calls
`Put` with nothing, so its flush still used the old writer's byte count
and copied that many bytes out of the fresh sink buffer:
- **Through a column writer** (pqarrow): no crash and the data reads
back correctly, but an all-null page is padded with the previous page's
length in zeros, and the stale `EstimatedDataEncodedSize` makes the
writer cut pages early. A nullable INT32 column of 20,000 random values,
5,000 nulls and 10 values, written with 1 KB pages, came out at 92,002
bytes / 25 pages on main versus 82,954 bytes / 21 pages with this
change.
- **Using the encoder directly**: after a large page, the next all-null
flush panics (`slice bounds out of range [:96061] with capacity 2048`).
Found while reviewing #1396.
### What changes are included in this PR?
`FlushValues` drops the bit writer and resets the first/current value
after flushing, so an empty page is exactly the header a fresh encoder
writes and the size estimate is 0 until the next `Put`. No API change.
### Are these changes tested?
Yes. `TestDeltaBitPackedReuseAfterAllNullPage` (in `encoding_test.go`,
next to the other delta tests) writes a first page of 3, 1,000 or
100,000 values and then checks that an all-null page is byte-identical
to a fresh encoder's, that the size estimate is 0, and that the
following page still decodes. It fails on main (wrong bytes and
estimate, then a panic) and passes with the change.
- `go test ./parquet/internal/encoding ./parquet/file
./parquet/pqarrow`: ok
- `go test -tags noasm` and `GOARCH=386 go test` for
`./parquet/internal/encoding`: ok; delta tests pass under `-race`
- `go vet ./parquet/internal/encoding` and `gofmt -l`: clean
This touches the same file as #1394 but different functions; the two
merge cleanly.
### Are there any user-facing changes?
No API changes. Delta-encoded columns with all-null pages are written
smaller and with correctly sized pages.
---
parquet/internal/encoding/delta_bit_packing.go | 5 +++
parquet/internal/encoding/encoding_test.go | 48 ++++++++++++++++++++++++++
2 files changed, 53 insertions(+)
diff --git a/parquet/internal/encoding/delta_bit_packing.go
b/parquet/internal/encoding/delta_bit_packing.go
index e0fbcbdd..3add6947 100644
--- a/parquet/internal/encoding/delta_bit_packing.go
+++ b/parquet/internal/encoding/delta_bit_packing.go
@@ -484,12 +484,17 @@ func (enc *deltaBitPackEncoder[T]) FlushValues() (Buffer,
error) {
buffer = buffer[:headerWriter.Written()]
enc.totalVals = 0
+ enc.firstVal, enc.currentVal = 0, 0
if enc.bitWriter != nil {
flushed := enc.sink.Finish()
defer flushed.Release()
buffer = append(buffer,
flushed.Buf()[:enc.bitWriter.Written()]...)
+ // Finish hands the sink's bytes to us, so this writer's
position no longer
+ // matches the sink. The next Put creates a new writer; until
then, a flush
+ // (for example of an all-null page) must write only the header.
+ enc.bitWriter = nil
}
return poolBuffer{memory.NewBufferBytes(buffer)}, nil
}
diff --git a/parquet/internal/encoding/encoding_test.go
b/parquet/internal/encoding/encoding_test.go
index 27ce54a2..a7009b2d 100644
--- a/parquet/internal/encoding/encoding_test.go
+++ b/parquet/internal/encoding/encoding_test.go
@@ -1043,6 +1043,54 @@ func TestDeltaBitPackedInt64DecodeAfterDiscard(t
*testing.T) {
}
}
+// A column writer reuses one encoder for every page. A page whose values are
all
+// null puts nothing, so its flush must produce just the empty-page header,
exactly
+// as a fresh encoder would, and the next page must still round-trip.
+func TestDeltaBitPackedReuseAfterAllNullPage(t *testing.T) {
+ column := schema.NewColumn(schema.NewInt32Node("int32",
parquet.Repetitions.Optional, -1), 1, 0)
+ flush := func(enc encoding.TypedEncoder) []byte {
+ buf, err := enc.FlushValues()
+ require.NoError(t, err)
+ defer buf.Release()
+ return bytes.Clone(buf.Bytes())
+ }
+ allNull := []int32{0, 0, 0}
+
+ fresh := encoding.NewEncoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, false, column,
memory.DefaultAllocator).(encoding.Int32Encoder)
+ defer fresh.Release()
+ fresh.PutSpaced(allNull, []byte{0}, 0)
+ emptyPage := flush(fresh)
+
+ // 100,000 values make the first page larger than the sink's next
buffer.
+ for _, n := range []int{3, 1000, 100_000} {
+ t.Run(fmt.Sprintf("first_page=%d", n), func(t *testing.T) {
+ values := make([]int32, n)
+ for i := range values {
+ values[i] = int32(i*i + 7)
+ }
+ enc := encoding.NewEncoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, false, column,
memory.DefaultAllocator).(encoding.Int32Encoder)
+ defer enc.Release()
+
+ enc.Put(values)
+ flush(enc)
+
+ enc.PutSpaced(allNull, []byte{0}, 0)
+ assert.Zero(t, enc.EstimatedDataEncodedSize())
+ assert.Equal(t, emptyPage, flush(enc))
+
+ enc.Put(values[:3])
+ page := flush(enc)
+ dec := encoding.NewDecoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, column,
memory.DefaultAllocator).(encoding.Int32Decoder)
+ require.NoError(t, dec.SetData(3, page))
+ out := make([]int32, 3)
+ got, err := dec.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, 3, got)
+ assert.Equal(t, values[:3], out)
+ })
+ }
+}
+
func TestDeltaLengthByteArrayEncoding(t *testing.T) {
column := schema.NewColumn(schema.NewByteArrayNode("bytearray",
parquet.Repetitions.Required, -1), 0, 0)
