Incremental Parquet Decoding: avoid the need to buffer pages for all used columns in a RowGroup in Async parquet reader #6946

Activity

1. alamb

   added

   enhancementNew feature/additionNew feature/addition

   on Jan 6, 2025
2. tustvold commented on Jan 7, 2025

   

   tustvold

   on Jan 7, 2025

   Last edited by tustvold

   Contributor

   More actions

   I think this boils down to what the characteristics of the backing store are:

   Memory

   Data is already buffered in memory it doesn't matter what approach we use, tbh people should be using the sync reader for this.

   Object Storage

   Object stores do not support true vectorized I/O, that is you can't fetch disjoint ranges in a single request. This means that to fetch 5 pages from each column, as in the intermediate buffering approach, it will actually perform one request per column. Unless the column chunk is small in which case it may just fetch the entire column chunk, and then chuck the extra pages away, which is even worse. These requests are made in parallel, but this does have potentially significant cost implications.

   I'd posit that the current approach is the right one for object stores.

   Local SSD

   Local SSD is the option where this gets potentially interesting, as something closer to the minimize buffering approach becomes viable. IMO this is where something like #5522 becomes more relevant, especially if wanting to use something like io_uring. The current abstraction just starts to fall apart at this point, and I'm not sure it is sensible to try to contort it to make it work.

   Alternatives

   A simpler option might just be to write files with smaller row groups, this effectively "materializes" the intermediate buffering approach into the file, but with the added benefit that it doesn't break IO coalescing. This is effectively the observation made by file formats like Lance when they got rid of the column chunk.

   There is a slight real difference between one row group with 50 pages and 5 row groups with 10 pages each, concerning dictionary pages, but based on what I guess is motivating this request, just writing smaller row groups may be a better option

   ❤️React with ❤️1Reacted by Trevor Hilton
3. alamb commented on Jan 7, 2025

   

   alamb

   on Jan 7, 2025

   Last edited by alamb

   ContributorAuthor

   More actions

   Local SSD

   Local SSD is the option where this gets potentially interesting, as something closer to the minimize buffering approach becomes viable. IMO this is where something like #5522 becomes more relevant, especially if wanting to use something like io_uring. The current abstraction just starts to fall apart at this point, and I'm not sure it is sensible to try to contort it to make it work.

   This is a very good observation -- the usecase we have is exactly this (data is local on SSD not on remote object storage)

   I also think there might be value in a hybrid approach to reduce RAM requirements: fetching data from on object store to a local SSD and then reading it more incrementally

   So if we can use the same abstraction that would be very nice

   A simpler option might just be to write files with smaller row groups, this effectively "materializes" the intermediate buffering approach into the file, but with the added benefit that it doesn't break IO coalescing. This is effectively the observation made by file formats like Lance when they got rid of the column chunk.

   Indeed, this is exactly the workaround we are trying internally and I or @hiltontj will report back here on how well it worked.
4. tustvold commented on Jan 7, 2025

   

   tustvold

   on Jan 7, 2025

   Contributor

   More actions

   Tbh if using local SSD I'd be tempted to just use the sync ParquetRecordBatchReader, the runtime needs to handle blocking anyway due to the CPU bound work, and the IO latencies are likely small enough to be irrelevant in comparison.

   👍React with 👍1Reacted by Andrew Lamb
5. alamb commented on Jan 7, 2025

   

   alamb

   on Jan 7, 2025

   Last edited by alamb

   ContributorAuthor

   More actions

   Tbh if using local SSD I'd be tempted to just use the sync ParquetRecordBatchReader, the runtime needs to handle blocking anyway due to the CPU bound work, and the IO latencies are likely small enough to be irrelevant in comparison.

   In the context of DataFusion all IO is done in terms of the ObjectStore trait -- so I think the trick is in figuring out how to wire in / call the right version.

   If we decide reducing buffering is important enough usecase for us at Influx we'll spend some more time trying out various options and report back
