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 3f734529 fix(parquet/encoding): read ahead in the streaming value
buffer (#937)
3f734529 is described below
commit 3f734529ae7366b258e23faa330a23dee4cc0a59
Author: Ruihao Chen <[email protected]>
AuthorDate: Fri Aug 7 23:44:10 2026 +0800
fix(parquet/encoding): read ahead in the streaming value buffer (#937)
## Summary
The streaming value buffer's `Fill` read only `need` bytes per call, so
the decoders call `Fill`(`Read`) per value. And each call just copies a
few bytes out of an already-decoded block. This makes streaming decoding
**1.2–1.9× slower** than the materialized path for small values.
Now `Fill` reads toward the end of the current chunk — so the reader is
called once per chunk. And `Recycle` moves the unconsumed data to the
front once the consumed data exceeds half the buffer, instead of
allocating a fresh chunk on rotate.
## Correction
I need to correct — and apologize for — something I wrote while landing
the streaming decode in #880:
> I prototyped a read-ahead buffer to close the zstd gap and measured ≈0
improvement (the cost is the extra copy, not the read count), so I
dropped it.
That conclusion was wrong, and worse, it reverted something that had
already been working. An earlier iteration of the value buffer *did*
read ahead correctly; but dropped in the later refactor. The
re-prototype behind the "≈0 improvement" claim was then **broken**. From
testing, this adds a **1.2–1.9× small-value decode regression** for
zstd.
Sorry for removing the code that works and then drawing the wrong
conclusion. This PR restores read-ahead with some changes and adds a
test for that path.
## Results
Decode-only, `ReadBatchInPage`; single ByteArray column, 32 MiB pages,
arrow-go's default `BufferSize` (16 KiB), median of 3.
Ratio = materialized / streaming, so **>1 means streaming is slower**:
| value size | zstd before | zstd after | gzip before | gzip after |
| ---------- | ----------- | ---------- | ----------- | ---------- |
| 9 B | 1.85× | **1.16×** | 1.78× | **1.33×** |
| 33 B | 1.66× | **1.15×** | 1.58× | **1.43×** |
| 129 B | 1.23× | **1.04×** | 0.91× | **0.65×** |
`ReadBatch` gives almost the same ratios.
The remaining ~1.04–1.16× (zstd) is the streaming path's inherent
overhead: the codec pulls its compressed input incrementally, so it
issues many more reads than the materialized path's single bulk read —
profiling attributes most of the gap to those read syscalls — plus a
`Fill`/`Advance` call per value. Read-ahead removes what the
*value-side* read count added; it does not remove those.
### gzip
gzip benefits too, but its residual is higher (~1.33× at 9 B vs zstd's
~1.16×). The extra gap is in flate's input reading. Both paths read the
compressed input byte by byte; the difference is **inlining**:
- The materialized path gives flate a `*bytes.Reader`, whose `ReadByte`
inlines, so flate keeps its bit-buffer state in registers across the
read loop (`huffmanBytesReader`).
- Streaming gives flate a plain reader, so `ReadByte` is a real call per
byte and the state spills around each call (`huffmanBufioReader`) —
~1.85× slower in that loop.
Feeding flate a `*bytes.Reader` under streaming recovers it (~1.06×),
but that means holding the whole compressed page in memory, which gives
up the streaming memory bound.
---------
Co-authored-by: Claude Opus 4.8 (1M context) <[email protected]>
---
.../internal/encoding/streaming/value_buffer.go | 27 ++++++---
.../streaming/value_buffer_internal_test.go | 70 ++++++++++++++++++++--
2 files changed, 86 insertions(+), 11 deletions(-)
diff --git a/parquet/internal/encoding/streaming/value_buffer.go
b/parquet/internal/encoding/streaming/value_buffer.go
index 4bfc7875..4b00d2f8 100644
--- a/parquet/internal/encoding/streaming/value_buffer.go
+++ b/parquet/internal/encoding/streaming/value_buffer.go
@@ -52,8 +52,8 @@ type streamBuffer struct {
onClose func() error
mem memory.Allocator
chunkSize int
- cur []byte // chunk currently being filled (off == n on Fill
entry)
- off, n int // consumed cursor / filled end within cur
+ cur []byte // chunk currently being filled
+ off, n int // consumed cursor / filled end; cur[off:n] is
buffered read-ahead
live [][]byte // earlier chunks still backing this Decode's aliases
remaining int // value-region bytes not yet consumed; bounds Fill
vs corrupt lengths
}
@@ -118,7 +118,9 @@ func (s *streamBuffer) Fill(need int) ([]byte, error) {
s.rotate(need)
}
for s.n-s.off < need {
- m, err := s.r.Read(s.cur[s.n : s.off+need])
+ // Read ahead to the end of cur, bounded by the value region.
+ end := min(len(s.cur), s.off+s.remaining)
+ m, err := s.r.Read(s.cur[s.n:end])
s.n += m
if err != nil {
if err == io.EOF {
@@ -151,12 +153,23 @@ func (s *streamBuffer) Recycle() {
s.mem.Free(b)
}
s.live = s.live[:0]
- // Rewind the primary chunk; swap out an oversized one so steady state
stays small.
- if len(s.cur) != s.chunkSize {
+ // Compact once the consumed prefix passes half the chunk to preserve
read-ahead
+ // room while keeping compactions and rotations rare.
+ if s.off*2 <= len(s.cur) {
+ return
+ }
+ tail := s.n - s.off
+ if len(s.cur) != s.chunkSize && tail <= s.chunkSize {
+ // shrink the oversized chunk back to steady state
+ nc := s.mem.Allocate(s.chunkSize)
+ copy(nc, s.cur[s.off:s.n])
s.mem.Free(s.cur)
- s.cur = s.mem.Allocate(s.chunkSize)
+ s.cur = nc
+ } else {
+ // compact in place
+ copy(s.cur, s.cur[s.off:s.n])
}
- s.off, s.n = 0, 0
+ s.off, s.n = 0, tail
}
func (s *streamBuffer) Close() error {
diff --git a/parquet/internal/encoding/streaming/value_buffer_internal_test.go
b/parquet/internal/encoding/streaming/value_buffer_internal_test.go
index b10d54a0..e2047dd2 100644
--- a/parquet/internal/encoding/streaming/value_buffer_internal_test.go
+++ b/parquet/internal/encoding/streaming/value_buffer_internal_test.go
@@ -79,12 +79,15 @@ func TestStreamBufferOversizedValue(t *testing.T) {
// TestStreamBufferSkip checks Skip discards both buffered and not-yet-read
bytes.
func TestStreamBufferSkip(t *testing.T) {
- data := []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+ data := make([]byte, 16)
+ for i := range data {
+ data[i] = byte(i)
+ }
s := newTestBuffer(memory.DefaultAllocator, data, 8)
- assert.Equal(t, []byte{0, 1}, fillValue(t, s, 2))
- require.NoError(t, s.Skip(6)) // 2 buffered + 4 straight from the reader
- assert.Equal(t, []byte{8, 9}, fillValue(t, s, 2))
+ assert.Equal(t, []byte{0, 1}, fillValue(t, s, 2)) // read-ahead buffers
cur[0:8]
+ require.NoError(t, s.Skip(10)) // 6 buffered + 4
discarded from the reader
+ assert.Equal(t, []byte{12, 13}, fillValue(t, s, 2))
require.NoError(t, s.Close())
}
@@ -134,3 +137,62 @@ func TestStreamBufferRotateCarriesTail(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, data[2:9], b[:7])
}
+
+func TestStreamBufferReadsAhead(t *testing.T) {
+ s := newTestBuffer(memory.DefaultAllocator, make([]byte, 100), 16)
+ defer s.Close()
+
+ _, err := s.Fill(4)
+ require.NoError(t, err)
+ assert.Equal(t, 16, s.n, "Fill should read ahead to fill the chunk")
+}
+
+func TestStreamBufferReadAheadSurvivesRecycle(t *testing.T) {
+ data := make([]byte, 100)
+ for i := range data {
+ data[i] = byte(i)
+ }
+ s := newTestBuffer(memory.DefaultAllocator, data, 16)
+ defer s.Close()
+
+ var got []byte
+ for len(got) < len(data) {
+ s.Recycle()
+ got = append(got, fillValue(t, s, 4)...)
+ }
+ assert.Equal(t, data, got)
+}
+
+func TestStreamBufferReadAheadStopsAtRegion(t *testing.T) {
+ // the value region is the first 8 bytes; the rest stands in for the
next page
+ data := []byte{0, 1, 2, 3, 4, 5, 6, 7, 100, 101, 102}
+ s := &streamBuffer{mem: memory.DefaultAllocator, r:
bytes.NewReader(data), chunkSize: 16, cur:
memory.DefaultAllocator.Allocate(16), remaining: 8}
+ defer s.Close()
+
+ for range 8 {
+ fillValue(t, s, 1)
+ }
+ rest, _ := io.ReadAll(s.r)
+ assert.Equal(t, []byte{100, 101, 102}, rest, "read-ahead over-read past
the value region")
+}
+
+func TestStreamBufferMixedValueSizes(t *testing.T) {
+ lengths := []int{1, 50, 2, 3, 100, 1, 7, 30, 2, 16, 1, 64, 4, 9, 40, 1,
5, 25, 8, 33}
+ var data []byte
+ for k, n := range lengths {
+ for j := range n {
+ data = append(data, byte(k*7+j))
+ }
+ }
+ s := newTestBuffer(memory.DefaultAllocator, data, 16)
+ defer s.Close()
+
+ off := 0
+ for k, n := range lengths {
+ if k%2 == 0 {
+ s.Recycle()
+ }
+ assert.Equal(t, data[off:off+n], fillValue(t, s, n), "value %d
(len %d)", k, n)
+ off += n
+ }
+}
