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 c23659c8 fix(parquet): stop BYTE_STREAM_SPLIT FLBA spaced decode
aliasing output (#1256)
c23659c8 is described below
commit c23659c8f0e4284acce0a978820abb4d9777cad8
Author: Matt Topol <[email protected]>
AuthorDate: Thu Sep 3 11:28:26 2026 -0400
fix(parquet): stop BYTE_STREAM_SPLIT FLBA spaced decode aliasing output
(#1256)
### Rationale for this change
Fixes #1255.
A `BYTE_STREAM_SPLIT` `FIXED_LEN_BYTE_ARRAY` column that contains nulls
**and** spans more than one data page decodes values shifted by one
position — silently, with no error returned.
`spacedExpand` moves values into their spaced positions with `copy` and
deliberately does not clean up the null slots:
```go
// because we technically don't care what is in the null slots we don't
actually have to clean
// up after ourselves ... Any data that happens to be left in the null
slots is fine
```
That reasoning holds for the scalar column types, whose buffers hold
values. But `ByteArray` / `FixedLenByteArray` buffers hold **slice
headers**, so `copy` leaves *duplicate headers* behind: after expansion
a null slot and a valid slot can reference the same backing array.
That is still harmless for a decoder that *replaces* the header, which
is what `PlainFixedLenByteArrayDecoder` does:
```go
out[idx] = pflba.data[:pflba.typeLen] // replaces
```
`ByteStreamSplitFixedLenByteArrayDecoder` instead writes **through** the
caller's existing slice:
```go
out[idx] = out[idx][:dec.typeLen] // reuses caller storage
...
out[element][stream] = data[encLoc] // writes through it
```
So once `flbaRecordReader` reused its value buffer for the next page,
two output slots shared one backing array and clobbered each other. This
explains the full shape of the bug: BYTE_STREAM_SPLIT only, nulls
required (to create the duplicates), and two or more pages required (the
first page creates the aliases, the second decodes into them).
### What changes are included in this PR?
- Add `spacedExpandSwap`, which swaps instead of copying so the buffer
remains a permutation of its original elements — no slot aliases
another, and every slot keeps its reusable capacity.
- Use it from `ByteStreamSplitFixedLenByteArrayDecoder.DecodeSpaced`.
- Leave `spacedExpand` itself untouched, so every other column type and
the non-spaced path are unaffected.
- Add a decoder-level regression test (two pages through one reused
buffer, widths 2/3/4/7/8/16) and a randomized differential test
asserting `spacedExpandSwap` places values in exactly the same slots as
`spacedExpand` while never leaving duplicates.
- Add a `pqarrow` round-trip test over a multi-page, nullable BSS FLBA
column — the integration-level case that was returning wrong data.
The decoder-level regression test fails at every width without the fix.
### Why swap rather than the simpler alternatives?
I measured two other approaches and rejected both:
**Making `spacedExpand` itself swap** is correct but replaces `memmove`
with element-wise swaps for *every* column type, which is far too
expensive on sparse-null runs:
```
SpacedExpandInt64/n65536/nullEvery0 4.457µ -> 81.265µ +1723%
SpacedExpandFLBA/n65536/nullEvery0 3.969µ -> 173.959µ +4283%
```
**`clear(out[:toRead])` in `DecodeSpaced`** is a one-liner, but discards
all reusable headers. On 4096 slots / 3511 values / width 16 that is
3511 allocations per `DecodeSpaced` on `main` today (it would drop to 1
once #1172 lands, but the fix should not depend on that).
Swapping keeps the reuse. Cost on the affected path only
(`BenchmarkBSSFLBADecodeSpaced`, 8192 slots):
```
w4/nullEvery100 41.64µ -> 60.55µ +45%
w4/nullEvery7 49.45µ -> 71.95µ +46%
w4/nullEvery2 95.83µ -> 89.59µ -7%
w16/nullEvery100 213.9µ -> 385.2µ +80%
w16/nullEvery7 244.7µ -> 251.3µ ~
w16/nullEvery2 154.3µ -> 160.4µ ~
geomean +25%
```
Steady-state allocations stay at 1–2 per call. A ~25% geomean cost on a
path that is currently returning **incorrect data** seemed clearly worth
it, and nothing outside BSS FLBA spaced decoding is touched.
### Are these changes tested?
Yes — new tests described above. `parquet/...` passes in full with
`PARQUET_TEST_DATA` supplied, and `-race` is clean on
`parquet/internal/encoding`.
### Are there any user-facing changes?
Yes: BYTE_STREAM_SPLIT FIXED_LEN_BYTE_ARRAY columns with nulls spanning
multiple data pages now decode correctly. Previously affected reads
returned silently incorrect values.
**This PR contains a "Critical Fix".** Reading an affected file produced
wrong values with no error, which could have been persisted or acted on
downstream without any indication of a problem.
---
parquet/internal/encoding/byte_stream_split.go | 7 +-
.../encoding/byte_stream_split_decode_test.go | 212 +++++++++++++++++++++
parquet/internal/encoding/decoder.go | 39 ++++
.../encoding/fixed_len_byte_array_decoder.go | 103 +++++++---
.../encoding/fixed_len_byte_array_decoder_test.go | 115 ++++++++++-
.../pqarrow/byte_stream_split_multipage_test.go | 113 +++++++++++
6 files changed, 551 insertions(+), 38 deletions(-)
diff --git a/parquet/internal/encoding/byte_stream_split.go
b/parquet/internal/encoding/byte_stream_split.go
index d5820093..62434aa2 100644
--- a/parquet/internal/encoding/byte_stream_split.go
+++ b/parquet/internal/encoding/byte_stream_split.go
@@ -107,15 +107,16 @@ func releaseBufferToPool(pooled *PooledBufferWriter) {
}
func validateByteStreamSplitPageData(typeLen, nvals int, data []byte) (int,
error) {
- if nvals*typeLen < len(data) {
+ encodedNvals, remainder := len(data)/typeLen, len(data)%typeLen
+ if encodedNvals > nvals || (encodedNvals == nvals && remainder != 0) {
return 0, fmt.Errorf("data size (%d) is too small for the
number of values in in BYTE_STREAM_SPLIT (%d)", len(data), nvals)
}
- if len(data)%typeLen != 0 {
+ if remainder != 0 {
return 0, fmt.Errorf("ByteStreamSplit data size %d not aligned
with byte_width: %d", len(data), typeLen)
}
- return len(data) / typeLen, nil
+ return encodedNvals, nil
}
type byteStreamSplitEncoder[T int32 | int64 | float32 | float64] struct {
diff --git a/parquet/internal/encoding/byte_stream_split_decode_test.go
b/parquet/internal/encoding/byte_stream_split_decode_test.go
index e6c72bdc..f1d8cfb2 100644
--- a/parquet/internal/encoding/byte_stream_split_decode_test.go
+++ b/parquet/internal/encoding/byte_stream_split_decode_test.go
@@ -19,11 +19,16 @@ package encoding
import (
"bytes"
"fmt"
+ "math/rand"
"testing"
"unsafe"
+ "github.com/apache/arrow-go/v18/arrow/bitutil"
+ "github.com/apache/arrow-go/v18/arrow/memory"
"github.com/apache/arrow-go/v18/internal/utils"
"github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/schema"
+ "github.com/stretchr/testify/require"
)
func TestDecodeByteStreamSplitWidth4(t *testing.T) {
@@ -393,3 +398,210 @@ func BenchmarkDecodeByteStreamSplitBatchFLBAWidth8(b
*testing.B) {
})
}
}
+
+// TestByteStreamSplitFLBADecodeSpacedReusedBuffer guards the aliasing bug
where decoding
+// a second page into a buffer previously expanded by DecodeSpaced silently
corrupted
+// values: spacedExpand moves slice headers with copy, leaving duplicate
headers behind in
+// the null slots, and because this decoder writes through the caller's slices
rather than
+// replacing them, two output slots shared one backing array and clobbered
each other.
+func TestByteStreamSplitFLBADecodeSpacedReusedBuffer(t *testing.T) {
+ for _, width := range []int{2, 3, 4, 7, 8, 16} {
+ t.Run(fmt.Sprintf("width=%d", width), func(t *testing.T) {
+ // 5 slots, 2 nulls: slots 0, 2 and 4 are valid.
+ validBits := []byte{0b00010101}
+ const nullCount = 2
+
+ col :=
schema.NewColumn(schema.NewFixedLenByteArrayNode("v",
parquet.Repetitions.Optional, int32(width), -1), 1, 0)
+ dec := NewDecoder(parquet.Types.FixedLenByteArray,
parquet.Encodings.ByteStreamSplit,
+ col,
memory.DefaultAllocator).(FixedLenByteArrayDecoder)
+
+ // A single output buffer reused across both pages, as
the record reader does.
+ out := make([]parquet.FixedLenByteArray, 5)
+
+ for page, offset := range []byte{0, 100} {
+ values := make([]parquet.FixedLenByteArray, 3)
+ for i := range values {
+ values[i] =
make(parquet.FixedLenByteArray, width)
+ for j := range values[i] {
+ values[i][j] = offset +
byte(i*width+j)
+ }
+ }
+
+ data := make([]byte, len(values)*width)
+ for vi, v := range values {
+ for bi, b := range v {
+ data[bi*len(values)+vi] = b
+ }
+ }
+
+ require.NoError(t, dec.SetData(len(values),
data))
+ n, err := dec.DecodeSpaced(out, nullCount,
validBits, 0)
+ require.NoError(t, err)
+ require.Equal(t, len(out), n)
+
+ require.Equal(t, values[0], out[0], "page %d
slot 0", page)
+ require.Equal(t, values[1], out[2], "page %d
slot 2", page)
+ require.Equal(t, values[2], out[4], "page %d
slot 4", page)
+ }
+ })
+ }
+}
+
+// TestSpacedExpandSwapMatchesSpacedExpand checks that swapping places values
in exactly
+// the same slots as copying, and additionally never leaves duplicate entries
behind.
+func TestSpacedExpandSwapMatchesSpacedExpand(t *testing.T) {
+ rng := rand.New(rand.NewSource(42))
+ for iter := 0; iter < 5000; iter++ {
+ n := 1 + rng.Intn(200)
+ validBits := make([]byte, bitutil.BytesForBits(int64(n)))
+ nullCount, density := 0, rng.Float64()
+ for i := 0; i < n; i++ {
+ if rng.Float64() < density {
+ bitutil.ClearBit(validBits, i)
+ nullCount++
+ } else {
+ bitutil.SetBit(validBits, i)
+ }
+ }
+
+ // distinct sentinels so duplicates are detectable
+ copied, swapped := make([]int64, n), make([]int64, n)
+ for i := range copied {
+ copied[i], swapped[i] = int64(i+1), int64(i+1)
+ }
+
+ spacedExpand(copied, nullCount, validBits, 0)
+ spacedExpandSwap(swapped, nullCount, validBits, 0)
+
+ for i := 0; i < n; i++ {
+ if bitutil.BitIsSet(validBits, i) {
+ require.Equalf(t, copied[i], swapped[i],
+ "iter %d n=%d nulls=%d: valid slot %d
differs", iter, n, nullCount, i)
+ }
+ }
+
+ seen := make(map[int64]struct{}, n)
+ for _, v := range swapped {
+ seen[v] = struct{}{}
+ }
+ require.Lenf(t, seen, n,
+ "iter %d n=%d nulls=%d: swap left duplicate entries",
iter, n, nullCount)
+ }
+}
+
+// bssEncodeFLBA lays out values in BYTE_STREAM_SPLIT order: all byte 0s, then
all
+// byte 1s, and so on.
+func bssEncodeFLBA(values []parquet.FixedLenByteArray, width int) []byte {
+ data := make([]byte, len(values)*width)
+ for vi, v := range values {
+ for bi, b := range v {
+ data[bi*len(values)+vi] = b
+ }
+ }
+ return data
+}
+
+// TestByteStreamSplitFLBADoesNotWriteThroughForeignBuffers guards the second
half of the
+// aliasing hazard reported on GH-1255: this decoder reuses any output slice
with enough
+// capacity and writes through it, so if the previous page left slices that
point at
+// memory the decoder does not own, decoding corrupts that memory.
+//
+// Two producers leave such slices behind, and neither needs nulls to do it:
+//
+// - RLE_DICTIONARY assigns dict[idx] into every slot for that index, so a
repeated
+// index leaves several slots aliasing one dictionary-backed slice.
Writing through
+// them both clobbers a decoded value and corrupts the dictionary itself.
+// - PLAIN slices the page buffer directly, so every slot points into the
previous
+// page's data.
+func TestByteStreamSplitFLBADoesNotWriteThroughForeignBuffers(t *testing.T) {
+ for _, width := range []int{2, 3, 4, 7, 8, 16} {
+ t.Run(fmt.Sprintf("width=%d", width), func(t *testing.T) {
+ col :=
schema.NewColumn(schema.NewFixedLenByteArrayNode("v",
parquet.Repetitions.Required, int32(width), -1), 0, 0)
+
+ // The BSS page: three distinct values, decoded into a
reused buffer.
+ bssValues := make([]parquet.FixedLenByteArray, 3)
+ for i := range bssValues {
+ bssValues[i] = make(parquet.FixedLenByteArray,
width)
+ for j := range bssValues[i] {
+ bssValues[i][j] = 100 + byte(i*width+j)
+ }
+ }
+ bssData := bssEncodeFLBA(bssValues, width)
+
+ t.Run("after RLE_DICTIONARY", func(t *testing.T) {
+ // Dictionary of two entries, referenced by
indices [0, 0, 1] so that
+ // slots 0 and 1 alias the same
dictionary-backed slice.
+ dictBuf := make([]byte, 2*width)
+ for i := range dictBuf {
+ dictBuf[i] = byte(i + 1)
+ }
+ // RLE_DICTIONARY index data: bit width byte,
then a bit-packed run.
+ idxBuf := []byte{1, 0b00000011, 0b00000100}
+
+ plainDict :=
NewDecoder(parquet.Types.FixedLenByteArray, parquet.Encodings.Plain,
+ col, memory.DefaultAllocator)
+ require.NoError(t, plainDict.SetData(2,
dictBuf))
+
+ dictDec :=
NewDictDecoder(parquet.Types.FixedLenByteArray, col,
memory.DefaultAllocator).(*DictFixedLenByteArrayDecoder)
+ dictDec.SetDict(plainDict)
+ require.NoError(t, dictDec.SetData(3, idxBuf))
+
+ out := make([]parquet.FixedLenByteArray, 3)
+ n, err := dictDec.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, 3, n)
+
+ dictBefore := bytes.Clone(dictBuf)
+
+ bssDec :=
NewDecoder(parquet.Types.FixedLenByteArray, parquet.Encodings.ByteStreamSplit,
+ col,
memory.DefaultAllocator).(FixedLenByteArrayDecoder)
+ require.NoError(t,
bssDec.SetData(len(bssValues), bssData))
+ n, err = bssDec.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, len(bssValues), n)
+
+ for i, want := range bssValues {
+ require.Equalf(t, want, out[i], "slot
%d decoded incorrectly", i)
+ }
+ require.Equal(t, dictBefore, dictBuf, "decoding
wrote through into the dictionary page buffer")
+ })
+
+ t.Run("after PLAIN", func(t *testing.T) {
+ plainValues :=
make([]parquet.FixedLenByteArray, 3)
+ for i := range plainValues {
+ plainValues[i] =
make(parquet.FixedLenByteArray, width)
+ for j := range plainValues[i] {
+ plainValues[i][j] =
byte(i*width + j)
+ }
+ }
+ plainData := make([]byte, 0,
len(plainValues)*width)
+ for _, v := range plainValues {
+ plainData = append(plainData, v...)
+ }
+
+ plainDec :=
NewDecoder(parquet.Types.FixedLenByteArray, parquet.Encodings.Plain,
+ col,
memory.DefaultAllocator).(FixedLenByteArrayDecoder)
+ require.NoError(t,
plainDec.SetData(len(plainValues), plainData))
+
+ out := make([]parquet.FixedLenByteArray, 3)
+ n, err := plainDec.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, len(plainValues), n)
+
+ plainBefore := bytes.Clone(plainData)
+
+ bssDec :=
NewDecoder(parquet.Types.FixedLenByteArray, parquet.Encodings.ByteStreamSplit,
+ col,
memory.DefaultAllocator).(FixedLenByteArrayDecoder)
+ require.NoError(t,
bssDec.SetData(len(bssValues), bssData))
+ n, err = bssDec.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, len(bssValues), n)
+
+ for i, want := range bssValues {
+ require.Equalf(t, want, out[i], "slot
%d decoded incorrectly", i)
+ }
+ require.Equal(t, plainBefore, plainData,
"decoding wrote through into the PLAIN page buffer")
+ })
+ })
+ }
+}
diff --git a/parquet/internal/encoding/decoder.go
b/parquet/internal/encoding/decoder.go
index e97b5a13..a6f7931d 100644
--- a/parquet/internal/encoding/decoder.go
+++ b/parquet/internal/encoding/decoder.go
@@ -229,6 +229,45 @@ func (d *dictDecoder[T]) DecodeIndicesSpaced(numValues,
nullCount int, validBits
return n, nil
}
+// spacedExpandSwap is spacedExpand for reusable slices whose entries may
later be
+// written through. spacedExpand moves values with copy and leaves the null
slots alone,
+// which for the slice-header column types (ByteArray / FixedLenByteArray)
leaves
+// duplicate headers behind: a null slot and a valid slot end up referencing
the same
+// backing array. A later decoder that writes through those entries would then
clobber
+// one value when the buffer is reused for another page.
+//
+// Swapping instead of copying keeps the buffer a permutation of its original
elements,
+// so no slot aliases another and every slot keeps its reusable capacity.
+func spacedExpandSwap[T parquet.ColumnTypes](buffer []T, nullCount int,
validBits []byte, validBitsOffset int64) int {
+ numValues := len(buffer)
+
+ idxDecode := int64(numValues - nullCount)
+ if idxDecode == 0 {
+ return numValues
+ }
+
+ rdr := bitutils.NewReverseSetBitRunReader(validBits, validBitsOffset,
int64(numValues))
+ for {
+ run := rdr.NextRun()
+ if run.Length == 0 {
+ break
+ }
+
+ idxDecode -= run.Length
+ // Once the decoded prefix is already aligned every remaining
swap is a
+ // self-swap, so there is nothing left to do. Mirrors
spacedExpand.
+ if idxDecode == run.Pos {
+ return numValues
+ }
+ for k := run.Length - 1; k >= 0; k-- {
+ dst, src := run.Pos+k, idxDecode+k
+ buffer[dst], buffer[src] = buffer[src], buffer[dst]
+ }
+ }
+
+ return numValues
+}
+
// spacedExpand is used to take a slice of data and utilize the bitmap
provided to fill in nulls into the
// correct slots according to the bitmap in order to produce a fully expanded
result slice with nulls
// in the correct slots.
diff --git a/parquet/internal/encoding/fixed_len_byte_array_decoder.go
b/parquet/internal/encoding/fixed_len_byte_array_decoder.go
index 080e9a06..e18223ae 100644
--- a/parquet/internal/encoding/fixed_len_byte_array_decoder.go
+++ b/parquet/internal/encoding/fixed_len_byte_array_decoder.go
@@ -20,6 +20,7 @@ import (
"errors"
"fmt"
"math"
+ "unsafe"
"github.com/apache/arrow-go/v18/internal/utils"
"github.com/apache/arrow-go/v18/parquet"
@@ -125,7 +126,9 @@ func (pflba *PlainFixedLenByteArrayDecoder)
DecodeSpaced(out []parquet.FixedLenB
return valuesRead, errors.New("parquet: number of values /
definitions levels read did not match")
}
- return spacedExpand(out, nullCount, validBits, validBitsOffset), nil
+ // Keep every output slot independent because a later decoder may write
through
+ // these slices when this buffer is reused across pages.
+ return spacedExpandSwap(out, nullCount, validBits, validBitsOffset), nil
}
// ByteStreamSplitFixedLenByteArrayDecoder is a decoder for
BYTE_STREAM_SPLIT-encoded
@@ -133,6 +136,10 @@ func (pflba *PlainFixedLenByteArrayDecoder)
DecodeSpaced(out []parquet.FixedLenB
type ByteStreamSplitFixedLenByteArrayDecoder struct {
decoder
stride int
+
+ // storage is the most recent block this decoder allocated for output
values.
+ // Decode may only write through an output slice backed by this block;
see owns.
+ storage []byte
}
func (dec *ByteStreamSplitFixedLenByteArrayDecoder) Type() parquet.Type {
@@ -140,18 +147,16 @@ func (dec *ByteStreamSplitFixedLenByteArrayDecoder)
Type() parquet.Type {
}
func (dec *ByteStreamSplitFixedLenByteArrayDecoder) SetData(nvals int, data
[]byte) error {
- if nvals*dec.typeLen < len(data) {
- return fmt.Errorf("data size (%d) is too small for the number
of values in in BYTE_STREAM_SPLIT (%d)", len(data), nvals)
+ encodedNvals, err := validateByteStreamSplitPageData(dec.typeLen,
nvals, data)
+ if err != nil {
+ return err
}
-
- if len(data)%dec.typeLen != 0 {
- return fmt.Errorf("ByteStreamSplit data size %d not aligned
with type %s and byte_width: %d", len(data), dec.Type(), dec.typeLen)
+ if dec.descr != nil && dec.descr.MaxDefinitionLevel() == 0 &&
encodedNvals != nvals {
+ return fmt.Errorf("BYTE_STREAM_SPLIT data contains %d values,
expected %d", encodedNvals, nvals)
}
- nvals = len(data) / dec.typeLen
- dec.stride = nvals
-
- return dec.decoder.SetData(nvals, data)
+ dec.stride = encodedNvals
+ return dec.decoder.SetData(encodedNvals, data)
}
func (dec *ByteStreamSplitFixedLenByteArrayDecoder) Discard(n int) (int,
error) {
@@ -174,13 +179,7 @@ func (dec *ByteStreamSplitFixedLenByteArrayDecoder)
Decode(out []parquet.FixedLe
}
out = out[:toRead]
- for idx := range out {
- if cap(out[idx]) < dec.typeLen {
- dec.prepareOutput(out[idx:])
- break
- }
- out[idx] = out[idx][:dec.typeLen]
- }
+ dec.prepareOutput(out)
switch dec.typeLen {
case 2:
@@ -198,29 +197,73 @@ func (dec *ByteStreamSplitFixedLenByteArrayDecoder)
Decode(out []parquet.FixedLe
return toRead, nil
}
-// prepareOutput allocates storage for the entries in out that do not have
enough
-// capacity, while continuing to reuse the entries that do.
-func (dec *ByteStreamSplitFixedLenByteArrayDecoder) prepareOutput(out
[]parquet.FixedLenByteArray) {
- missing := 0
+// owns reports whether every entry of out is backed by the block this decoder
most
+// recently allocated, and so may be written through.
+//
+// The check matters because this decoder decodes in place: it writes bytes
through the
+// slice headers the caller hands it. Reusing a header whose memory belongs to
something
+// else corrupts that memory. Two earlier decoders leave such headers in a
shared value
+// buffer, neither of which requires nulls to do so:
+//
+// - RLE_DICTIONARY assigns dict[idx] into every slot holding that index, so
repeated
+// indices leave several slots aliasing one dictionary-backed slice.
Writing through
+// them clobbers a decoded value and corrupts the dictionary itself.
+// - PLAIN slices the page buffer directly, so every slot points into that
page's data.
+//
+// Capacity alone cannot distinguish these from our own storage, so we compare
against
+// the bounds of the block we allocated.
+func (dec *ByteStreamSplitFixedLenByteArrayDecoder) owns(out
[]parquet.FixedLenByteArray) bool {
+ if len(dec.storage) == 0 {
+ return false
+ }
+
+ base := uintptr(unsafe.Pointer(unsafe.SliceData(dec.storage)))
+ end := base + uintptr(len(dec.storage))
for idx := range out {
if cap(out[idx]) < dec.typeLen {
- missing++
+ return false
+ }
+ p := uintptr(unsafe.Pointer(unsafe.SliceData(out[idx])))
+ if p < base || p+uintptr(dec.typeLen) > end {
+ return false
}
}
+ return true
+}
- storage := make([]byte, missing*dec.typeLen)
- for idx := range out {
- if cap(out[idx]) < dec.typeLen {
- out[idx] = storage[:dec.typeLen:dec.typeLen]
- storage = storage[dec.typeLen:]
- } else {
+// prepareOutput points every entry of out at storage this decoder owns, so
that decoding
+// in place cannot write through into memory belonging to a previous page's
decoder.
+//
+// When out already sits entirely within our own block the headers are reused
as-is and
+// nothing is allocated, which keeps repeated decodes into the same buffer
allocation
+// free. Otherwise a single block is allocated for the whole window. Earlier
windows keep
+// pointing at the blocks they were given, so callers that decode into
successive windows
+// of one buffer (as the record reader does) keep their previously decoded
values.
+func (dec *ByteStreamSplitFixedLenByteArrayDecoder) prepareOutput(out
[]parquet.FixedLenByteArray) {
+ if dec.owns(out) {
+ for idx := range out {
out[idx] = out[idx][:dec.typeLen]
}
+ return
+ }
+
+ storage := make([]byte, len(out)*dec.typeLen)
+ dec.storage = storage
+ for idx := range out {
+ out[idx] = storage[:dec.typeLen:dec.typeLen]
+ storage = storage[dec.typeLen:]
}
}
func (dec *ByteStreamSplitFixedLenByteArrayDecoder) DecodeSpaced(out
[]parquet.FixedLenByteArray, nullCount int, validBits []byte, validBitsOffset
int64) (int, error) {
toRead := len(out) - nullCount
+
+ // Back every slot, the null slots included, before decoding. The
expansion below
+ // permutes headers across the whole window, so preparing only the
decoded prefix
+ // would let headers we do not own migrate into it and force the next
page to
+ // reallocate. Preparing the window once keeps repeated decodes
allocation free.
+ dec.prepareOutput(out)
+
valuesRead, err := dec.Decode(out[:toRead])
if err != nil {
return valuesRead, err
@@ -229,5 +272,7 @@ func (dec *ByteStreamSplitFixedLenByteArrayDecoder)
DecodeSpaced(out []parquet.F
return valuesRead, errors.New("parquet: number of values /
definitions levels read did not match")
}
- return spacedExpand(out, nullCount, validBits, validBitsOffset), nil
+ // This decoder writes through the caller's slices, so it must not
leave aliased
+ // headers behind for the next page; see spacedExpandSwap.
+ return spacedExpandSwap(out, nullCount, validBits, validBitsOffset), nil
}
diff --git a/parquet/internal/encoding/fixed_len_byte_array_decoder_test.go
b/parquet/internal/encoding/fixed_len_byte_array_decoder_test.go
index 732d61e7..8c07a55a 100644
--- a/parquet/internal/encoding/fixed_len_byte_array_decoder_test.go
+++ b/parquet/internal/encoding/fixed_len_byte_array_decoder_test.go
@@ -27,6 +27,37 @@ import (
"github.com/stretchr/testify/require"
)
+func TestByteStreamSplitFixedLenByteArrayDecoderRejectsTruncatedRequiredData(t
*testing.T) {
+ const width = 4
+ node := schema.NewFixedLenByteArrayNode("value",
parquet.Repetitions.Required, width, -1)
+ column := schema.NewColumn(node, 0, 0)
+ decoder := NewDecoder(parquet.Types.FixedLenByteArray,
parquet.Encodings.ByteStreamSplit, column, memory.DefaultAllocator)
+
+ err := decoder.SetData(3, make([]byte, 2*width))
+ require.EqualError(t, err, "BYTE_STREAM_SPLIT data contains 2 values,
expected 3")
+ require.Zero(t, decoder.ValuesLeft())
+}
+
+func TestByteStreamSplitFixedLenByteArrayDecoderAllowsNulls(t *testing.T) {
+ const width = 4
+ values := makeFixedLenByteArrayValues(2, width, 0)
+ node := schema.NewFixedLenByteArrayNode("value",
parquet.Repetitions.Optional, width, -1)
+ column := schema.NewColumn(node, 1, 0)
+ decoder := NewDecoder(parquet.Types.FixedLenByteArray,
parquet.Encodings.ByteStreamSplit, column,
memory.DefaultAllocator).(FixedLenByteArrayDecoder)
+
+ require.NoError(t, decoder.SetData(3,
encodeByteStreamSplitFixedLenByteArray(values, width)))
+ out := make([]parquet.FixedLenByteArray, len(values))
+ decoded, err := decoder.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, len(values), decoded)
+ require.Equal(t, values, out)
+
+ // An optional page can legitimately contain no physical values even
when its
+ // logical count would overflow when multiplied by the byte width.
+ require.NoError(t, decoder.SetData(int(^uint(0)>>1), nil))
+ require.Zero(t, decoder.ValuesLeft())
+}
+
func TestByteStreamSplitFixedLenByteArrayDecoderContiguousOutput(t *testing.T)
{
for _, width := range []int{2, 4, 8, 16, 32} {
t.Run(fmt.Sprintf("width=%d", width), func(t *testing.T) {
@@ -51,15 +82,24 @@ func
TestByteStreamSplitFixedLenByteArrayDecoderContiguousOutput(t *testing.T) {
}
}
-func TestByteStreamSplitFixedLenByteArrayDecoderReusesProvidedOutput(t
*testing.T) {
+//
TestByteStreamSplitFixedLenByteArrayDecoderDoesNotWriteThroughProvidedOutput
pins the
+// narrowed reuse contract introduced for GH-1255.
+//
+// This decoder writes bytes through the slice headers it is handed, so it may
only reuse
+// a header backed by storage it allocated itself. Output buffers are shared
across pages
+// and encodings, and RLE_DICTIONARY and PLAIN both leave headers pointing at
memory they
+// own (dictionary entries and the page buffer respectively). Reusing those on
capacity
+// alone corrupted that memory, so capacity is no longer sufficient to claim a
header.
+func
TestByteStreamSplitFixedLenByteArrayDecoderDoesNotWriteThroughProvidedOutput(t
*testing.T) {
const width = 16
values := makeFixedLenByteArrayValues(4, width, 0)
decoder := newByteStreamSplitFixedLenByteArrayDecoder(t, width, values)
+ caller := make([]byte, width)
out := []parquet.FixedLenByteArray{
make([]byte, 0, width+4),
nil,
- make([]byte, width),
+ caller,
nil,
}
firstPtr := unsafe.Pointer(unsafe.SliceData(out[0]))
@@ -69,10 +109,44 @@ func
TestByteStreamSplitFixedLenByteArrayDecoderReusesProvidedOutput(t *testing.
require.NoError(t, err)
require.Equal(t, len(values), decoded)
require.Equal(t, values, out)
- require.Equal(t, firstPtr, unsafe.Pointer(unsafe.SliceData(out[0])))
- require.Equal(t, thirdPtr, unsafe.Pointer(unsafe.SliceData(out[2])))
- require.Equal(t, uintptr(width),
-
uintptr(unsafe.Pointer(&out[3][0]))-uintptr(unsafe.Pointer(&out[1][0])))
+
+ // The caller's buffers are re-pointed rather than written through, and
are left
+ // untouched.
+ require.NotEqual(t, firstPtr, unsafe.Pointer(unsafe.SliceData(out[0])))
+ require.NotEqual(t, thirdPtr, unsafe.Pointer(unsafe.SliceData(out[2])))
+ require.Equal(t, make([]byte, width), caller, "decoding wrote through a
caller buffer")
+
+ // Every slot now comes from one contiguous block the decoder owns.
+ for idx := 1; idx < len(out); idx++ {
+ previous := uintptr(unsafe.Pointer(&out[idx-1][0]))
+ current := uintptr(unsafe.Pointer(&out[idx][0]))
+ require.Equal(t, uintptr(width), current-previous)
+ }
+}
+
+// TestByteStreamSplitFixedLenByteArrayDecoderReusesOwnStorage checks that the
narrowed
+// contract still keeps repeated decodes into the same buffer allocation free,
which is
+// the case GH-1172 optimized.
+func TestByteStreamSplitFixedLenByteArrayDecoderReusesOwnStorage(t *testing.T)
{
+ const width = 16
+ values := makeFixedLenByteArrayValues(4, width, 0)
+ data := encodeByteStreamSplitFixedLenByteArray(values, width)
+ decoder := newByteStreamSplitFixedLenByteArrayDecoder(t, width, values)
+
+ // First decode hands out the decoder's own block.
+ out := make([]parquet.FixedLenByteArray, len(values))
+ _, err := decoder.Decode(out)
+ require.NoError(t, err)
+ require.Equal(t, values, out)
+
+ // Subsequent decodes into that same window recognize the block as
their own.
+ allocs := testing.AllocsPerRun(100, func() {
+ require.NoError(t, decoder.SetData(len(values), data))
+ _, err := decoder.Decode(out)
+ require.NoError(t, err)
+ })
+ require.Zero(t, allocs)
+ require.Equal(t, values, out)
}
func TestByteStreamSplitFixedLenByteArrayDecoderMixedOutputAllocations(t
*testing.T) {
@@ -154,6 +228,35 @@ func
TestByteStreamSplitFixedLenByteArrayDecoderSpacedOutput(t *testing.T) {
require.Equal(t, uintptr(width), third-second)
}
+func TestFixedLenByteArrayDecoderSpacedOutputAcrossEncodings(t *testing.T) {
+ const width = 4
+ validBits := []byte{0b00010101}
+ firstValues := makeFixedLenByteArrayValues(3, width, 0)
+ secondValues := makeFixedLenByteArrayValues(3, width, 100)
+
+ node := schema.NewFixedLenByteArrayNode("value",
parquet.Repetitions.Required, width, -1)
+ column := schema.NewColumn(node, 0, 0)
+ plain := NewDecoder(parquet.Types.FixedLenByteArray,
parquet.Encodings.Plain, column,
memory.DefaultAllocator).(FixedLenByteArrayDecoder)
+ plainData := make([]byte, 0, len(firstValues)*width)
+ for _, value := range firstValues {
+ plainData = append(plainData, value...)
+ }
+ require.NoError(t, plain.SetData(len(firstValues), plainData))
+
+ out := make([]parquet.FixedLenByteArray, 5)
+ decoded, err := plain.DecodeSpaced(out, 2, validBits, 0)
+ require.NoError(t, err)
+ require.Equal(t, len(out), decoded)
+
+ byteStreamSplit := newByteStreamSplitFixedLenByteArrayDecoder(t, width,
secondValues)
+ decoded, err = byteStreamSplit.DecodeSpaced(out, 2, validBits, 0)
+ require.NoError(t, err)
+ require.Equal(t, len(out), decoded)
+ require.Equal(t, secondValues[0], out[0])
+ require.Equal(t, secondValues[1], out[2])
+ require.Equal(t, secondValues[2], out[4])
+}
+
func newByteStreamSplitFixedLenByteArrayDecoder(t *testing.T, width int,
values []parquet.FixedLenByteArray) FixedLenByteArrayDecoder {
t.Helper()
diff --git a/parquet/pqarrow/byte_stream_split_multipage_test.go
b/parquet/pqarrow/byte_stream_split_multipage_test.go
new file mode 100644
index 00000000..b1886ea7
--- /dev/null
+++ b/parquet/pqarrow/byte_stream_split_multipage_test.go
@@ -0,0 +1,113 @@
+// 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 pqarrow_test
+
+import (
+ "bytes"
+ "context"
+ "fmt"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/arrow"
+ "github.com/apache/arrow-go/v18/arrow/array"
+ "github.com/apache/arrow-go/v18/arrow/memory"
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/apache/arrow-go/v18/parquet/file"
+ "github.com/apache/arrow-go/v18/parquet/pqarrow"
+ "github.com/stretchr/testify/require"
+)
+
+// TestByteStreamSplitFLBANullsMultiPage covers a BYTE_STREAM_SPLIT
FIXED_LEN_BYTE_ARRAY
+// column whose chunk spans several data pages and contains nulls. The record
reader
+// reuses one value buffer across pages, and DecodeSpaced previously left
aliased slice
+// headers in it, so values decoded from the second page onwards came back
shifted.
+func TestByteStreamSplitFLBANullsMultiPage(t *testing.T) {
+ for _, width := range []int{4, 17} {
+ t.Run(fmt.Sprintf("width=%d", width), func(t *testing.T) {
+ const nrows = 5000
+ mem :=
memory.NewCheckedAllocator(memory.DefaultAllocator)
+ defer mem.AssertSize(t, 0)
+
+ dt := &arrow.FixedSizeBinaryType{ByteWidth: width}
+ sc := arrow.NewSchema([]arrow.Field{{Name: "v", Type:
dt, Nullable: true}}, nil)
+
+ bldr := array.NewFixedSizeBinaryBuilder(mem, dt)
+ defer bldr.Release()
+
+ expected := make([][]byte, nrows)
+ for i := range expected {
+ if i%7 == 3 {
+ bldr.AppendNull()
+ continue
+ }
+ v := make([]byte, width)
+ for j := range v {
+ v[j] = byte(i*width + j)
+ }
+ bldr.Append(v)
+ expected[i] = v
+ }
+
+ arr := bldr.NewArray()
+ defer arr.Release()
+ rec := array.NewRecordBatch(sc, []arrow.Array{arr},
nrows)
+ defer rec.Release()
+
+ var buf bytes.Buffer
+ props := parquet.NewWriterProperties(
+ parquet.WithAllocator(mem),
+
parquet.WithEncoding(parquet.Encodings.ByteStreamSplit),
+ parquet.WithDictionaryDefault(false),
+ // small pages so the column chunk spans more
than one data page
+ parquet.WithDataPageSize(512),
+ parquet.WithBatchSize(128),
+ )
+ w, err := pqarrow.NewFileWriter(sc, &buf, props,
pqarrow.DefaultWriterProps())
+ require.NoError(t, err)
+ require.NoError(t, w.Write(rec))
+ require.NoError(t, w.Close())
+
+ rdr, err :=
file.NewParquetReader(bytes.NewReader(buf.Bytes()),
+
file.WithReadProps(parquet.NewReaderProperties(mem)))
+ require.NoError(t, err)
+ defer rdr.Close()
+
+ fr, err := pqarrow.NewFileReader(rdr,
pqarrow.ArrowReadProperties{BatchSize: 137}, mem)
+ require.NoError(t, err)
+ tbl, err := fr.ReadTable(context.Background())
+ require.NoError(t, err)
+ defer tbl.Release()
+
+ require.EqualValues(t, nrows, tbl.NumRows())
+
+ row := 0
+ for _, chunk := range tbl.Column(0).Data().Chunks() {
+ fsb := chunk.(*array.FixedSizeBinary)
+ for i := 0; i < fsb.Len(); i++ {
+ if expected[row] == nil {
+ require.Truef(t, fsb.IsNull(i),
"row %d should be null", row)
+ } else {
+ require.Falsef(t,
fsb.IsNull(i), "row %d should be valid", row)
+ require.Equalf(t,
expected[row], fsb.Value(i), "row %d", row)
+ }
+ row++
+ }
+ }
+ require.Equal(t, nrows, row)
+ })
+ }
+}
