Parquet reader refactor: move to a Streamly-based streaming pipeline (bounded memory, clearer structure, optional concurrency)

[mchav]

Summary

The current Parquet reader in dataframe works, but its control-flow and data-flow are largely “batch oriented”: we tend to read/accumulate buffers and intermediate structures (page bytes, decoded values, etc.) in lists/vectors before producing the final Column/DataFrame. This makes it harder to:

• guarantee bounded memory for large row groups / wide tables
• structure the decode pipeline as composable stages
• add concurrency safely (e.g. per-column-chunk, per-page, per-row-group)
• instrument progress/backpressure/early termination

This issue proposes refactoring the Parquet reader to use Streamly as the backbone for:
file → row groups → column chunks → pages → decoded values → column builders.

The goal is not to “rewrite everything”, but to introduce a streaming API and progressively migrate internals so we can keep memory stable and make future features easier (predicate pushdown, projection pushdown, row-group skipping, async IO, etc.).

---

Motivation / Why Streamly

Streamly gives us:

• Streaming IO with backpressure (avoid holding entire pages/columns in memory)
• Compositional pipelines (clear “stages” with well-defined responsibilities)
• Optional concurrency (map/parMap style over row groups / column chunks)
• A path to incremental decoding + building without inventing our own streaming abstractions

Even if we keep a “read whole column” convenience API at the top-level, the internals can be streaming.

---

Current implementation (as-is, high level)

This is a conceptual description of what the current reader effectively does (names may differ by module/function):

1. Open file and parse footer/metadata to get:

   • schema, logical types
   • row groups, column chunks (offsets, sizes, compression codecs)
   • encodings, statistics, dictionary page presence, etc.

2. For each row group:

   • For each column chunk:

     • Seek to the chunk offset

     • Read chunk bytes (or read page-by-page, but often collecting)

     • Parse PageHeader → read page payload

     • Decode:

       • repetition levels / definition levels
       • dictionary pages + indices
       • data pages (plain / dict / RLE / bit-pack)
       • decompress if needed (snappy/gzip/zstd/etc.)

     • Convert decoded values into a DI.Column / Column representation:

       • handle nullability via def levels
       • handle types (int32/int64/byte array/fixed len byte array, etc.)
       • produce final vector(s), sometimes via intermediate lists/vectors

3. Combine decoded columns into a DataFrame.

Pain points in the current shape

• Intermediate accumulation: lists of pages, indices, values, def levels, etc.

• IO patterns: ad-hoc seeking/reading, unclear buffering strategy, some repeated allocation

• Difficult to add concurrency without risking spikes in memory or complicating code

• Harder to implement:

  • “stop early” (e.g. row limit)
  • projection pushdown (read only selected columns)
  • row-group skipping based on stats (without preloading more than necessary)
  • robust progress instrumentation

---

Proposal: Streamly-based Parquet decode pipeline

Guiding principles

• Keep a stable public API (or add a streaming variant alongside it).
• Define a small set of streaming primitives that become the “spine” of Parquet reading.
• Decode page-by-page, producing incremental column chunks.
• Avoid “decode everything into lists then convert”; instead “decode stream → builder”.

New internal pipeline (conceptual)

Stage A — Metadata

• Parse footer/metadata as today (this is small; no need to stream).

Stage B — Row group streaming

• Stream RowGroupInfo

Stage C — Column chunk streaming

• For each row group:

  • Stream ColumnChunkInfo

Stage D — Page streaming

• For a given column chunk:

  • Seek to chunk start

  • Stream:

    • PageHeader
    • PagePayload (decompressed bytes or a decompression stream)

Stage E — Decode streaming

• Convert page payload stream into typed value stream:

  • handle dictionary pages by updating per-chunk decode state
  • handle data pages by emitting decoded values incrementally
  • handle def/rep levels by emitting Maybe a or a (values + bitmap) representation

Stage F — Column building

• Feed the decoded stream into a builder:

  • Stream (Maybe a) or Stream a + null bitmap
  • produce final DI.Column / Column at end of stream
  • optionally produce chunked output and concatenate at end

[sharmrj]

I can take a look at this.

[ShilohAlleyne]

Hi, is this part of the summer of haskell projects? I've really enthusiastic about this and would love some more direction/instruction on how to contribute.

[mchav]

@ShilohAlleyne the GSOC project is the writer. We want to improve the reader prior to GSOC so the code base is in relatively good shape for a student to work in.

[manavgupta26]

Hi, I’ve been reviewing the current Parquet reader along with this proposal.

My understanding is that the main shift is from accumulating pages/vectors to a streaming decode to builder pipeline using Streamly. I’m particularly curious how dictionary page state would be handled in a streaming model, since it affects subsequent data pages within a column chunk. Would this be managed as per chunk decode state threaded through the stream?

Also, would the first milestone focus on introducing page streaming while keeping column building mostly intact, or is the intent to refactor end-to-end from pages to builders immediately?

I’m exploring this with the goal of contributing, so I’d appreciate guidance on a good entry point.

[Eiko-Tokura]

I have a question (or suggestion):

currently, the reader performs no seeking, it loads the entire file bytestring in memory starting from readMetadataFromPath

-- | read the file in memory at once, parse magicString and return the entire file ByteString
readMetadataFromPath :: FilePath -> IO (FileMetadata, BSO.ByteString)
readMetadataFromPath path = do
    contents <- BSO.readFile path
    let (size, magicString) = contents `seq` readMetadataSizeFromFooter contents
    when (magicString /= "PAR1") $ error "Invalid Parquet file"
    meta <- readMetadata contents size
    pure (meta, contents)

in principle we should not do this, we should use a seekable handle, like

readMetadataFromHandle :: SeekableHandle -> IO FileMetadata
readMetadataFromHandle sh = do
    footerBs <- lastBytes (fromIntegral footerSize) sh
    let (size, magicString) = readMetadataSizeFromFooterSlice footerBs
    when (magicString /= "PAR1") $ error "Invalid Parquet file"
    readMetadataByHandleMetaSize sh size

After we read all the metadata, I think we should only read column chunks we are interested in, performing seeking in the file handle and never read the entire file bytestring in memory.

If possible, we/I can start by changing the current implementations so that it all start by looking at a handle, and avoid reading entire file. I'm also looking at this because I think understanding the reader would help me understanding the writer. Thanks

[sharmrj]

Yes we're not going to be loading the whole file into memory with the refactor: https://github.com/sharmrj/dataframe/blob/streaming-parquet/src/DataFrame/IO/Utils/RandomAccess.hs#L60-L71

That should reduce our memory consumption.

[sharmrj]

Hi, I’ve been reviewing the current Parquet reader along with this proposal.

My understanding is that the main shift is from accumulating pages/vectors to a streaming decode to builder pipeline using Streamly. I’m particularly curious how dictionary page state would be handled in a streaming model, since it affects subsequent data pages within a column chunk. Would this be managed as per chunk decode state threaded through the stream?

@manavgupta26

From https://github.com/apache/parquet-format/blob/master/src/main/thrift/parquet.thrift#L698C1-L702C5:

/**
 * The dictionary page must be placed at the first position of the column chunk
 * if it is partly or completely dictionary encoded. At most one dictionary page
 * can be placed in a column chunk.
 **/
struct DictionaryPageHeader {

So we need only check the very first position of each columnChunk to see if there's a dataheader. If one in the wrong position we could very well error out, or we could simply ignore it by Skipping it if we wanted to be permissive

In any case streamly streams are state machines and thread state through themselves so we it wouldn't be problem even if the above were not the case.