blackmwk commented on code in PR #3514:
URL: https://github.com/apache/iceberg-rust/pull/3514#discussion_r4465474987
##########
crates/iceberg/src/arrow/reader/options.rs:
##########
@@ -15,70 +15,234 @@
// specific language governing permissions and limitations
// under the License.
-//! Tunables for Parquet file I/O used by `ArrowReader`.
-
+use serde::{Deserialize, Serialize};
use typed_builder::TypedBuilder;
use super::{
DEFAULT_METADATA_SIZE_HINT, DEFAULT_RANGE_COALESCE_BYTES,
DEFAULT_RANGE_FETCH_CONCURRENCY,
};
+/// Configures Arrow reads independently of file planning and local I/O
resources.
+///
+/// Serializable alongside planned file tasks so readers can be constructed
later.
+/// In self-describing map formats such as JSON, field names are part of the
+/// serialized format: missing fields use defaults and unknown fields are
ignored.
+/// Compatibility across different crate versions is not otherwise guaranteed.
+///
+/// ```no_run
+/// # use iceberg::{Result, arrow::{ArrowReaderOptions, ParquetReadOptions},
table::Table};
+/// # async fn read(table: &Table) -> Result<()> {
+/// let options = ArrowReaderOptions::builder()
+/// .with_batch_size(1024)
+/// .with_parquet_options(
+/// ParquetReadOptions::builder()
+/// .with_metadata_size_hint(512 * 1024)
+/// .build(),
+/// )
+/// .build();
+/// let stream = table.scan().build()?.to_arrow(options).await?;
+/// # Ok(())
+/// # }
+/// ```
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TypedBuilder)]
+#[serde(default)]
+#[builder(field_defaults(default, setter(prefix = "with_")))]
+pub struct ArrowReaderOptions {
+ /// Desired batch size, or the reader default when unset.
+ #[builder(default = None, setter(strip_option))]
+ batch_size: Option<usize>,
+ /// Maximum in-flight data files, or the construction path's default when
unset.
+ #[builder(default = None, setter(strip_option))]
+ data_file_concurrency_limit: Option<usize>,
+ /// Whether to prune row groups using their statistics.
+ #[builder(default = true)]
+ row_group_filtering_enabled: bool,
+ /// Whether to use page indexes for row selection.
+ row_selection_enabled: bool,
+ /// Whether to use bloom filters for row-group pruning.
+ bloom_filter_enabled: bool,
+ /// I/O tuning applied when reading Parquet files.
+ parquet_options: ParquetReadOptions,
+}
+
+impl Default for ArrowReaderOptions {
+ fn default() -> Self {
+ Self::builder().build()
+ }
+}
+
+impl ArrowReaderOptions {
+ /// Supplies the scan's fallback limit without replacing an explicit
reader limit.
+ pub(crate) fn with_default_data_file_concurrency_limit(mut self, limit:
usize) -> Self {
+ self.data_file_concurrency_limit.get_or_insert(limit);
+ self
+ }
+
+ /// Returns the configured batch size.
+ pub fn batch_size(&self) -> Option<usize> {
+ self.batch_size
+ }
+
+ /// Returns the configured data file concurrency limit.
+ ///
+ /// When unset, `TableScan::to_arrow` inherits the scan's data-file limit,
while
+ /// direct reader construction uses execution-host parallelism. Zero is
treated as one.
+ pub fn data_file_concurrency_limit(&self) -> Option<usize> {
+ self.data_file_concurrency_limit
+ }
+
+ /// Returns whether row-group statistics filtering is enabled.
+ ///
+ /// Enabled by default: statistics can skip groups that cannot match the
+ /// predicate, generally improving performance with little overhead.
+ pub fn row_group_filtering_enabled(&self) -> bool {
+ self.row_group_filtering_enabled
+ }
+
+ /// Returns whether page-index row selection is enabled.
+ ///
+ /// Disabled by default: parsing page indexes can cost more than the reads
+ /// it avoids. Evaluate it against the table's sorting, partitioning,
row-group
+ /// size, and page size before enabling it.
+ pub fn row_selection_enabled(&self) -> bool {
+ self.row_selection_enabled
+ }
+
+ /// Returns whether bloom-filter pruning is enabled.
+ ///
+ /// Disabled by default: bloom filters require extra reads per relevant
+ /// column and row group. They can eliminate groups for equality and IN
predicates.
+ pub fn bloom_filter_enabled(&self) -> bool {
+ self.bloom_filter_enabled
+ }
+
+ /// Returns the configured parquet options.
+ pub fn parquet_options(&self) -> &ParquetReadOptions {
+ &self.parquet_options
+ }
+
+ pub(super) fn set_parquet_options(&mut self, value: ParquetReadOptions) {
+ self.parquet_options = value;
+ }
+}
+
/// Options for tuning Parquet file I/O.
-#[derive(Clone, Copy, Debug, TypedBuilder)]
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, TypedBuilder)]
+#[serde(default)]
#[builder(field_defaults(setter(prefix = "with_")))]
-pub(crate) struct ParquetReadOptions {
+pub struct ParquetReadOptions {
/// 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).
///
/// Defaults to 512 KiB, matching DataFusion's default
`ParquetOptions::metadata_size_hint`.
- #[builder(default = Some(DEFAULT_METADATA_SIZE_HINT))]
- pub(crate) metadata_size_hint: Option<usize>,
+ #[builder(default = DEFAULT_METADATA_SIZE_HINT)]
+ metadata_size_hint: usize,
/// 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`.
#[builder(default = DEFAULT_RANGE_COALESCE_BYTES)]
- pub(crate) range_coalesce_bytes: u64,
+ range_coalesce_bytes: u64,
/// Maximum number of merged byte ranges to fetch concurrently.
///
/// Defaults to 10, matching object_store's
`OBJECT_STORE_COALESCE_PARALLEL`.
#[builder(default = DEFAULT_RANGE_FETCH_CONCURRENCY)]
- pub(crate) range_fetch_concurrency: usize,
- /// Whether to preload the column index when reading Parquet metadata.
- #[builder(default = true)]
- pub(crate) preload_column_index: bool,
Review Comment:
Removing them is not a behavior change, but I still perfer to keep them here
rather seeing them changed in ArrowFileReader.
##########
crates/iceberg/src/arrow/reader/options.rs:
##########
@@ -15,70 +15,333 @@
// specific language governing permissions and limitations
// under the License.
-//! Tunables for Parquet file I/O used by `ArrowReader`.
-
+use serde::{Deserialize, Serialize};
use typed_builder::TypedBuilder;
use super::{
DEFAULT_METADATA_SIZE_HINT, DEFAULT_RANGE_COALESCE_BYTES,
DEFAULT_RANGE_FETCH_CONCURRENCY,
};
+/// Configures Arrow reads independently of file planning and local I/O
resources.
+///
+/// Serializable alongside planned file tasks so readers can be constructed
later.
+/// In self-describing map formats such as JSON, field names are part of the
+/// serialized format: missing fields use defaults and unknown fields are
ignored.
+/// Compatibility across different crate versions is not otherwise guaranteed.
+///
+/// ```no_run
+/// # use iceberg::{Result, arrow::{ArrowReaderOptions, ParquetReadOptions},
table::Table};
+/// # async fn read(table: &Table) -> Result<()> {
+/// let options = ArrowReaderOptions::builder()
+/// .with_batch_size(1024)
+/// .with_parquet_options(
+/// ParquetReadOptions::builder()
+/// .with_metadata_size_hint(512 * 1024)
+/// .build(),
+/// )
+/// .build();
+/// let stream = table.scan().build()?.to_arrow(options).await?;
+/// # Ok(())
+/// # }
+/// ```
+#[derive(Debug, Clone, Serialize, Deserialize, TypedBuilder)]
+#[cfg_attr(test, derive(PartialEq))]
+#[serde(default)]
+#[builder(field_defaults(default, setter(prefix = "with_")))]
+pub struct ArrowReaderOptions {
+ /// Desired batch size, or the reader default when unset.
+ #[builder(default = None, setter(strip_option))]
+ batch_size: Option<usize>,
+ /// Maximum in-flight data files, or the construction path's default when
unset.
+ #[builder(default = None, setter(strip_option))]
+ data_file_concurrency_limit: Option<usize>,
+ /// Whether to prune row groups using their statistics.
+ #[builder(default = true)]
+ row_group_filtering_enabled: bool,
+ /// Whether to use page indexes for row selection.
+ row_selection_enabled: bool,
+ /// Whether to use bloom filters for row-group pruning.
+ bloom_filter_enabled: bool,
+ /// I/O tuning applied when reading Parquet files.
+ parquet_options: ParquetReadOptions,
+}
+
+impl Default for ArrowReaderOptions {
+ fn default() -> Self {
+ Self::builder().build()
+ }
+}
+
+impl ArrowReaderOptions {
+ pub(crate) fn with_default_data_file_concurrency_limit(mut self, limit:
usize) -> Self {
Review Comment:
Sorry, I mean, it should be `with_data_file_concurrency_limit_if_unset`
##########
crates/iceberg/src/arrow/reader/tests.rs:
##########
Review Comment:
I don't see this a good practice to put them in a separate file, they are
uts and please put then along side with codes.
##########
crates/iceberg/src/arrow/reader/mod.rs:
##########
@@ -52,114 +54,53 @@ 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;
- 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;
+ /// 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
}
- /// 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.set_parquet_options(options);
self
}
/// Build the ArrowReader.
pub fn build(self) -> ArrowReader {
+ let concurrency_limit_data_files = self
+ .options
+ .data_file_concurrency_limit()
+ .unwrap_or_else(|| available_parallelism().get())
+ .max(1);
ArrowReader {
- batch_size: self.batch_size,
+ batch_size: self.options.batch_size(),
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,
+ concurrency_limit_data_files,
+ row_group_filtering_enabled:
self.options.row_group_filtering_enabled(),
+ row_selection_enabled: self.options.row_selection_enabled(),
+ bloom_filter_enabled: self.options.bloom_filter_enabled(),
+ parquet_read_options: self.options.parquet_options().clone(),
Review Comment:
I think we no longer keep these fields in ArrowReader, instead just keep the
ArrowReaderOptions in it? If you don't want to do it here, create an issue to
track it.
##########
crates/iceberg/src/arrow/reader/options.rs:
##########
@@ -15,70 +15,234 @@
// specific language governing permissions and limitations
// under the License.
-//! Tunables for Parquet file I/O used by `ArrowReader`.
-
+use serde::{Deserialize, Serialize};
use typed_builder::TypedBuilder;
use super::{
DEFAULT_METADATA_SIZE_HINT, DEFAULT_RANGE_COALESCE_BYTES,
DEFAULT_RANGE_FETCH_CONCURRENCY,
};
+/// Configures Arrow reads independently of file planning and local I/O
resources.
+///
+/// Serializable alongside planned file tasks so readers can be constructed
later.
+/// In self-describing map formats such as JSON, field names are part of the
+/// serialized format: missing fields use defaults and unknown fields are
ignored.
+/// Compatibility across different crate versions is not otherwise guaranteed.
+///
+/// ```no_run
+/// # use iceberg::{Result, arrow::{ArrowReaderOptions, ParquetReadOptions},
table::Table};
+/// # async fn read(table: &Table) -> Result<()> {
+/// let options = ArrowReaderOptions::builder()
+/// .with_batch_size(1024)
+/// .with_parquet_options(
+/// ParquetReadOptions::builder()
+/// .with_metadata_size_hint(512 * 1024)
+/// .build(),
+/// )
+/// .build();
+/// let stream = table.scan().build()?.to_arrow(options).await?;
+/// # Ok(())
+/// # }
+/// ```
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, TypedBuilder)]
+#[serde(default)]
+#[builder(field_defaults(default, setter(prefix = "with_")))]
+pub struct ArrowReaderOptions {
+ /// Desired batch size, or the reader default when unset.
+ #[builder(default = None, setter(strip_option))]
+ batch_size: Option<usize>,
+ /// Maximum in-flight data files, or the construction path's default when
unset.
+ #[builder(default = None, setter(strip_option))]
+ data_file_concurrency_limit: Option<usize>,
+ /// Whether to prune row groups using their statistics.
+ #[builder(default = true)]
+ row_group_filtering_enabled: bool,
+ /// Whether to use page indexes for row selection.
+ row_selection_enabled: bool,
+ /// Whether to use bloom filters for row-group pruning.
+ bloom_filter_enabled: bool,
+ /// I/O tuning applied when reading Parquet files.
+ parquet_options: ParquetReadOptions,
+}
+
+impl Default for ArrowReaderOptions {
+ fn default() -> Self {
+ Self::builder().build()
+ }
+}
+
+impl ArrowReaderOptions {
+ /// Supplies the scan's fallback limit without replacing an explicit
reader limit.
+ pub(crate) fn with_default_data_file_concurrency_limit(mut self, limit:
usize) -> Self {
+ self.data_file_concurrency_limit.get_or_insert(limit);
+ self
+ }
+
+ /// Returns the configured batch size.
+ pub fn batch_size(&self) -> Option<usize> {
+ self.batch_size
+ }
+
+ /// Returns the configured data file concurrency limit.
+ ///
+ /// When unset, `TableScan::to_arrow` inherits the scan's data-file limit,
while
+ /// direct reader construction uses execution-host parallelism. Zero is
treated as one.
+ pub fn data_file_concurrency_limit(&self) -> Option<usize> {
+ self.data_file_concurrency_limit
+ }
+
+ /// Returns whether row-group statistics filtering is enabled.
+ ///
+ /// Enabled by default: statistics can skip groups that cannot match the
+ /// predicate, generally improving performance with little overhead.
+ pub fn row_group_filtering_enabled(&self) -> bool {
+ self.row_group_filtering_enabled
+ }
+
+ /// Returns whether page-index row selection is enabled.
+ ///
+ /// Disabled by default: parsing page indexes can cost more than the reads
+ /// it avoids. Evaluate it against the table's sorting, partitioning,
row-group
+ /// size, and page size before enabling it.
+ pub fn row_selection_enabled(&self) -> bool {
+ self.row_selection_enabled
+ }
+
+ /// Returns whether bloom-filter pruning is enabled.
+ ///
+ /// Disabled by default: bloom filters require extra reads per relevant
+ /// column and row group. They can eliminate groups for equality and IN
predicates.
+ pub fn bloom_filter_enabled(&self) -> bool {
+ self.bloom_filter_enabled
+ }
+
+ /// Returns the configured parquet options.
+ pub fn parquet_options(&self) -> &ParquetReadOptions {
+ &self.parquet_options
+ }
+
+ pub(super) fn set_parquet_options(&mut self, value: ParquetReadOptions) {
Review Comment:
```suggestion
pub(super) fn with_parquet_options(mut self, value: ParquetReadOptions)
-> Self {
```
Consistent with other fluent api.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
