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

crates/catalog/s3tables/src/catalog.rs

@@ -1329,7 +1329,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/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.
@@ -65,6 +65,36 @@ async fn main() -> Result<()> {
 }
 ```
 
+### Reader settings
+
+Pass reader settings separately from file planning:
+
+```rust
+let options = iceberg::arrow::ArrowReaderOptions::builder()
+    .with_batch_size(Some(1024))
+    .with_row_selection_enabled(true)
+    .with_parquet_options(
+        iceberg::arrow::ParquetReadOptions::builder()
+            .with_metadata_size_hint(Some(512 * 1024))
+            .build(),
+    )
+    .build();
+let stream = table.scan().build()?.to_arrow(options).await?;
+```
+
+When upgrading, move reader-setting calls from `TableScanBuilder` to
+`ArrowReaderOptions::builder()` and pass the result to `to_arrow(options)`.
+Use `to_arrow(Default::default())` for default settings. Scan
+`with_concurrency_limit()` controls manifest planning only; set
+`with_data_file_concurrency_limit(Some(limit))` on the reader options to also
+limit data-file reads. An unset data-file limit uses execution-host parallelism.
+
+Locally constructed readers accept the same options through
+`ArrowReaderBuilder::with_options(options)`, including Parquet I/O tuning.
+This replaces the complete configuration; subsequent convenience setters override
+individual settings. `with_parquet_options()` replaces only the nested Parquet
+configuration.
+
 ## Storage Backends
 
 For storage backend support (S3, GCS, local filesystem, etc.), use the [`iceberg-storage-opendal`](https://crates.io/crates/iceberg-storage-opendal) crate. See its [README](../storage/opendal/README.md) for available backends and feature flags.

crates/iceberg/public-api.txt

@@ -75,10 +75,45 @@ pub fn iceberg::arrow::ArrowReaderBuilder::with_batch_size(self, batch_size: usi
 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_options(self, options: iceberg::arrow::ArrowReaderOptions) -> Self
+pub fn iceberg::arrow::ArrowReaderBuilder::with_parquet_options(self, options: iceberg::arrow::ParquetReadOptions) -> 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 struct iceberg::arrow::ArrowReaderOptions
+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 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::Copy for iceberg::arrow::ParquetReadOptions
+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 +1472,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 +1483,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/mod.rs

@@ -19,8 +19,6 @@
 
 use crate::arrow::caching_delete_file_loader::CachingDeleteFileLoader;
 use crate::io::FileIO;
-use crate::runtime::Runtime;
-use crate::util::available_parallelism;
 
 /// Default gap between byte ranges below which they are coalesced into a
 /// single request. Matches object_store's `OBJECT_STORE_COALESCE_DEFAULT`.
@@ -36,134 +34,22 @@ const DEFAULT_METADATA_SIZE_HINT: usize = 512 * 1024;
 
 mod file_reader;
 mod options;
+mod parquet_options;
 mod pipeline;
 mod positional_deletes;
 mod predicate_visitor;
 mod projection;
 mod row_filter;
 mod row_lineage;
 pub use file_reader::ArrowFileReader;
-pub(crate) use options::ParquetReadOptions;
+pub use options::{ArrowReaderBuilder, ArrowReaderOptions, ArrowReaderOptionsBuilder};
+pub use parquet_options::{ParquetReadOptions, ParquetReadOptionsBuilder};
 use predicate_visitor::{CollectFieldIdVisitor, PredicateConverter};
 use projection::{
     add_fallback_field_ids_to_arrow_schema, apply_name_mapping_to_arrow_schema,
     find_leaf_by_field_id,
 };
 
-/// Builder to create ArrowReader
-pub struct ArrowReaderBuilder {
-    batch_size: Option<usize>,
-    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,
-            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;
-        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;
-        self
-    }
-
-    /// Build the ArrowReader.
-    pub fn build(self) -> ArrowReader {
-        ArrowReader {
-            batch_size: self.batch_size,
-            file_io: self.file_io.clone(),
-            delete_file_loader: CachingDeleteFileLoader::new(
-                self.file_io.clone(),
-                self.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,
-        }
-    }
-}
-
 /// Reads data from Parquet files
 #[derive(Clone)]
 pub struct ArrowReader {