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 1683b276 fix(parquet): wrap DELTA_BINARY_PACKED deltas at the physical
type width (#1027)
1683b276 is described below
commit 1683b276c1dbad9d2ec306efae535d89d3d7dcdd
Author: Tom Frank <[email protected]>
AuthorDate: Mon Jul 27 18:56:04 2026 +0300
fix(parquet): wrap DELTA_BINARY_PACKED deltas at the physical type width
(#1027)
### Rationale for this change
The Parquet spec requires DELTA_BINARY_PACKED delta arithmetic to wrap
in two's complement at the physical type's width
([Encodings.md](https://github.com/apache/parquet-format/blob/master/Encodings.md#delta-encoding-delta_binary_packed--5)).
The encoder computed INT32 deltas in `int64`, so wide-range int32
columns (e.g. values spanning most of the int32 range) produced deltas
outside the int32 range and miniblock bit-widths up to 34.
parquet-cpp/pyarrow reject such pages as corrupt (`Unexpected end of
stream: InitBlock EOF`), as does arrow-rs (`'min_delta' is too large`) —
only arrow-go itself could read them back, because its decoder uses the
same widened arithmetic. parquet-cpp explicitly declined to accept
>32-bit INT32 miniblocks (apache/arrow#20374), so the writer must not
emit them.
### What changes are included in this PR?
`deltaBitPackEncoder[T].Put` now computes each delta at the physical
type's width (`int64(in[idx] - T(enc.currentVal))`) so overflow wraps in
two's complement as the spec requires, bounding miniblock bit-widths at
32 for INT32. INT64 encoding is unchanged (the arithmetic was already at
type width). The decoder needs no change: its int64 accumulation with a
final cast to `T` decodes wrapped deltas correctly, and it still reads
previously written out-of-spec files.
### Are these changes tested?
Yes. New `TestDeltaBitPackedDeltaOverflow` covers alternating
`MinInt32`/`MaxInt32`, hand-crafted pathological cases for int32, a
golden-bytes case for wrapped-delta encoding, and an int64 no-regression
case. A new `checkDeltaBitPackedWidths` helper parses the encoded stream
and asserts every miniblock bit-width is within the physical type's
width — round-trip tests alone cannot catch this, since arrow-go
tolerates its own out-of-spec output. The test fails on the previous
encoder and passes with the fix. Also verified end-to-end: a file
written via `pqarrow` with a wide-range int32 DELTA_BINARY_PACKED column
is now read correctly by pyarrow, where it previously failed with
`InitBlock EOF`.
### Are there any user-facing changes?
DELTA_BINARY_PACKED INT32 pages written for wide-value-range columns are
now spec-compliant and readable by parquet-cpp/pyarrow, arrow-rs, and
other strict readers. Files previously written with the out-of-spec
widths remain readable by arrow-go.
---
parquet/internal/encoding/delta_bit_packing.go | 10 +-
parquet/internal/encoding/encoding_test.go | 160 +++++++++++++++++++++++++
2 files changed, 167 insertions(+), 3 deletions(-)
diff --git a/parquet/internal/encoding/delta_bit_packing.go
b/parquet/internal/encoding/delta_bit_packing.go
index b74c27d0..0ee45423 100644
--- a/parquet/internal/encoding/delta_bit_packing.go
+++ b/parquet/internal/encoding/delta_bit_packing.go
@@ -402,9 +402,13 @@ func (enc *deltaBitPackEncoder[T]) Put(in []T) {
enc.totalVals += uint64(len(in))
for ; idx < len(in); idx++ {
- val := int64(in[idx])
- enc.deltas = append(enc.deltas, val-enc.currentVal)
- enc.currentVal = val
+ // compute the delta at the physical type's width so that
overflow wraps in
+ // two's complement, as the spec requires. Widening to int64
first would produce
+ // deltas (and miniblock bit-widths) beyond the type's width
for INT32 columns,
+ // which other readers (parquet-cpp, arrow-rs) reject as
corrupt.
+ delta := int64(in[idx] - T(enc.currentVal))
+ enc.deltas = append(enc.deltas, delta)
+ enc.currentVal = int64(in[idx])
if len(enc.deltas) == int(enc.blockSize) {
enc.flushBlock()
}
diff --git a/parquet/internal/encoding/encoding_test.go
b/parquet/internal/encoding/encoding_test.go
index 3ec9c4d2..8242e56e 100644
--- a/parquet/internal/encoding/encoding_test.go
+++ b/parquet/internal/encoding/encoding_test.go
@@ -18,9 +18,11 @@ package encoding_test
import (
"bufio"
+ "bytes"
"encoding/binary"
"fmt"
"io"
+ "math"
"os"
"path"
"reflect"
@@ -34,6 +36,7 @@ import (
"github.com/apache/arrow-go/v18/parquet"
"github.com/apache/arrow-go/v18/parquet/internal/encoding"
"github.com/apache/arrow-go/v18/parquet/internal/testutils"
+ "github.com/apache/arrow-go/v18/parquet/internal/utils"
"github.com/apache/arrow-go/v18/parquet/schema"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -698,6 +701,163 @@ func TestWriteDeltaBitPackedInt32(t *testing.T) {
})
}
+// checkDeltaBitPackedWidths walks an encoded DELTA_BINARY_PACKED stream and
asserts
+// that no miniblock bit-width exceeds the physical type's width. The spec
requires
+// delta arithmetic to wrap at the type width, so a compliant INT32 stream can
never
+// need more than 32 bits per delta; parquet-cpp and arrow-rs reject wider
miniblocks.
+func checkDeltaBitPackedWidths(t *testing.T, data []byte, maxWidth byte) {
+ rdr := utils.NewBitReader(bytes.NewReader(data))
+
+ blockSize, ok := rdr.GetVlqInt()
+ require.True(t, ok)
+ numMini, ok := rdr.GetVlqInt()
+ require.True(t, ok)
+ totalValues, ok := rdr.GetVlqInt()
+ require.True(t, ok)
+ _, ok = rdr.GetZigZagVlqInt() // first value
+ require.True(t, ok)
+
+ valsPerMini := blockSize / numMini
+ batch := make([]uint64, valsPerMini)
+ remaining := int64(totalValues) - 1
+ for remaining > 0 {
+ _, ok = rdr.GetZigZagVlqInt() // min delta
+ require.True(t, ok)
+
+ widths := make([]byte, numMini)
+ for i := range widths {
+ w, err := rdr.ReadByte()
+ require.NoError(t, err)
+ widths[i] = w
+ }
+
+ for i := uint64(0); i < numMini && remaining > 0; i++ {
+ assert.LessOrEqualf(t, widths[i], maxWidth, "miniblock
%d bit-width exceeds physical type width", i)
+ if widths[i] > 0 {
+ // miniblocks are always padded to full length
+ n, err := rdr.GetBatch(uint(widths[i]), batch)
+ require.NoError(t, err)
+ require.EqualValues(t, valsPerMini, n)
+ }
+ remaining -= min(int64(valsPerMini), remaining)
+ }
+ }
+}
+
+// Deltas between consecutive INT32 values can exceed the int32 range; the spec
+// requires them to wrap at the type width. Encoding them at int64 width
instead
+// produces >32-bit miniblocks that only arrow-go itself can read back.
+func TestDeltaBitPackedDeltaOverflow(t *testing.T) {
+ t.Run("int32 alternating min max", func(t *testing.T) {
+ column := schema.NewColumn(schema.NewInt32Node("int32",
parquet.Repetitions.Required, -1), 0, 0)
+ values := make([]int32, 1024)
+ for i := range values {
+ if i%2 == 0 {
+ values[i] = math.MinInt32
+ } else {
+ values[i] = math.MaxInt32
+ }
+ }
+
+ enc := encoding.NewEncoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, false, column, memory.DefaultAllocator)
+ enc.(encoding.Int32Encoder).Put(values)
+ buf, _ := enc.FlushValues()
+ defer buf.Release()
+
+ checkDeltaBitPackedWidths(t, buf.Bytes(), 32)
+
+ dec := encoding.NewDecoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, column, memory.DefaultAllocator)
+ require.NoError(t,
dec.(encoding.Int32Decoder).SetData(len(values), buf.Bytes()))
+ out := make([]int32, len(values))
+ _, err := dec.(encoding.Int32Decoder).Decode(out)
+ require.NoError(t, err)
+ assert.Equal(t, values, out)
+ })
+
+ t.Run("int32 boundary deltas", func(t *testing.T) {
+ column := schema.NewColumn(schema.NewInt32Node("int32",
parquet.Repetitions.Required, -1), 0, 0)
+ // every boundary of the wrapped 32-bit delta domain; comments
show the
+ // two's-complement-wrapped delta from the previous value
+ values := []int32{
+ 0,
+ math.MinInt32, // -2^31 (most negative wrapped
delta)
+ -1, // +2^31-1 (most positive wrapped
delta); with the previous delta this miniblock spans 2^32-1 -> width 32
+ math.MaxInt32, // true delta +2^31 wraps to -2^31
+ -2, // true delta -(2^31+1) wraps to
+2^31-1
+ math.MaxInt32 - 1, // true delta +2^31 wraps to -2^31
+ math.MaxInt32 - 1, // 0 (zero delta alongside extreme
deltas)
+ math.MinInt32 + 1, // true delta -(2^32-3) wraps to +3
+ 0, // +2^31-1
+ 1, // +1
+ math.MinInt32, // true delta -(2^31+1) wraps to
+2^31-1
+ math.MaxInt32, // true delta +(2^32-1) wraps to -1
+ math.MinInt32, // true delta -(2^32-1) wraps to +1
+ }
+
+ enc := encoding.NewEncoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, false, column, memory.DefaultAllocator)
+ enc.(encoding.Int32Encoder).Put(values)
+ buf, _ := enc.FlushValues()
+ defer buf.Release()
+
+ checkDeltaBitPackedWidths(t, buf.Bytes(), 32)
+
+ dec := encoding.NewDecoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, column, memory.DefaultAllocator)
+ require.NoError(t,
dec.(encoding.Int32Decoder).SetData(len(values), buf.Bytes()))
+ out := make([]int32, len(values))
+ _, err := dec.(encoding.Int32Decoder).Decode(out)
+ require.NoError(t, err)
+ assert.Equal(t, values, out)
+ })
+
+ t.Run("int32 wrapped delta encoding", func(t *testing.T) {
+ // MaxInt32 -> MinInt32 wraps to a delta of +1, so the block
encodes
+ // min_delta=1 with zero-width miniblocks.
+ column := schema.NewColumn(schema.NewInt32Node("int32",
parquet.Repetitions.Required, -1), 0, 0)
+ values := []int32{math.MaxInt32, math.MinInt32}
+
+ enc := encoding.NewEncoder(parquet.Types.Int32,
parquet.Encodings.DeltaBinaryPacked, false, column, memory.DefaultAllocator)
+ enc.(encoding.Int32Encoder).Put(values)
+ buf, _ := enc.FlushValues()
+ defer buf.Release()
+
+ expected := []byte{
+ 128, 1, // block size 128
+ 4, // 4 miniblocks per block
+ 2, // 2 values
+ 254, 255, 255, 255, 15, // first value zigzag(MaxInt32)
+ 2, // min delta zigzag(+1)
+ 0, 0, 0, 0, // all-zero miniblock widths
+ }
+ assert.Equal(t, expected, buf.Bytes())
+ })
+
+ t.Run("int64 alternating min max", func(t *testing.T) {
+ column := schema.NewColumn(schema.NewInt64Node("int64",
parquet.Repetitions.Required, -1), 0, 0)
+ values := make([]int64, 1024)
+ for i := range values {
+ if i%2 == 0 {
+ values[i] = math.MinInt64
+ } else {
+ values[i] = math.MaxInt64
+ }
+ }
+
+ enc := encoding.NewEncoder(parquet.Types.Int64,
parquet.Encodings.DeltaBinaryPacked, false, column, memory.DefaultAllocator)
+ enc.(encoding.Int64Encoder).Put(values)
+ buf, _ := enc.FlushValues()
+ defer buf.Release()
+
+ checkDeltaBitPackedWidths(t, buf.Bytes(), 64)
+
+ dec := encoding.NewDecoder(parquet.Types.Int64,
parquet.Encodings.DeltaBinaryPacked, column, memory.DefaultAllocator)
+ require.NoError(t,
dec.(encoding.Int64Decoder).SetData(len(values), buf.Bytes()))
+ out := make([]int64, len(values))
+ _, err := dec.(encoding.Int64Decoder).Decode(out)
+ require.NoError(t, err)
+ assert.Equal(t, values, out)
+ })
+}
+
func TestWriteDeltaBitPackedInt64(t *testing.T) {
column := schema.NewColumn(schema.NewInt64Node("int64",
parquet.Repetitions.Required, -1), 0, 0)
