diff --git a/Cargo.lock b/Cargo.lock index 6f776e5e965f9..e16457997a49a 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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#7fc5501c34fec845f8a1e730f5b1e964b78acff6" 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" @@ -4851,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", @@ -4870,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", 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" } 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..4bfa40582ce40 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: _, @@ -494,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, @@ -613,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 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..b280c7123873b 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,43 @@ 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(Box>>), +} + +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 && let Ok(handle) = tokio::runtime::Handle::try_current() { + return Self::Adaptive(Box::new(AdaptiveFetcher::new( + BoundedStreamingOptions::default(), + handle, + reader, + ))); + } + Self::Plain(reader) + } +} + #[derive(Debug, Clone)] pub(crate) struct RgPlanEntry { pub(crate) rg_index: usize, @@ -243,7 +281,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 +411,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 +436,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" |