6. XiangpengHao commented on Jan 7, 2025

   

   XiangpengHao

   on Jan 7, 2025

   Contributor

   More actions

   I believe we need something like AsyncChunkReader (https://github.com/apache/arrow-rs/blob/main/parquet/src/file/reader.rs#L63) which allows us to read the pages on demand in async, and thus avoiding buffering the entire row group in memory before creating the record batch stream.
   But that means we need to plague almost the entire crate to be async functions.
7. tustvold commented on Jan 7, 2025

   

   tustvold

   on Jan 7, 2025

   Contributor

   More actions

   I strongly disagree that intertwining IO in that way is desirable, let alone practical. I think it is far better to expose lower-level push-based APIs for decoding, that allow users to orchestrate IO as they see fit. This is the approach we take for JSON and CSV, parquet poses some additional challenges, but none that are insurmountable I don't think.
8. alamb commented on Jun 24, 2025

   

   alamb

   on Jun 24, 2025

   ContributorAuthor

   More actions

   I think this tradeoff between buffering / reducing IO will also appear in other cases (like optimizing filter pushdown) so maybe we will revisit it (perhaps with a setting to control the max buffer size or something)

   • related [EPIC] Faster performance for parquet predicate evaluation for non selective filters #7456
9. adriangb commented on Feb 22, 2026

   

   adriangb

   on Feb 22, 2026

   Contributor

   More actions

   @alamb does your work on the push decoder help here?
10. alamb commented on Feb 27, 2026

    

    alamb

    on Feb 27, 2026

    ContributorAuthor

    More actions

    TLDR is yes -- I think so

    It is now possible to

    1. configure the Parquet reader to read any arbitrary slice of a row group (using RowSelections)
    2. schedule the IO however you want (e.g. using the PushDecoder)

    So for example if you wanted to read only 100k rows at a time from a parquet file, you could make multiple ParquetPushDecoders that each read a 100k chunk of the file, and then could run them in whatever order / amount of pipelineing you wanted

    I don't think this is well explained (it would make a great example on the blog I would love to write when I have infinte time)

    • Blog / paper post about parquet push decoder #8035
11. adriangb commented on Jul 23, 2026

    

    adriangb

    on Jul 23, 2026

    Last edited by adriangb

    Contributor

    More actions

    I was looking into this again on our production system. There's two problems that this is creating for us:

    1. Because our columns are very wide / have large values (e.g. 1MB LLM prompts) buffering at the row group level causes high memory usage. Even if we could set some row limit that's still not optimal since depending on what columns the query reads, the shape of the data, etc. the optimal number varies widely.
    2. Since fetching data off of object storage is slow there's latency because every time we need more data we have to wait to buffer an entire row group before we start decoding / processing it.

    I started looking into a design that has a stream per column. This results in more object store GETs but allows each column to stream at whatever pace it needs to to feed decoding and minimizes how much data has to be buffered in memory to keep the CPU fully fed. I initially got mixed results, but I'm posting here to bump the issue and share the pain / solution ideas.
12. alamb commented on Jul 23, 2026

    

    alamb

    on Jul 23, 2026

    ContributorAuthor

    More actions

    I started looking into a design that has a stream per column. This results in more object store GETs but allows each column to stream at whatever pace it needs to to feed decoding and minimizes how much data has to be buffered in memory to keep the CPU fully fed. I initially got mixed results, but I'm posting here to bump the issue and share the pain / solution ideas.

    FWIW I think turning on predicate pushdown also (effectively) increases the GET requests but reduces the buffering (depending on the selectivyt of the filters) and that has seen to be a very mixed bag.

    Because our columns are very wide / have large values (e.g. 1MB LLM prompts) buffering at the row group level causes high memory usage. Even if we could set some row limit that's still not optimal since depending on what columns the query reads, the shape of the data, etc. the optimal number varies widely.

    @adriangb I wonder if you could be slightly more clever and compute the amount of data that would be buffered for some slices of rows, using the OffsetIndex -- you might be able to use that to determine how many rows and bytes there are for each page

    https://github.com/apache/parquet-format/blob/21ecc2b0f5786070561ae0d34152230853c83e2b/src/main/thrift/parquet.thrift#L1239-L1251
13. alamb commented on Sep 25, 2026

    

    alamb

    on Sep 25, 2026

    ContributorAuthor

    More actions

    This came up in a discussion @adriangb had today. Basically the idea is to enable reads at the page granularity, while today reads are done at the row group granularity.

    In my mind, one of the main use cases for the ScanPlan API in #10555 is to allow incremental decoding of row groups (aka not having to fetch all the pages for all columns used for the columns in that row group up front)

    Right now, when the ParquetPushDecoder starts to read a row group, it will compute all the pages necessary for either evaluating a pushdown filter OR all the pages for the projection. @adriangb mentioned to me that he thought it was possible to pre-push the data needed into the push decoder and it would not ask for more data until it ran out of them.

    I want to prove/disprove this by finding or writing a test that demonstrates pushing only the pages needed for the next record batch (not the whole row group) and decoding it if possible

    Here is some research about how the ranges are computed / needed today

    Details

    Where the ranges for all columns are computed

    • arrow-rs/parquet/src/arrow/in_memory_row_group.rs

      Lines 59 to 134 in 2f2360d

      	pub(crate) fn fetch_ranges(
      	&self,
      	projection: &ProjectionMask,
      	selection: Option<&RowSelection>,
      	batch_size: usize,
      	cache_mask: Option<&ProjectionMask>,
      	) -> FetchRanges {
      	let metadata = self.metadata.row_group(self.row_group_idx);
      	if let Some((selection, page_index)) = selection.zip(self.page_index.as_ref()) {
      	let expanded_selection =
      	selection.expand_to_batch_boundaries(batch_size, self.row_count);
      	// If we have a `RowSelection` and an `OffsetIndex` then only fetch
      	// pages required for the `RowSelection`
      	// Consider preallocating outer vec: https://github.com/apache/arrow-rs/issues/8667
      	let mut page_start_offsets: Vec<Option<Vec<u64>>> = vec![];
      	let ranges = self
      	.column_chunks
      	.iter()
      	.zip(metadata.columns())
      	.enumerate()
      	.filter(|&(idx, (chunk, _chunk_meta))| {
      	chunk.is_none() && projection.leaf_included(idx)
      	})
      	.flat_map(|(idx, (_chunk, chunk_meta))| {
      	// If the first page does not start at the beginning of the column,
      	// then we need to also fetch a dictionary page.
      	let mut ranges: Vec<Range<u64>> = vec![];
      	let (start, len) = chunk_meta.byte_range();
      	let Some(offset_idx) = page_index.offset_index(idx) else {
      	// No offset index for this column, fetch the entire column
      	ranges.push(start..start + len);
      	page_start_offsets.push(None);
      	return ranges;
      	};
      	match offset_idx.page_locations().first() {
      	Some(first) if first.offset as u64 != start => {
      	ranges.push(start..first.offset as u64);
      	}
      	_ => (),
      	}
      	// Expand selection to batch boundaries if needed for caching
      	// (see doc comment for this function for details on `cache_mask`)
      	let use_expanded = cache_mask.map(|m| m.leaf_included(idx)).unwrap_or(false);
      	if use_expanded {
      	ranges.extend(expanded_selection.scan_ranges(offset_idx.page_locations()));
      	} else {
      	ranges.extend(selection.scan_ranges(offset_idx.page_locations()));
      	}
      	page_start_offsets.push(Some(ranges.iter().map(|range| range.start).collect()));
      	ranges
      	})
      	.collect();
      	FetchRanges {
      	ranges,
      	page_start_offsets: Some(page_start_offsets),
      	}
      	} else {
      	let ranges = self
      	.column_chunks
      	.iter()
      	.enumerate()
      	.filter(|&(idx, chunk)| chunk.is_none() && projection.leaf_included(idx))
      	.map(|(idx, _chunk)| {
      	let column = metadata.column(idx);
      	let (start, length) = column.byte_range();
      	start..(start + length)
      	})
      	.collect();
      	FetchRanges {
      	ranges,
      	page_start_offsets: None,

      InMemoryRowGroup::fetch_ranges iterates every leaf column and returns either a single range per column chunk or ranges for each page

    • arrow-rs/parquet/src/arrow/push_decoder/reader_builder/data.rs

      Lines 190 to 230 in 2f2360d

      	let Self {
      	row_group_idx,
      	row_count,
      	batch_size,
      	parquet_metadata,
      	projection,
      	selection,
      	cache_projection,
      	column_chunks,
      	} = self;
      	let row_group_meta_data = parquet_metadata.row_group(row_group_idx);
      	// If no previously read column chunks are provided, create a new location to hold them
      	let column_chunks =
      	column_chunks.unwrap_or_else(|| vec![None; row_group_meta_data.columns().len()]);
      	let page_index = if parquet_metadata
      	.page_index()
      	.is_some_and(|pi| pi.has_offset_indexes())
      	{
      	Some(parquet_metadata.page_index_for_row_group(row_group_idx))
      	} else {
      	None
      	};
      	// Create an InMemoryRowGroup to hold the column chunks, this is a
      	// temporary structure used to tell the ArrowReaders what pages are
      	// needed for decoding
      	let row_group = InMemoryRowGroup {
      	row_count,
      	column_chunks,
      	page_index,
      	row_group_idx,
      	metadata: parquet_metadata,
      	};
      	let FetchRanges {
      	ranges,
      	page_start_offsets,
      	} = row_group.fetch_ranges(projection, selection, batch_size, cache_projection);

      DataRequestBuilder::build built from the result of fetch_ranges

    Where the push decoder issues the requests, per row group

    • arrow-rs/parquet/src/arrow/push_decoder/reader_builder/mod.rs

      Lines 544 to 562 in 2f2360d

      	// Make a request for the data needed to evaluate the current predicate
      	let predicate = filter_info.current();
      	// need to fetch pages the column needs for decoding, figure
      	// that out based on the current selection and projection
      	let data_request = DataRequestBuilder::new(
      	row_group_idx,
      	row_count,
      	self.batch_size,
      	&self.metadata,
      	predicate.projection(), // use the predicate's projection
      	)
      	.with_selection(plan_builder.selection())
      	// Cached output columns reuse these predicate-stage chunks. Expand their
      	// selection to cache batch boundaries so a cache miss can safely fetch a
      	// complete batch from the retained sparse column data.
      	.with_cache_projection(Some(filter_info.cache_projection()))
      	.with_column_chunks(column_chunks)
      	.build();

      the filter path builds a DataRequest using the current predicate's projection.

    • arrow-rs/parquet/src/arrow/push_decoder/reader_builder/mod.rs

      Lines 720 to 731 in 2f2360d

      	let data_request = DataRequestBuilder::new(
      	row_group_idx,
      	row_count,
      	self.batch_size,
      	&self.metadata,
      	&self.projection,
      	)
      	.with_selection(plan_builder.selection())
      	.with_column_chunks(column_chunks)
      	// Final projection fetch shouldn't expand selection for cache
      	// so don't call with_cache_projection here
      	.build();

      the output path builds a DataRequest using the final projection.

    Where decoder refuses to proceed until every range is present

    • arrow-rs/parquet/src/arrow/push_decoder/reader_builder/data.rs

      Lines 48 to 54 in 2f2360d

      	pub fn needed_ranges(&self, buffers: &PushBuffers) -> Vec<Range<u64>> {
      	self.ranges
      	.iter()
      	.filter(|&range| !buffers.has_range(range))
      	.cloned()
      	.collect()
      	}

      DataRequest::needed_ranges returns whichever ranges the buffers do not yet hold.

    • arrow-rs/parquet/src/arrow/push_decoder/reader_builder/mod.rs

      Lines 582 to 594 in 2f2360d

      	// figure out what ranges we still need
      	let needed_ranges = data_request.needed_ranges(&self.buffers);
      	if !needed_ranges.is_empty() {
      	// still need data
      	return Ok(NextState::result(
      	RowGroupDecoderState::WaitingOnFilterData {
      	row_group_info,
      	filter_info,
      	data_request,
      	},
      	RowGroupBuildResult::NeedsData(needed_ranges),
      	));
      	}

      and

      arrow-rs/parquet/src/arrow/push_decoder/reader_builder/mod.rs

      Lines 762 to 773 in 2f2360d

      	let needed_ranges = data_request.needed_ranges(&self.buffers);
      	if !needed_ranges.is_empty() {
      	// still need data
      	return Ok(NextState::result(
      	RowGroupDecoderState::WaitingOnData {
      	row_group_info,
      	data_request,
      	cache_info,
      	},
      	RowGroupBuildResult::NeedsData(needed_ranges),
      	));
      	}

      return NeedsData while any range is missing. The row group is only materialized once the full set is buffered.
14. alamb commented on Sep 25, 2026

    

    alamb

    on Sep 25, 2026

    ContributorAuthor

    More actions

    My test suggests that the PushDecoder still tries to request the entire row group today:

    • test(parquet): show push decoder needs a whole row group before decoding a batch #11218

    👍React with 👍1Reacted by Adrian Garcia Badaracco
15. alamb

    changed the title [-]Avoid buffering the entire row group in Async parquet reader[/-] [+]Incremental Parquet Decoding: avoid the need to buffer pages for all used columns in a RowGroup in Async parquet reader[/+] on Sep 25, 2026
16. alamb

    added

    parquetChanges to the parquet crateChanges to the parquet crate

    on Sep 25, 2026
17. adriangb commented on Sep 25, 2026

    

    adriangb

    on Sep 25, 2026

    Contributor

    More actions

    Yes, you are right! Turns out I was hacking around that. Do you think we can change that (first step is your test) and go from there? I see no reason why the push decoder wouldn't be able to emit a batch as soon as it has all of the data needed to decode and materialize it.
18. alamb commented on Sep 25, 2026

    

    alamb

    on Sep 25, 2026

    Last edited by alamb

    ContributorAuthor

    More actions

    Yes, you are right! Turns out I was hacking around that. Do you think we can change that (first step is your test) and go from there? I see no reason why the push decoder wouldn't be able to emit a batch as soon as it has all of the data needed to decode and materialize it.

    Yes! I think this is the first step for sure

    Then we can add the ScanPlan API and I think we'll be in a pretty great position to let callers incrementally decide how to feed data to the push decoder, and give them APIs to know what to push

    🚀React with 🚀1Reacted by Adrian Garcia Badaracco
19. adriangb

    added 4 commits that reference this issue on Sep 25, 2026

    chore: pin arrow-rs with batch-granular push decoding and scan_plan

    ...

    d38c786

    feat(parquet): batch-granular fetching in ParquetPushDecoder

    ...

    3f9ef62

    feat(parquet): batch-granular fetching in ParquetPushDecoder

    ...

    ae710a6

    chore: pin arrow-rs with batch-granular push decoding and scan_plan

    ...

    c1301b5
20. alamb

    added a commit that references this issue on Sep 28, 2026

    test(parquet): show push decoder needs a whole row group before decod…

    ...

    21f1156
21. adriangb

    added 4 commits that reference this issue on Sep 28, 2026

    feat(parquet): batch-granular fetching in ParquetPushDecoder

    ...

    471560d

    test(parquet): show push decoder needs a whole row group before decod…

    ...

    e9ae859

    chore: pin arrow-rs with batch-granular push decoding and scan_plan

    ...

    f679435

    chore: pin arrow-rs with batch-granular push decoding and scan_plan

    ...

    bd71221
22. Jefffrey

    added a commit that references this issue on Oct 6, 2026

    refactor(parquet): share push decoder fetch-range and stage logic (pr…

    ...

    02accce
23. mbutrovich

    added a commit that references this issue on Oct 9, 2026

    feat(parquet): add ParquetPushDecoder::scan_plan for read-ahead plann…

    ...

    cf3cda1
24. alamb

    added a commit that references this issue on Oct 9, 2026

    perf(parquet): store a ScanPlan cursor as one heap entry (apache#11334)

    ...

    4ef327c