blackmwk commented on code in PR #3514:
URL: https://github.com/apache/iceberg-rust/pull/3514#discussion_r4511392788
##########
crates/iceberg/src/arrow/reader/mod.rs:
##########
@@ -52,114 +52,49 @@ 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;
+ /// 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
}
- /// 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 concurrency_limit_data_files = self
+ .options
+ .data_file_concurrency_limit()
+ .unwrap_or_else(|| available_parallelism().get())
+ .max(1);
Review Comment:
Also I would suggest to use `with_data_file_concurrency_limit_if_unset` so
that we could test it in ut.
##########
crates/iceberg/src/arrow/reader/mod.rs:
##########
@@ -52,114 +52,49 @@ 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;
+ /// 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
}
- /// 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 concurrency_limit_data_files = self
+ .options
+ .data_file_concurrency_limit()
+ .unwrap_or_else(|| available_parallelism().get())
+ .max(1);
Review Comment:
I think this should be min(1)?
--
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]
