QuakeWang commented on code in PR #633:
URL: https://github.com/apache/paimon-rust/pull/633#discussion_r3680220037
##########
crates/paimon/src/arrow/format/parquet.rs:
##########
@@ -460,6 +467,61 @@ impl FormatFileReader for ParquetFormatReader {
MapShreddingReadPlan::create(&scan_fields,
batch_stream_builder.schema())?
.map(Arc::new);
+ // A normal Parquet stream fetches and decodes row groups one by one.
+ // For remote object stores, a full scan of a compacted file can
+ // therefore serialize dozens of independent range requests behind one
+ // DataFusion partition. Build one stream per row group on the
+ // predicate-free path and collect a bounded number concurrently.
+ //
+ // `buffered` preserves the input order, so positional `_ROW_ID`,
+ // sort-order, and downstream merge assumptions remain unchanged. Reads
+ // with predicates or an explicit row selection retain the original
+ // single-stream path until their selections are split per row group.
+ let row_group_parallelism =
+ if preds.is_empty() && row_filter_factory.is_none() &&
row_selection.is_none() {
+
parquet_row_group_read_parallelism(batch_stream_builder.metadata(), &mask)
+ } else {
+ 1
+ };
+ if row_group_parallelism > 1 {
+ let row_group_count =
batch_stream_builder.metadata().num_row_groups();
+ let reader_metadata = ArrowReaderMetadata::try_new(
+ batch_stream_builder.metadata().clone(),
+ ArrowReaderOptions::new(),
+ )?;
+ let mut row_groups =
+ futures::stream::iter((0..row_group_count).map(move
|row_group_index| {
+ let shared_reader = Arc::clone(&shared_reader);
+ let reader_metadata = reader_metadata.clone();
+ let mask = mask.clone();
+ async move {
+ let mut builder =
ParquetRecordBatchStreamBuilder::new_with_metadata(
+ ArrowFileReader::new(file_size, shared_reader),
+ reader_metadata,
+ )
+ .with_projection(mask)
+ .with_row_groups(vec![row_group_index]);
+ if let Some(size) = batch_size {
+ builder = builder.with_batch_size(size);
+ }
+ let stream = builder.build().map_err(Error::from)?;
+
stream.try_collect::<Vec<_>>().await.map_err(Error::from)
Review Comment:
`try_collect::<Vec<_>>()` materializes all batches in each row group before
yielding any output. With multiple row groups running concurrently, decoded
Arrow batches are retained in addition to the Parquet bytes used to
calculate`row_group_parallelism`, so actual in-flight memory can substantially
exceed the intended budget and batch-level backpressure is lost.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
