feat(scan): expose portable reader options for planned tasks

crates/catalog/s3tables/src/catalog.rs

@@ -1345,7 +1345,7 @@ mod tests {
             .select_all()
             .build()
             .expect("scan to be valid (snapshot exists, schema is OK)")
-            .to_arrow()
+            .to_arrow(Default::default())
             .await
             .expect("scan tasks should be OK")
             .try_collect()

crates/examples/src/oss_backend.rs

@@ -81,7 +81,7 @@ async fn main() {
     println!("Table {TABLE_NAME} loaded!");
 
     let scan = table.scan().select_all().build().unwrap();
-    let reader = scan.to_arrow().await.unwrap();
+    let reader = scan.to_arrow(Default::default()).await.unwrap();
     let buf = reader.collect::<Vec<_>>().await;
 
     println!("Table {TABLE_NAME} has {} batches.", buf.len());

crates/iceberg/Cargo.toml

@@ -99,6 +99,7 @@ mockall = { workspace = true }
 pretty_assertions = { workspace = true }
 rand = { workspace = true }
 regex = { workspace = true }
+rstest = { workspace = true }
 serde_arrow = { version = "0.14", features = ["arrow-59"] }
 tempfile = { workspace = true }
 tracing-subscriber = { workspace = true }

crates/iceberg/README.md

@@ -56,7 +56,7 @@ async fn main() -> Result<()> {
         .scan()
         .select(["name", "id"])
         .build()?
-        .to_arrow()
+        .to_arrow(Default::default())
         .await?;
 
     // Consume this stream like arrow record batch stream.

crates/iceberg/public-api.txt

@@ -71,14 +71,54 @@ pub struct iceberg::arrow::ArrowReaderBuilder
 impl iceberg::arrow::ArrowReaderBuilder
 pub fn iceberg::arrow::ArrowReaderBuilder::build(self) -> iceberg::arrow::ArrowReader
 pub fn iceberg::arrow::ArrowReaderBuilder::new(file_io: iceberg::io::FileIO, runtime: iceberg::Runtime) -> Self
-pub fn iceberg::arrow::ArrowReaderBuilder::with_batch_size(self, batch_size: usize) -> Self
-pub fn iceberg::arrow::ArrowReaderBuilder::with_bloom_filter_enabled(self, bloom_filter_enabled: bool) -> Self
-pub fn iceberg::arrow::ArrowReaderBuilder::with_data_file_concurrency_limit(self, val: usize) -> Self
-pub fn iceberg::arrow::ArrowReaderBuilder::with_metadata_size_hint(self, metadata_size_hint: usize) -> Self
-pub fn iceberg::arrow::ArrowReaderBuilder::with_range_coalesce_bytes(self, range_coalesce_bytes: u64) -> Self
-pub fn iceberg::arrow::ArrowReaderBuilder::with_range_fetch_concurrency(self, range_fetch_concurrency: usize) -> Self
-pub fn iceberg::arrow::ArrowReaderBuilder::with_row_group_filtering_enabled(self, row_group_filtering_enabled: bool) -> Self
-pub fn iceberg::arrow::ArrowReaderBuilder::with_row_selection_enabled(self, row_selection_enabled: bool) -> Self
+pub fn iceberg::arrow::ArrowReaderBuilder::with_options(self, options: iceberg::arrow::ArrowReaderOptions) -> Self
+pub fn iceberg::arrow::ArrowReaderBuilder::with_parquet_options(self, options: iceberg::arrow::ParquetReadOptions) -> Self
+pub struct iceberg::arrow::ArrowReaderOptions
+impl iceberg::arrow::ArrowReaderOptions
+pub fn iceberg::arrow::ArrowReaderOptions::batch_size(&self) -> core::option::Option<usize>
+pub fn iceberg::arrow::ArrowReaderOptions::bloom_filter_enabled(&self) -> bool
+pub fn iceberg::arrow::ArrowReaderOptions::data_file_concurrency_limit(&self) -> core::option::Option<usize>
+pub fn iceberg::arrow::ArrowReaderOptions::parquet_options(&self) -> &iceberg::arrow::ParquetReadOptions
+pub fn iceberg::arrow::ArrowReaderOptions::row_group_filtering_enabled(&self) -> bool
+pub fn iceberg::arrow::ArrowReaderOptions::row_selection_enabled(&self) -> bool
+impl core::clone::Clone for iceberg::arrow::ArrowReaderOptions
+pub fn iceberg::arrow::ArrowReaderOptions::clone(&self) -> iceberg::arrow::ArrowReaderOptions
+impl core::cmp::PartialEq for iceberg::arrow::ArrowReaderOptions
+pub fn iceberg::arrow::ArrowReaderOptions::eq(&self, other: &iceberg::arrow::ArrowReaderOptions) -> bool
+impl core::default::Default for iceberg::arrow::ArrowReaderOptions
+pub fn iceberg::arrow::ArrowReaderOptions::default() -> Self
+impl core::fmt::Debug for iceberg::arrow::ArrowReaderOptions
+pub fn iceberg::arrow::ArrowReaderOptions::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl core::marker::StructuralPartialEq for iceberg::arrow::ArrowReaderOptions
+impl iceberg::arrow::ArrowReaderOptions
+pub fn iceberg::arrow::ArrowReaderOptions::builder() -> ArrowReaderOptionsBuilder<((), (), (), (), (), ())>
+impl serde_core::ser::Serialize for iceberg::arrow::ArrowReaderOptions
+pub fn iceberg::arrow::ArrowReaderOptions::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
+impl<'de> serde_core::de::Deserialize<'de> for iceberg::arrow::ArrowReaderOptions where iceberg::arrow::ArrowReaderOptions: core::default::Default
+pub fn iceberg::arrow::ArrowReaderOptions::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
+pub struct iceberg::arrow::ParquetReadOptions
+impl iceberg::arrow::ParquetReadOptions
+pub fn iceberg::arrow::ParquetReadOptions::metadata_size_hint(&self) -> usize
+pub fn iceberg::arrow::ParquetReadOptions::preload_column_index(&self) -> bool
+pub fn iceberg::arrow::ParquetReadOptions::preload_offset_index(&self) -> bool
+pub fn iceberg::arrow::ParquetReadOptions::preload_page_index(&self) -> bool
+pub fn iceberg::arrow::ParquetReadOptions::range_coalesce_bytes(&self) -> u64
+pub fn iceberg::arrow::ParquetReadOptions::range_fetch_concurrency(&self) -> usize
+impl core::clone::Clone for iceberg::arrow::ParquetReadOptions
+pub fn iceberg::arrow::ParquetReadOptions::clone(&self) -> iceberg::arrow::ParquetReadOptions
+impl core::cmp::PartialEq for iceberg::arrow::ParquetReadOptions
+pub fn iceberg::arrow::ParquetReadOptions::eq(&self, other: &iceberg::arrow::ParquetReadOptions) -> bool
+impl core::default::Default for iceberg::arrow::ParquetReadOptions
+pub fn iceberg::arrow::ParquetReadOptions::default() -> Self
+impl core::fmt::Debug for iceberg::arrow::ParquetReadOptions
+pub fn iceberg::arrow::ParquetReadOptions::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl core::marker::StructuralPartialEq for iceberg::arrow::ParquetReadOptions
+impl iceberg::arrow::ParquetReadOptions
+pub fn iceberg::arrow::ParquetReadOptions::builder() -> ParquetReadOptionsBuilder<((), (), (), (), (), ())>
+impl serde_core::ser::Serialize for iceberg::arrow::ParquetReadOptions
+pub fn iceberg::arrow::ParquetReadOptions::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
+impl<'de> serde_core::de::Deserialize<'de> for iceberg::arrow::ParquetReadOptions where iceberg::arrow::ParquetReadOptions: core::default::Default
+pub fn iceberg::arrow::ParquetReadOptions::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub struct iceberg::arrow::PartitionValueCalculator
 impl iceberg::arrow::PartitionValueCalculator
 pub fn iceberg::arrow::PartitionValueCalculator::calculate(&self, batch: &arrow_array::record_batch::RecordBatch) -> iceberg::Result<arrow_array::array::ArrayRef>
@@ -1437,7 +1477,7 @@ impl iceberg::scan::TableScan
 pub fn iceberg::scan::TableScan::column_names(&self) -> core::option::Option<&[alloc::string::String]>
 pub async fn iceberg::scan::TableScan::plan_files(&self) -> iceberg::Result<iceberg::scan::FileScanTaskStream>
 pub fn iceberg::scan::TableScan::snapshot(&self) -> core::option::Option<&iceberg::spec::SnapshotRef>
-pub async fn iceberg::scan::TableScan::to_arrow(&self) -> iceberg::Result<iceberg::scan::ArrowRecordBatchStream>
+pub async fn iceberg::scan::TableScan::to_arrow(&self, options: iceberg::arrow::ArrowReaderOptions) -> iceberg::Result<iceberg::scan::ArrowRecordBatchStream>
 impl core::fmt::Debug for iceberg::scan::TableScan
 pub fn iceberg::scan::TableScan::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
 pub struct iceberg::scan::TableScanBuilder<'a>
@@ -1448,15 +1488,10 @@ pub fn iceberg::scan::TableScanBuilder<'a>::select(self, column_names: impl core
 pub fn iceberg::scan::TableScanBuilder<'a>::select_all(self) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::select_empty(self) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::snapshot_id(self, snapshot_id: i64) -> Self
-pub fn iceberg::scan::TableScanBuilder<'a>::with_batch_size(self, batch_size: core::option::Option<usize>) -> Self
-pub fn iceberg::scan::TableScanBuilder<'a>::with_bloom_filter_enabled(self, bloom_filter_enabled: bool) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::with_case_sensitive(self, case_sensitive: bool) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::with_concurrency_limit(self, limit: usize) -> Self
-pub fn iceberg::scan::TableScanBuilder<'a>::with_data_file_concurrency_limit(self, limit: usize) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::with_filter(self, predicate: iceberg::expr::Predicate) -> Self
 pub fn iceberg::scan::TableScanBuilder<'a>::with_manifest_entry_concurrency_limit(self, limit: usize) -> Self
-pub fn iceberg::scan::TableScanBuilder<'a>::with_row_group_filtering_enabled(self, row_group_filtering_enabled: bool) -> Self
-pub fn iceberg::scan::TableScanBuilder<'a>::with_row_selection_enabled(self, row_selection_enabled: bool) -> Self
 pub type iceberg::scan::ArrowRecordBatchStream = futures_core::stream::BoxStream<'static, iceberg::Result<arrow_array::record_batch::RecordBatch>>
 pub type iceberg::scan::FileScanTaskStream = futures_core::stream::BoxStream<'static, iceberg::Result<iceberg::scan::FileScanTask>>
 pub mod iceberg::sensitive

crates/iceberg/src/arrow/reader/file_reader.rs

@@ -118,7 +118,7 @@ impl AsyncFileReader for ArrowFileReader {
 
         async move {
             let reader = ParquetMetaDataReader::new()
-                .with_prefetch_hint(self.parquet_read_options.metadata_size_hint())
+                .with_prefetch_hint(Some(self.parquet_read_options.metadata_size_hint()))
                 // Set the page policy first because it updates both column and offset policies.
                 .with_page_index_policy(PageIndexPolicy::from(
                     self.parquet_read_options.preload_page_index(),

crates/iceberg/src/arrow/reader/mod.rs

@@ -43,7 +43,7 @@ mod projection;
 mod row_filter;
 mod row_lineage;
 pub use file_reader::ArrowFileReader;
-pub(crate) use options::ParquetReadOptions;
+pub use options::{ArrowReaderOptions, ParquetReadOptions};
 use predicate_visitor::{CollectFieldIdVisitor, PredicateConverter};
 use projection::{
     add_fallback_field_ids_to_arrow_schema, apply_name_mapping_to_arrow_schema,
@@ -52,114 +52,50 @@ use projection::{
 
 /// Builder to create ArrowReader
 pub struct ArrowReaderBuilder {
-    batch_size: Option<usize>,
+    options: ArrowReaderOptions,
     file_io: FileIO,
-    concurrency_limit_data_files: usize,
-    row_group_filtering_enabled: bool,
-    row_selection_enabled: bool,
-    bloom_filter_enabled: bool,
-    parquet_read_options: ParquetReadOptions,
     runtime: Runtime,
 }
 
 impl ArrowReaderBuilder {
     /// Create a new ArrowReaderBuilder
     pub fn new(file_io: FileIO, runtime: Runtime) -> Self {
-        let num_cpus = available_parallelism().get();
-
         ArrowReaderBuilder {
-            batch_size: None,
+            options: ArrowReaderOptions::default(),
             file_io,
-            concurrency_limit_data_files: num_cpus,
-            row_group_filtering_enabled: true,
-            row_selection_enabled: false,
-            bloom_filter_enabled: false,
-            parquet_read_options: ParquetReadOptions::builder().build(),
             runtime,
         }
     }
 
-    /// Sets the max number of in flight data files that are being fetched
-    pub fn with_data_file_concurrency_limit(mut self, val: usize) -> Self {
-        self.concurrency_limit_data_files = val;
-        self
-    }
-
-    /// Sets the desired size of batches in the response
-    /// to something other than the default
-    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
-        self.batch_size = Some(batch_size);
-        self
-    }
-
-    /// Determines whether to enable row group filtering.
-    pub fn with_row_group_filtering_enabled(mut self, row_group_filtering_enabled: bool) -> Self {
-        self.row_group_filtering_enabled = row_group_filtering_enabled;
+    /// Replaces all reader settings while preserving local I/O and runtime resources.
+    pub fn with_options(mut self, options: ArrowReaderOptions) -> Self {
+        self.options = options;
         self
     }
 
-    /// Determines whether to enable row selection.
-    pub fn with_row_selection_enabled(mut self, row_selection_enabled: bool) -> Self {
-        self.row_selection_enabled = row_selection_enabled;
-        self
-    }
-
-    /// Determines whether to enable bloom filter-based row group filtering.
-    ///
-    /// When enabled, if a read is performed with an equality or IN predicate,
-    /// the bloom filter for relevant columns in each row group is read and
-    /// checked. Row groups where the bloom filter proves the value is absent
-    /// are skipped entirely.
-    ///
-    /// Defaults to disabled. Each bloom filter is a separate read, and they are
-    /// issued serially — one round trip per relevant column per row group, before
-    /// any data is read. TODO(#3191)
-    pub fn with_bloom_filter_enabled(mut self, bloom_filter_enabled: bool) -> Self {
-        self.bloom_filter_enabled = bloom_filter_enabled;
-        self
-    }
-
-    /// Provide a hint as to the number of bytes to prefetch for parsing the Parquet metadata
-    ///
-    /// This hint can help reduce the number of fetch requests. For more details see the
-    /// [ParquetMetaDataReader documentation](https://docs.rs/parquet/latest/parquet/file/metadata/struct.ParquetMetaDataReader.html#method.with_prefetch_hint).
-    pub fn with_metadata_size_hint(mut self, metadata_size_hint: usize) -> Self {
-        self.parquet_read_options.metadata_size_hint = Some(metadata_size_hint);
-        self
-    }
-
-    /// Sets the gap threshold for merging nearby byte ranges into a single request.
-    /// Ranges with gaps smaller than this value will be coalesced.
-    ///
-    /// Defaults to 1 MiB, matching object_store's OBJECT_STORE_COALESCE_DEFAULT.
-    pub fn with_range_coalesce_bytes(mut self, range_coalesce_bytes: u64) -> Self {
-        self.parquet_read_options.range_coalesce_bytes = range_coalesce_bytes;
-        self
-    }
-
-    /// Sets the maximum number of merged byte ranges to fetch concurrently.
-    ///
-    /// Defaults to 10, matching object_store's OBJECT_STORE_COALESCE_PARALLEL.
-    pub fn with_range_fetch_concurrency(mut self, range_fetch_concurrency: usize) -> Self {
-        self.parquet_read_options.range_fetch_concurrency = range_fetch_concurrency;
+    /// Replaces Parquet I/O settings while preserving other reader settings.
+    pub fn with_parquet_options(mut self, options: ParquetReadOptions) -> Self {
+        self.options = self.options.with_parquet_options(options);
         self
     }
 
     /// Build the ArrowReader.
     pub fn build(self) -> ArrowReader {
+        let options = self
+            .options
+            .with_data_file_concurrency_limit_if_unset(available_parallelism().get().max(1));
+        let concurrency_limit_data_files = options
+            .data_file_concurrency_limit()
+            .expect("reader concurrency was set above")
+            .max(1);
         ArrowReader {
-            batch_size: self.batch_size,
+            options,
             file_io: self.file_io.clone(),
             delete_file_loader: CachingDeleteFileLoader::new(
                 self.file_io.clone(),
-                self.concurrency_limit_data_files,
+                concurrency_limit_data_files,
                 self.runtime.clone(),
             ),
-            concurrency_limit_data_files: self.concurrency_limit_data_files,
-            row_group_filtering_enabled: self.row_group_filtering_enabled,
-            row_selection_enabled: self.row_selection_enabled,
-            bloom_filter_enabled: self.bloom_filter_enabled,
-            parquet_read_options: self.parquet_read_options,
             runtime: self.runtime.clone(),
         }
     }
@@ -168,16 +104,90 @@ impl ArrowReaderBuilder {
 /// Reads data from Parquet files
 #[derive(Clone)]
 pub struct ArrowReader {
-    batch_size: Option<usize>,
+    options: ArrowReaderOptions,
     file_io: FileIO,
     delete_file_loader: CachingDeleteFileLoader,
 
-    /// the maximum number of data files that can be fetched at the same time
-    concurrency_limit_data_files: usize,
-
-    row_group_filtering_enabled: bool,
-    row_selection_enabled: bool,
-    bloom_filter_enabled: bool,
-    parquet_read_options: ParquetReadOptions,
     runtime: Runtime,
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use futures::{TryStreamExt, stream};
+
+    use super::{ArrowReaderBuilder, ArrowReaderOptions, ParquetReadOptions};
+    use crate::io::FileIO;
+    use crate::test_utils::test_runtime;
+    use crate::util::available_parallelism;
+
+    #[tokio::test]
+    async fn deserialized_zero_concurrency_completes_reads() {
+        let options: ArrowReaderOptions =
+            serde_json::from_str(r#"{"data_file_concurrency_limit":0}"#).unwrap();
+        let reader = ArrowReaderBuilder::new(FileIO::new_with_memory(), test_runtime())
+            .with_options(options)
+            .build();
+        let batches = tokio::time::timeout(
+            Duration::from_secs(1),
+            reader
+                .read(Box::pin(stream::empty()))
+                .unwrap()
+                .stream()
+                .try_collect::<Vec<_>>(),
+        )
+        .await
+        .expect("zero concurrency must not stall the task stream")
+        .unwrap();
+        assert!(batches.is_empty());
+    }
+
+    #[test]
+    fn default_options_restore_local_reader_defaults() {
+        let reader = ArrowReaderBuilder::new(FileIO::new_with_memory(), test_runtime())
+            .with_options(
+                ArrowReaderOptions::builder()
+                    .with_batch_size(1024)
+                    .with_data_file_concurrency_limit(3)
+                    .with_row_group_filtering_enabled(false)
+                    .with_row_selection_enabled(true)
+                    .with_bloom_filter_enabled(true)
+                    .with_parquet_options(
+                        ParquetReadOptions::builder()
+                            .with_range_fetch_concurrency(99)
+                            .build(),
+                    )
+                    .build(),
+            )
+            .with_options(ArrowReaderOptions::default())
+            .build();
+        assert_eq!(
+            reader.options,
+            ArrowReaderOptions::builder()
+                .with_data_file_concurrency_limit(available_parallelism().get())
+                .build()
+        );
+    }
+
+    #[test]
+    fn explicit_options_configure_reader_settings() {
+        let parquet = ParquetReadOptions::builder()
+            .with_metadata_size_hint(4096)
+            .with_range_coalesce_bytes(8192)
+            .with_range_fetch_concurrency(2)
+            .build();
+        let options = ArrowReaderOptions::builder()
+            .with_batch_size(17)
+            .with_data_file_concurrency_limit(3)
+            .with_row_group_filtering_enabled(false)
+            .with_row_selection_enabled(true)
+            .with_bloom_filter_enabled(true)
+            .with_parquet_options(parquet)
+            .build();
+        let reader = ArrowReaderBuilder::new(FileIO::new_with_memory(), test_runtime())
+            .with_options(options.clone())
+            .build();
+        assert_eq!(reader.options, options);
+    }
+}