[Go][Parquet] support streaming reads for large data pages

[joechenrh]

Describe the enhancement requested

We have meet a Parquet file with extremely large data pages, making materializing the whole page expensive.

One observed page has metadata similar to:

type: DATA_PAGE
compressedSize: 1079936746
uncompressedSize: 1426870008
numValues: 233
encoding: PLAIN
definitionLevelEncoding: RLE
repetitionLevelEncoding: BIT_PACKED
hasCrc: true

Today the Parquet reader materializes the full uncompressed page buffer before decoding values, which requires allocating memory proportional to the page's uncompressed size.

It would be useful for the reader to support a streaming/lazy data page read path for large pages. For Data Page V1, the reader could parse repetition and definition levels first into small buffers, then leave the decompressed stream positioned at the encoded values and decode values on demand.

A possible first step would be an opt-in reader property or strategy that enables streaming value decoding for supported page shapes, for example:

• Data Page V1
• PLAIN encoding
• BYTE_ARRAY / string-like columns
• RLE definition levels
• RLE or BIT_PACKED repetition levels
• codecs that can be exposed as an io.Reader stream

Unsupported encodings/codecs/page layouts could continue to fall back to the existing materialized page path.

The goal is to allow applications to read valid Parquet files with very large pages while keeping peak reader memory closer to the requested batch size plus small level buffers.

[zeroshade]

This is an interesting idea! I like it in theory, it will just be tricky to implement properly. Are you interested in giving implementing this a shot yourself?

[joechenrh]

Yes, I'm still trying to figure out a clean way to implement this. 😸

[zeroshade]

Sounds good! Looking forward to reviewing it. Feel free to use this issue for discussion if you need to bounce ideas off me or otherwise discuss options

[joechenrh]

A sketch of the approach — would appreciate a sanity check on the shape before I build it out.

Scope: Data Page V1 + V2, PLAIN encoding, types int32/int64/float/double/FLBA/byte_array. Codecs: UNCOMPRESSED/GZIP/BROTLI/ZSTD only, everything else falls back to the current path. Opt-in via a new ReaderProperties.EnablePageStreaming (default off, no behavior change).

Core idea

Currently the decoder reads from a fully-decompressed []byte slice.

arrow-go/parquet/internal/encoding/types.go

Line 37 in d17f687

SetData(buffered int, buf []byte) error

arrow-go/parquet/file/page_reader.go

Line 501 in d17f687

func (p *serializedPageReader) decompress(rd io.Reader, lenCompressed int, buf []byte) ([]byte, error) {

Here we introduce a ValueBuffer between page and decoder.

type ValueBuffer interface {
    Bytes() []byte                           // currently buffered; valid until next Fill
    Fill(consumed, need int) ([]byte, error) // drop consumed, ensure >= need contiguous (grows for oversized values)
    Stable() bool                            // true => caller may zero-copy alias (buffer never slides)
}

For the existing path it's just a wrapper over the decompressed buffer, for the streaming path it stores codec.NewReader.

TypedDecoder.SetData([]byte) stays unchanged, so the other ~9 decoder implementations (boolean, DELTA_*, byte-stream-split, dictionary) are untouched. Only the three in-scope PLAIN decoders — PlainDecoder[T], PlainByteArrayDecoder, PlainFixedLenByteArrayDecoder — read through ValueBuffer and additionally implement an optional StreamingDecoder { SetSource(ValueBuffer) }. For those three the materialized path is the degenerate "never refill" case, so existing behavior is unchanged; streaming copies each value out, since a slice into a sliding buffer can't be aliased across a batch.

Pages: Add two functions to the DataPage interface:

type DataPage interface {
    ...
    LevelData() []byte                               // decompressed bytes for the level decoders
    ValueBuffer(levelBytes int) encoding.ValueBuffer // source for the value decoder
}

For the streaming path we decompress the level region in advance in serializedPageReader.Next(). For V1, rep&def region need to be peeled from the decompressed stream, we need the same logic LevelDecoder.SetData uses.

func (p *serializedPageReader) readLevelData(r io.Reader, repEnc, defEnc format.Encoding, maxRep, maxDef int16, numValues int) ([]byte, error)