From 069d68d3cf56459ec537e73d3c8b2756a3dff20f Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 22 Jul 2026 04:03:53 -0500 Subject: [PATCH 1/5] Add experimental bounded streaming option for reading large parquet column chunks Adds datafusion.execution.parquet.bounded_streaming (default false). When enabled, the parquet push decoder decodes row groups with large projected column chunks in bounded windows, and the fetch loop consumes those chunks via single ranged requests whose bodies stream into a bounded buffer (parquet AdaptiveFetcher), instead of buffering every projected byte of a row group before decode starts. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG --- datafusion/common/src/config.rs | 11 +++ .../common/src/file_options/parquet_writer.rs | 1 + .../datasource-parquet/src/opener/mod.rs | 17 ++++- .../datasource-parquet/src/push_decoder.rs | 71 ++++++++++++++++--- datafusion/datasource-parquet/src/reader.rs | 34 +++++++++ datafusion/datasource-parquet/src/source.rs | 1 + datafusion/proto-common/src/from_proto/mod.rs | 2 + .../proto/src/logical_plan/file_formats.rs | 2 + .../test_files/information_schema.slt | 2 + docs/source/user-guide/configs.md | 1 + 10 files changed, 129 insertions(+), 13 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index 81f573fc2a23e..1e7111e2fc15f 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -1189,6 +1189,17 @@ config_namespace! { /// parquet reader setting. 0 means no caching. pub max_predicate_cache_size: Option, default = None + /// (reading) If true, large parquet column chunks are read with + /// bounded streaming: instead of buffering every projected column + /// chunk of a row group in memory before decoding, chunks over a + /// threshold are fetched with a single ranged request each whose body + /// feeds the decoder incrementally through a bounded buffer. This + /// bounds reader memory by the buffer size rather than the row group + /// size, and overlaps decoding with data transfer. Requires the + /// parquet offset index (see `enable_page_index`); row groups without + /// one, or with pushed-down filters, fall back to the buffered path. + pub bounded_streaming: bool, default = false + // The following options affect writing to parquet files // and map to parquet::file::properties::WriterProperties diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 320bfcf33e488..7e8c5e26c4c4a 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -238,6 +238,7 @@ impl ParquetOptions { pushdown_filters: _, reorder_filters: _, force_filter_selections: _, // not used for writer props + bounded_streaming: _, // not used for writer props allow_single_file_parallelism: _, maximum_parallel_row_group_writers: _, maximum_buffered_record_batches_per_stream: _, diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index af97a192fa7ce..f3dec9935e832 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -27,7 +27,7 @@ use crate::access_plan::PreparedAccessPlan; use crate::decoder_projection::DecoderProjection; use crate::page_filter::PagePruningAccessPlanFilter; use crate::push_decoder::{ - DecoderBuilderConfig, PushDecoderStreamState, RgPlanEntry, RowGroupPruner, + DecoderBuilderConfig, FetchState, PushDecoderStreamState, RgPlanEntry, RowGroupPruner, }; use crate::row_filter::RowFilterGenerator; use crate::row_group_filter::RowGroupAccessPlanFilter; @@ -289,6 +289,10 @@ pub(super) struct ParquetMorselizer { /// Maximum size of the predicate cache, in bytes. If none, uses /// the arrow-rs default. pub max_predicate_cache_size: Option, + /// See [`ParquetOptions::bounded_streaming`] + /// + /// [`ParquetOptions::bounded_streaming`]: datafusion_common::config::ParquetOptions::bounded_streaming + pub bounded_streaming: bool, /// Whether to read row groups in reverse order pub reverse_row_groups: bool, /// Optional sort order used to reorder row groups by their min/max statistics. @@ -451,6 +455,7 @@ struct PreparedParquetOpen { expr_adapter_factory: Arc, predicate_creation_errors: Count, max_predicate_cache_size: Option, + bounded_streaming: bool, reverse_row_groups: bool, sort_order_for_reorder: Option, preserve_order: bool, @@ -850,6 +855,7 @@ impl ParquetMorselizer { expr_adapter_factory: Arc::clone(&self.expr_adapter_factory), predicate_creation_errors, max_predicate_cache_size: self.max_predicate_cache_size, + bounded_streaming: self.bounded_streaming, reverse_row_groups: self.reverse_row_groups, sort_order_for_reorder: self.sort_order_for_reorder.clone(), preserve_order: self.preserve_order, @@ -1412,6 +1418,7 @@ impl RowGroupsPrunedParquetOpen { arrow_reader_metrics: &arrow_reader_metrics, force_filter_selections: prepared.force_filter_selections, decoder_limit: prepared.limit, + bounded_streaming: prepared.bounded_streaming, }; let prepared_access_plan = prepare_access_plan(access_plan)?; @@ -1481,7 +1488,10 @@ impl RowGroupsPrunedParquetOpen { decoder: Some(decoder), active_reader: None, rg_plan, - reader: prepared.async_file_reader, + reader: FetchState::new( + prepared.async_file_reader, + prepared.bounded_streaming, + ), decoder_projection, arrow_reader_metrics, predicate_cache_inner_records, @@ -1752,6 +1762,7 @@ mod test { enable_row_group_stats_pruning: bool, coerce_int96: Option, max_predicate_cache_size: Option, + bounded_streaming: bool, reverse_row_groups: bool, preserve_order: bool, } @@ -1860,6 +1871,7 @@ mod test { enable_row_group_stats_pruning: false, coerce_int96: None, max_predicate_cache_size: None, + bounded_streaming: false, reverse_row_groups: false, preserve_order: false, } @@ -2037,6 +2049,7 @@ mod test { #[cfg(feature = "parquet_encryption")] encryption_factory: None, max_predicate_cache_size: self.max_predicate_cache_size, + bounded_streaming: self.bounded_streaming, reverse_row_groups: self.reverse_row_groups, sort_order_for_reorder: None, virtual_state, diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 31bd365a4631d..3c68b6d2b5165 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -47,9 +47,10 @@ use parquet::DecodeResult; use parquet::arrow::ProjectionMask; use parquet::arrow::arrow_reader::metrics::ArrowReaderMetrics; use parquet::arrow::arrow_reader::{ - ArrowReaderMetadata, ParquetRecordBatchReader, RowSelectionPolicy, + ArrowReaderMetadata, BoundedStreamingOptions, ParquetRecordBatchReader, + RowSelectionPolicy, }; -use parquet::arrow::async_reader::AsyncFileReader; +use parquet::arrow::async_reader::{AdaptiveFetcher, AsyncFileReader}; use parquet::arrow::push_decoder::{ParquetPushDecoder, ParquetPushDecoderBuilder}; use parquet::file::metadata::ParquetMetaData; @@ -75,6 +76,10 @@ pub(crate) struct DecoderBuilderConfig<'a> { pub(crate) arrow_reader_metrics: &'a ArrowReaderMetrics, pub(crate) force_filter_selections: bool, pub(crate) decoder_limit: Option, + /// See [`ParquetOptions::bounded_streaming`] + /// + /// [`ParquetOptions::bounded_streaming`]: datafusion_common::config::ParquetOptions::bounded_streaming + pub(crate) bounded_streaming: bool, } impl DecoderBuilderConfig<'_> { @@ -102,10 +107,45 @@ impl DecoderBuilderConfig<'_> { if let Some(limit) = self.decoder_limit { builder = builder.with_limit(limit); } + if self.bounded_streaming { + builder = builder.with_bounded_streaming(BoundedStreamingOptions::default()); + } builder } } +/// How [`PushDecoderStreamState`] fetches the byte ranges the decoder asks +/// for. +pub(crate) enum FetchState { + /// Fetch each `NeedsData` request in full via + /// [`AsyncFileReader::get_byte_ranges`] + Plain(Box), + /// Bounded streaming: small coalesced ranges are materialized as in + /// `Plain`; large column chunks are consumed incrementally, window by + /// window. See [`ParquetOptions::bounded_streaming`]. + /// + /// [`ParquetOptions::bounded_streaming`]: datafusion_common::config::ParquetOptions::bounded_streaming + Adaptive(AdaptiveFetcher>), +} + +impl FetchState { + /// Wrap `reader` for bounded streaming when enabled (and a tokio runtime + /// is available to drive streamed request bodies; without one, fall back + /// to the plain path, which behaves identically to the option being off). + pub(crate) fn new(reader: Box, bounded_streaming: bool) -> Self { + if bounded_streaming { + if let Ok(handle) = tokio::runtime::Handle::try_current() { + return Self::Adaptive(AdaptiveFetcher::new( + BoundedStreamingOptions::default(), + handle, + reader, + )); + } + } + Self::Plain(reader) + } +} + #[derive(Debug, Clone)] pub(crate) struct RgPlanEntry { pub(crate) rg_index: usize, @@ -243,7 +283,7 @@ pub(crate) struct PushDecoderStreamState { pub(crate) decoder: Option, pub(crate) active_reader: Option, pub(crate) rg_plan: VecDeque, - pub(crate) reader: Box, + pub(crate) reader: FetchState, /// Per-file projection: the mask installed on every decoder and the /// per-batch transform applied by [`Self::project_batch`]. pub(crate) decoder_projection: DecoderProjection, @@ -373,13 +413,22 @@ impl PushDecoderStreamState { let decoder = self.decoder.as_mut().expect("decoder present"); match decoder.try_next_reader() { Ok(DecodeResult::NeedsData(ranges)) => { - let data = self - .reader - .get_byte_ranges(ranges.clone()) - .await - .map_err(DataFusionError::from); - match data { - Ok(data) => { + let fetched = match &mut self.reader { + FetchState::Plain(reader) => reader + .get_byte_ranges(ranges.clone()) + .await + .map(|data| (ranges, data)), + FetchState::Adaptive(fetcher) => { + // The adaptive fetcher may return only part of the + // request (the decoder's next window); the decoder + // re-requests what is still missing on the next + // iteration. + let plan = decoder.upcoming_fetch_plan(); + fetcher.fetch_more(ranges, plan).await + } + }; + match fetched { + Ok((ranges, data)) => { if let Err(e) = self .decoder .as_mut() @@ -389,7 +438,7 @@ impl PushDecoderStreamState { return Some((Err(DataFusionError::from(e)), self)); } } - Err(e) => return Some((Err(e), self)), + Err(e) => return Some((Err(DataFusionError::from(e)), self)), } } Ok(DecodeResult::Data(reader)) => { diff --git a/datafusion/datasource-parquet/src/reader.rs b/datafusion/datasource-parquet/src/reader.rs index 4df636b894940..b8c42d23f40b3 100644 --- a/datafusion/datasource-parquet/src/reader.rs +++ b/datafusion/datasource-parquet/src/reader.rs @@ -122,6 +122,23 @@ impl AsyncFileReader for ParquetFileReader { self.inner.get_byte_ranges(ranges) } + fn get_bytes_stream( + &mut self, + range: Range, + ) -> Option< + BoxFuture< + 'static, + parquet::errors::Result< + futures::stream::BoxStream<'static, parquet::errors::Result>, + >, + >, + > { + let bytes_scanned = range.end - range.start; + let stream = self.inner.get_bytes_stream(range)?; + self.file_metrics.bytes_scanned.add(bytes_scanned as usize); + Some(stream) + } + fn get_metadata<'a>( &'a mut self, options: Option<&'a ArrowReaderOptions>, @@ -288,6 +305,23 @@ impl AsyncFileReader for CachedParquetFileReader { self.inner.get_byte_ranges(ranges) } + fn get_bytes_stream( + &mut self, + range: Range, + ) -> Option< + BoxFuture< + 'static, + parquet::errors::Result< + futures::stream::BoxStream<'static, parquet::errors::Result>, + >, + >, + > { + let bytes_scanned = range.end - range.start; + let stream = self.inner.get_bytes_stream(range)?; + self.file_metrics.bytes_scanned.add(bytes_scanned as usize); + Some(stream) + } + fn get_metadata<'a>( &'a mut self, options: Option<&'a ArrowReaderOptions>, diff --git a/datafusion/datasource-parquet/src/source.rs b/datafusion/datasource-parquet/src/source.rs index 3443b08475e0d..4f6c17a2bf85e 100644 --- a/datafusion/datasource-parquet/src/source.rs +++ b/datafusion/datasource-parquet/src/source.rs @@ -647,6 +647,7 @@ impl FileSource for ParquetSource { #[cfg(feature = "parquet_encryption")] encryption_factory: self.get_encryption_factory_with_config(), max_predicate_cache_size: self.max_predicate_cache_size(), + bounded_streaming: self.table_parquet_options.global.bounded_streaming, reverse_row_groups: self.reverse_row_groups, sort_order_for_reorder: self.sort_order_for_reorder.clone(), virtual_state, diff --git a/datafusion/proto-common/src/from_proto/mod.rs b/datafusion/proto-common/src/from_proto/mod.rs index 97cc9af230105..475b6462cf5c0 100644 --- a/datafusion/proto-common/src/from_proto/mod.rs +++ b/datafusion/proto-common/src/from_proto/mod.rs @@ -1063,6 +1063,8 @@ impl TryFrom<&protobuf::ParquetOptions> for ParquetOptions { pushdown_filters: value.pushdown_filters, reorder_filters: value.reorder_filters, force_filter_selections: value.force_filter_selections, + // not serialized: an experimental reading option, defaults off + bounded_streaming: false, data_pagesize_limit: value.data_pagesize_limit as usize, write_batch_size: value.write_batch_size as usize, writer_version: value.writer_version.parse().map_err(|e| { diff --git a/datafusion/proto/src/logical_plan/file_formats.rs b/datafusion/proto/src/logical_plan/file_formats.rs index 8940b16bf83f5..38716723d1ed6 100644 --- a/datafusion/proto/src/logical_plan/file_formats.rs +++ b/datafusion/proto/src/logical_plan/file_formats.rs @@ -546,6 +546,8 @@ mod parquet { pushdown_filters: proto.pushdown_filters, reorder_filters: proto.reorder_filters, force_filter_selections: proto.force_filter_selections, + // not serialized: an experimental reading option, defaults off + bounded_streaming: false, data_pagesize_limit: proto.data_pagesize_limit as usize, write_batch_size: proto.write_batch_size as usize, writer_version, diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 77acaa4747f9d..bc5f73fd07c00 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -237,6 +237,7 @@ datafusion.execution.parquet.bloom_filter_fpp NULL datafusion.execution.parquet.bloom_filter_ndv NULL datafusion.execution.parquet.bloom_filter_on_read true datafusion.execution.parquet.bloom_filter_on_write false +datafusion.execution.parquet.bounded_streaming false datafusion.execution.parquet.coerce_int96 NULL datafusion.execution.parquet.coerce_int96_tz NULL datafusion.execution.parquet.column_index_truncate_length 64 @@ -396,6 +397,7 @@ datafusion.execution.parquet.bloom_filter_fpp NULL (writing) Sets bloom filter f datafusion.execution.parquet.bloom_filter_ndv NULL (writing) Sets bloom filter number of distinct values. If NULL, uses default parquet writer setting datafusion.execution.parquet.bloom_filter_on_read true (reading) Use any available bloom filters when reading parquet files datafusion.execution.parquet.bloom_filter_on_write false (writing) Write bloom filters for all columns when creating parquet files +datafusion.execution.parquet.bounded_streaming false (reading) If true, large parquet column chunks are read with bounded streaming: instead of buffering every projected column chunk of a row group in memory before decoding, chunks over a threshold are fetched with a single ranged request each whose body feeds the decoder incrementally through a bounded buffer. This bounds reader memory by the buffer size rather than the row group size, and overlaps decoding with data transfer. Requires the parquet offset index (see `enable_page_index`); row groups without one, or with pushed-down filters, fall back to the buffered path. datafusion.execution.parquet.coerce_int96 NULL (reading) If true, parquet reader will read columns of physical type int96 as originating from a different resolution than nanosecond. This is useful for reading data from systems like Spark which stores microsecond resolution timestamps in an int96 allowing it to write values with a larger date range than 64-bit timestamps with nanosecond resolution. datafusion.execution.parquet.coerce_int96_tz NULL (reading) Optional timezone applied to INT96 columns when `coerce_int96` is set. When `Some`, INT96 columns coerce to `Timestamp(, Some())` instead of the default `Timestamp(, None)`. Spark and other systems write INT96 values as UTC-adjusted instants, so callers that need the resulting Arrow type to be timezone-aware (e.g. for Spark `TimestampType` semantics) should set this to `"UTC"`. No effect when `coerce_int96` is `None`. datafusion.execution.parquet.column_index_truncate_length 64 (writing) Sets column index truncate length diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index e01af3476b94c..832effb9f22fb 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -93,6 +93,7 @@ The following configuration settings are available: | datafusion.execution.parquet.coerce_int96_tz | NULL | (reading) Optional timezone applied to INT96 columns when `coerce_int96` is set. When `Some`, INT96 columns coerce to `Timestamp(, Some())` instead of the default `Timestamp(, None)`. Spark and other systems write INT96 values as UTC-adjusted instants, so callers that need the resulting Arrow type to be timezone-aware (e.g. for Spark `TimestampType` semantics) should set this to `"UTC"`. No effect when `coerce_int96` is `None`. | | datafusion.execution.parquet.bloom_filter_on_read | true | (reading) Use any available bloom filters when reading parquet files | | datafusion.execution.parquet.max_predicate_cache_size | NULL | (reading) The maximum predicate cache size, in bytes. When `pushdown_filters` is enabled, sets the maximum memory used to cache the results of predicate evaluation between filter evaluation and output generation. Decreasing this value will reduce memory usage, but may increase IO and CPU usage. None means use the default parquet reader setting. 0 means no caching. | +| datafusion.execution.parquet.bounded_streaming | false | (reading) If true, large parquet column chunks are read with bounded streaming: instead of buffering every projected column chunk of a row group in memory before decoding, chunks over a threshold are fetched with a single ranged request each whose body feeds the decoder incrementally through a bounded buffer. This bounds reader memory by the buffer size rather than the row group size, and overlaps decoding with data transfer. Requires the parquet offset index (see `enable_page_index`); row groups without one, or with pushed-down filters, fall back to the buffered path. | | datafusion.execution.parquet.data_pagesize_limit | 1048576 | (writing) Sets best effort maximum size of data page in bytes | | datafusion.execution.parquet.write_batch_size | 1024 | (writing) Sets write_batch_size in rows | | datafusion.execution.parquet.writer_version | 1.0 | (writing) Sets parquet writer version valid values are "1.0" and "2.0" | From 5e4bc316190f21bb1cde16fb44eb0833f20689bb Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 22 Jul 2026 11:37:39 -0500 Subject: [PATCH 2/5] TEMPORARY: patch arrow-rs to the apache/arrow-rs#10410 draft branch Makes the PR buildable by CI and benchmark infrastructure until the parquet changes are released. Must be removed before merge. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG --- Cargo.lock | 84 +++++++++++++++++++++++------------------------------- Cargo.toml | 21 ++++++++++++++ 2 files changed, 56 insertions(+), 49 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6f776e5e965f9..c1809ae5ff7ce 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -165,8 +165,7 @@ checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" [[package]] name = "arrow" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b952ca5a8046ad741b60f142d6eca4aeebcad615694202bc64c5341f23e32c5b" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-arith", "arrow-array", @@ -188,8 +187,7 @@ dependencies = [ [[package]] name = "arrow-arith" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "64a13b8d3008c4e9063c597a08f46446fe3fd5789277127672d6c0bdbb43b1ff" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-array", "arrow-buffer", @@ -202,8 +200,7 @@ dependencies = [ [[package]] name = "arrow-array" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9486151b2f0785bafc6fa04fc5c99fcb4495455662e58787ea32eaaed33c4192" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "ahash", "arrow-buffer", @@ -213,6 +210,7 @@ dependencies = [ "chrono-tz", "half", "hashbrown 0.17.1", + "libc", "num-complex", "num-integer", "num-traits", @@ -221,8 +219,7 @@ dependencies = [ [[package]] name = "arrow-avro" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e4f9b23a0d7b613acb59fa20bdbe0f80ffdae6411498378340b3915e45f5b84" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-array", "arrow-buffer", @@ -245,20 +242,18 @@ dependencies = [ [[package]] name = "arrow-buffer" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4776577a87794bfdf0b4e90e2ea12454fa7738ea2823c4be5b9d1851da7b434" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "bytes", "half", - "num-bigint", + "num-bigint 0.5.1", "num-traits", ] [[package]] name = "arrow-cast" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9ad451ce4f98710828a455b96991b8f031deb2e67f5fcad6773f017e4a69c3a" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-array", "arrow-buffer", @@ -279,8 +274,7 @@ dependencies = [ [[package]] name = "arrow-csv" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8aa7bf96d6141a7bcca2eed57c7c9767d2a2175281857b8a7b68308992864784" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-array", "arrow-cast", @@ -294,8 +288,7 @@ dependencies = [ [[package]] name = "arrow-data" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b38fe43e2e8704360f1464e6e8cc4fc381ef02cc4fb0192afa8df1aaa0115c66" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-buffer", "arrow-schema", @@ -307,8 +300,7 @@ dependencies = [ [[package]] name = "arrow-flight" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42115e09dbb694b5955da998912121451c6910b338228cb80a5701370dba43ff" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-arith", "arrow-array", @@ -325,7 +317,6 @@ dependencies = [ "bytes", "futures", "once_cell", - "paste", "prost", "prost-types", "tonic", @@ -335,8 +326,7 @@ dependencies = [ [[package]] name = "arrow-ipc" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29dac499fcbc6ba74ee0324057821d381929a48526a3966bd9dffb44aa06d98c" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-array", "arrow-buffer", @@ -351,8 +341,7 @@ dependencies = [ [[package]] name = "arrow-json" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fe05e916ddc50f4c7a363cd69c0ef5894fcee063517e9a0b8582f0c56746af6" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-array", "arrow-buffer", @@ -376,8 +365,7 @@ dependencies = [ [[package]] name = "arrow-ord" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0e13dbdc2a9c053c10c7baa6e30faee04a180aa7ce88e471835850ce37abd20b" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-array", "arrow-buffer", @@ -389,8 +377,7 @@ dependencies = [ [[package]] name = "arrow-row" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4d5a1f8c733d15260b305683472ee8ad89c62cbd706703ca873b90d051b41592" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-array", "arrow-buffer", @@ -402,8 +389,7 @@ dependencies = [ [[package]] name = "arrow-schema" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d9e4969dc350d571766247143ab36a5187d095d3d3690970408bc630d47c69e5" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "bitflags", "serde", @@ -414,8 +400,7 @@ dependencies = [ [[package]] name = "arrow-select" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "402770dba90865359d98d1ef92ef16e23d75c0cca9c2c880c8a05468b7743bf9" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "ahash", "arrow-array", @@ -428,8 +413,7 @@ dependencies = [ [[package]] name = "arrow-string" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2b0afbb8b9016700938291123df30838b89decc3213dba00852021988b170d3" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "arrow-array", "arrow-buffer", @@ -1007,7 +991,7 @@ checksum = "4d6867f1565b3aad85681f1015055b087fcfd840d6aeee6eee7f2da317603695" dependencies = [ "autocfg", "libm", - "num-bigint", + "num-bigint 0.4.6", "num-integer", "num-traits", ] @@ -4053,9 +4037,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] name = "lz4_flex" -version = "0.13.0" +version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "db9a0d582c2874f68138a16ce1867e0ffde6c0bb0a0df85e1f36d04146db488a" +checksum = "ecbdfe44b1bd960b68170b417450a628c43f7cf56bb3c5317e61cb230ee7f226" dependencies = [ "twox-hash", ] @@ -4198,7 +4182,7 @@ version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "35bd024e8b2ff75562e5f34e7f4905839deb4b22955ef5e73d2fea1b9813cb23" dependencies = [ - "num-bigint", + "num-bigint 0.4.6", "num-complex", "num-integer", "num-iter", @@ -4216,6 +4200,16 @@ dependencies = [ "num-traits", ] +[[package]] +name = "num-bigint" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0" +dependencies = [ + "num-integer", + "num-traits", +] + [[package]] name = "num-complex" version = "0.4.6" @@ -4257,7 +4251,7 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f83d14da390562dca69fc84082e73e548e1ad308d24accdedd2720017cb37824" dependencies = [ - "num-bigint", + "num-bigint 0.4.6", "num-integer", "num-traits", ] @@ -4464,8 +4458,7 @@ dependencies = [ [[package]] name = "parquet" version = "59.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5302d4da74d6596a1f11f9928767995b53bca657cbeea1e4e8c5074f8a1157dd" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" dependencies = [ "ahash", "arrow-array", @@ -4483,11 +4476,10 @@ dependencies = [ "half", "hashbrown 0.17.1", "lz4_flex", - "num-bigint", + "num-bigint 0.5.1", "num-integer", "num-traits", "object_store", - "paste", "ring", "seq-macro", "simdutf8", @@ -4522,12 +4514,6 @@ dependencies = [ "syn", ] -[[package]] -name = "paste" -version = "1.0.15" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" - [[package]] name = "pbjson" version = "0.8.0" diff --git a/Cargo.toml b/Cargo.toml index c10c9c16e890c..158da91519c9d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -294,3 +294,24 @@ debug = false debug-assertions = false strip = "debuginfo" incremental = false + +# TEMPORARY (remove before merge): build against the arrow-rs bounded streaming +# draft (apache/arrow-rs#10410) until it is released to crates.io. +[patch.crates-io] +arrow = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-arith = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-array = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-avro = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-buffer = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-cast = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-csv = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-data = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-flight = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-ipc = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-json = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-ord = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-row = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-schema = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-select = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +arrow-string = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } +parquet = { git = "https://github.com/pydantic/arrow-rs", branch = "bounded-column-streaming" } From a2b95df3f2fecdc8dce3664f91e16ea0ca7fc506 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 22 Jul 2026 12:04:54 -0500 Subject: [PATCH 3/5] Update arrow-rs pin: dictionary page cache across decode windows Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG --- Cargo.lock | 38 +++++++++++++++++++------------------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c1809ae5ff7ce..e16457997a49a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -165,7 +165,7 @@ checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" [[package]] name = "arrow" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-arith", "arrow-array", @@ -187,7 +187,7 @@ dependencies = [ [[package]] name = "arrow-arith" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-array", "arrow-buffer", @@ -200,7 +200,7 @@ dependencies = [ [[package]] name = "arrow-array" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "ahash", "arrow-buffer", @@ -219,7 +219,7 @@ dependencies = [ [[package]] name = "arrow-avro" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-array", "arrow-buffer", @@ -242,7 +242,7 @@ dependencies = [ [[package]] name = "arrow-buffer" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "bytes", "half", @@ -253,7 +253,7 @@ dependencies = [ [[package]] name = "arrow-cast" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-array", "arrow-buffer", @@ -274,7 +274,7 @@ dependencies = [ [[package]] name = "arrow-csv" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-array", "arrow-cast", @@ -288,7 +288,7 @@ dependencies = [ [[package]] name = "arrow-data" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-buffer", "arrow-schema", @@ -300,7 +300,7 @@ dependencies = [ [[package]] name = "arrow-flight" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-arith", "arrow-array", @@ -326,7 +326,7 @@ dependencies = [ [[package]] name = "arrow-ipc" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-array", "arrow-buffer", @@ -341,7 +341,7 @@ dependencies = [ [[package]] name = "arrow-json" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-array", "arrow-buffer", @@ -365,7 +365,7 @@ dependencies = [ [[package]] name = "arrow-ord" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-array", "arrow-buffer", @@ -377,7 +377,7 @@ dependencies = [ [[package]] name = "arrow-row" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-array", "arrow-buffer", @@ -389,7 +389,7 @@ dependencies = [ [[package]] name = "arrow-schema" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "bitflags", "serde", @@ -400,7 +400,7 @@ dependencies = [ [[package]] name = "arrow-select" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "ahash", "arrow-array", @@ -413,7 +413,7 @@ dependencies = [ [[package]] name = "arrow-string" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "arrow-array", "arrow-buffer", @@ -4458,7 +4458,7 @@ dependencies = [ [[package]] name = "parquet" version = "59.1.0" -source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#2c09b00a40040965cdc2a411970367bb90d2c39e" +source = "git+https://github.com/pydantic/arrow-rs?branch=bounded-column-streaming#7fc5501c34fec845f8a1e730f5b1e964b78acff6" dependencies = [ "ahash", "arrow-array", @@ -4837,7 +4837,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.14.0", + "itertools 0.13.0", "log", "multimap", "petgraph", @@ -4856,7 +4856,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.14.0", + "itertools 0.13.0", "proc-macro2", "quote", "syn", From 4a9f09d3b4cb14061494ef1c061a72e820d6b6f7 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 22 Jul 2026 12:15:22 -0500 Subject: [PATCH 4/5] Add bounded_streaming to remaining ParquetOptions test initializers Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG --- datafusion/common/src/file_options/parquet_writer.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/datafusion/common/src/file_options/parquet_writer.rs b/datafusion/common/src/file_options/parquet_writer.rs index 7e8c5e26c4c4a..4bfa40582ce40 100644 --- a/datafusion/common/src/file_options/parquet_writer.rs +++ b/datafusion/common/src/file_options/parquet_writer.rs @@ -495,6 +495,7 @@ mod tests { pushdown_filters: defaults.pushdown_filters, reorder_filters: defaults.reorder_filters, force_filter_selections: defaults.force_filter_selections, + bounded_streaming: defaults.bounded_streaming, allow_single_file_parallelism: defaults.allow_single_file_parallelism, maximum_parallel_row_group_writers: defaults .maximum_parallel_row_group_writers, @@ -614,6 +615,7 @@ mod tests { pushdown_filters: global_options_defaults.pushdown_filters, reorder_filters: global_options_defaults.reorder_filters, force_filter_selections: global_options_defaults.force_filter_selections, + bounded_streaming: global_options_defaults.bounded_streaming, allow_single_file_parallelism: global_options_defaults .allow_single_file_parallelism, maximum_parallel_row_group_writers: global_options_defaults From c69caf7e0064a74de3832cd22418d508630ce230 Mon Sep 17 00:00:00 2001 From: Adrian Garcia Badaracco <1755071+adriangb@users.noreply.github.com> Date: Wed, 22 Jul 2026 12:43:12 -0500 Subject: [PATCH 5/5] Fix clippy: box the AdaptiveFetcher variant, collapse nested if Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01MGnT9oETcFMydc9cp95HLG --- .../datasource-parquet/src/push_decoder.rs | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 3c68b6d2b5165..b280c7123873b 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -125,7 +125,7 @@ pub(crate) enum FetchState { /// window. See [`ParquetOptions::bounded_streaming`]. /// /// [`ParquetOptions::bounded_streaming`]: datafusion_common::config::ParquetOptions::bounded_streaming - Adaptive(AdaptiveFetcher>), + Adaptive(Box>>), } impl FetchState { @@ -133,14 +133,12 @@ impl FetchState { /// is available to drive streamed request bodies; without one, fall back /// to the plain path, which behaves identically to the option being off). pub(crate) fn new(reader: Box, bounded_streaming: bool) -> Self { - if bounded_streaming { - if let Ok(handle) = tokio::runtime::Handle::try_current() { - return Self::Adaptive(AdaptiveFetcher::new( - BoundedStreamingOptions::default(), - handle, - reader, - )); - } + if bounded_streaming && let Ok(handle) = tokio::runtime::Handle::try_current() { + return Self::Adaptive(Box::new(AdaptiveFetcher::new( + BoundedStreamingOptions::default(), + handle, + reader, + ))); } Self::Plain(reader) }