XiaoHongbo-Hope commented on code in PR #633:
URL: https://github.com/apache/paimon-rust/pull/633#discussion_r3680764910
##########
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:
Fixed in 4553e82. Each row-group producer now has a one-batch channel and
reserves capacity before polling the next batch, restoring batch-level
backpressure. The concurrency and projected-byte semaphores are shared by the
entire scan, including concurrent DataFusion partitions. Both limits are
configurable, and the new tests cover slow consumers, multiple readers sharing
one budget, ordering, and permit release.
--
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]
