blackmwk commented on code in PR #3514:
URL: https://github.com/apache/iceberg-rust/pull/3514#discussion_r4440656409
##########
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 {
+ 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, Serialize, Deserialize, TypedBuilder)]
+#[cfg_attr(test, derive(PartialEq))]
Review Comment:
Ditto.
##########
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 {
+ 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, Serialize, Deserialize, TypedBuilder)]
+#[cfg_attr(test, derive(PartialEq))]
+#[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,
- /// Whether to preload the offset index when reading Parquet metadata.
- #[builder(default = true)]
- pub(crate) preload_offset_index: bool,
- /// Whether to preload the page index when reading Parquet metadata.
- #[builder(default = false)]
- pub(crate) preload_page_index: bool,
+ range_fetch_concurrency: usize,
+}
+
+impl Default for ParquetReadOptions {
+ fn default() -> Self {
+ Self::builder().build()
+ }
}
impl ParquetReadOptions {
- pub(crate) fn metadata_size_hint(&self) -> Option<usize> {
+ /// Returns the configured metadata size hint.
+ pub fn metadata_size_hint(&self) -> usize {
self.metadata_size_hint
}
- pub(crate) fn range_coalesce_bytes(&self) -> u64 {
+ /// Returns the byte-gap threshold for coalescing read requests.
+ pub fn range_coalesce_bytes(&self) -> u64 {
self.range_coalesce_bytes
}
- pub(crate) fn range_fetch_concurrency(&self) -> usize {
+ /// Returns the maximum number of concurrent range requests.
+ pub fn range_fetch_concurrency(&self) -> usize {
self.range_fetch_concurrency
}
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use futures::{TryStreamExt, stream};
+
+ use super::*;
+ use crate::arrow::ArrowReaderBuilder;
+ use crate::io::FileIO;
+ use crate::test_utils::test_runtime;
+ use crate::util::available_parallelism;
+
+ #[tokio::test]
+ async fn deserialized_zero_concurrency_completes_reads() {
Review Comment:
As ut of this mod, we should focus on the options's behavior. This should
should be part of ArrowReader.
##########
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:
Add doc for this method. Also the function name is confusing, I would
suggest `with_default_data_file_concurrency_limit`
##########
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))]
Review Comment:
```suggestion
#[derive(PartialEq)]
```
I'm fine with deriving PartialEq
##########
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 {
+ 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, Serialize, Deserialize, TypedBuilder)]
+#[cfg_attr(test, derive(PartialEq))]
+#[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,
- /// Whether to preload the offset index when reading Parquet metadata.
- #[builder(default = true)]
- pub(crate) preload_offset_index: bool,
- /// Whether to preload the page index when reading Parquet metadata.
- #[builder(default = false)]
- pub(crate) preload_page_index: bool,
+ range_fetch_concurrency: usize,
+}
+
+impl Default for ParquetReadOptions {
+ fn default() -> Self {
+ Self::builder().build()
+ }
}
impl ParquetReadOptions {
- pub(crate) fn metadata_size_hint(&self) -> Option<usize> {
+ /// Returns the configured metadata size hint.
+ pub fn metadata_size_hint(&self) -> usize {
self.metadata_size_hint
}
- pub(crate) fn range_coalesce_bytes(&self) -> u64 {
+ /// Returns the byte-gap threshold for coalescing read requests.
+ pub fn range_coalesce_bytes(&self) -> u64 {
self.range_coalesce_bytes
}
- pub(crate) fn range_fetch_concurrency(&self) -> usize {
+ /// Returns the maximum number of concurrent range requests.
+ pub fn range_fetch_concurrency(&self) -> usize {
self.range_fetch_concurrency
}
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use futures::{TryStreamExt, stream};
+
+ use super::*;
+ use crate::arrow::ArrowReaderBuilder;
+ 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();
+ assert_eq!(reader.concurrency_limit_data_files, 1);
+ 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]
Review Comment:
Use rstest to simplify this. Also add doc to each step.
##########
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 {
+ 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, Serialize, Deserialize, TypedBuilder)]
+#[cfg_attr(test, derive(PartialEq))]
+#[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,
- /// Whether to preload the offset index when reading Parquet metadata.
- #[builder(default = true)]
- pub(crate) preload_offset_index: bool,
- /// Whether to preload the page index when reading Parquet metadata.
- #[builder(default = false)]
- pub(crate) preload_page_index: bool,
+ range_fetch_concurrency: usize,
+}
+
+impl Default for ParquetReadOptions {
+ fn default() -> Self {
+ Self::builder().build()
+ }
}
impl ParquetReadOptions {
- pub(crate) fn metadata_size_hint(&self) -> Option<usize> {
+ /// Returns the configured metadata size hint.
+ pub fn metadata_size_hint(&self) -> usize {
self.metadata_size_hint
}
- pub(crate) fn range_coalesce_bytes(&self) -> u64 {
+ /// Returns the byte-gap threshold for coalescing read requests.
+ pub fn range_coalesce_bytes(&self) -> u64 {
self.range_coalesce_bytes
}
- pub(crate) fn range_fetch_concurrency(&self) -> usize {
+ /// Returns the maximum number of concurrent range requests.
+ pub fn range_fetch_concurrency(&self) -> usize {
self.range_fetch_concurrency
}
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use futures::{TryStreamExt, stream};
+
+ use super::*;
+ use crate::arrow::ArrowReaderBuilder;
+ 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();
+ assert_eq!(reader.concurrency_limit_data_files, 1);
+ 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_data_file_concurrency_preserves_explicit_options() {
+ for (configured, expected) in [(None, 7), (Some(3), 3), (Some(0), 0)] {
+ let options: ArrowReaderOptions =
serde_json::from_value(serde_json::json!({
+ "batch_size": 17,
+ "data_file_concurrency_limit": configured,
+ "row_selection_enabled": true,
+ }))
+ .unwrap();
+ let expected = ArrowReaderOptions::builder()
+ .with_batch_size(17)
+ .with_data_file_concurrency_limit(expected)
+ .with_row_selection_enabled(true)
+ .build();
+ assert_eq!(
+ options.with_default_data_file_concurrency_limit(7),
+ expected
+ );
+ }
+ }
+
+ #[test]
+ fn missing_options_use_defaults_and_unknown_options_are_ignored() {
+ let decoded: ArrowReaderOptions =
+ serde_json::from_str(r#"{"future_option":true}"#).unwrap();
+ assert_eq!(decoded, ArrowReaderOptions::default());
+ let decoded: ArrowReaderOptions =
serde_json::from_str(r#"{"batch_size":17}"#).unwrap();
+ assert_eq!(
+ decoded,
+ ArrowReaderOptions::builder().with_batch_size(17).build()
+ );
+ assert_eq!(
+
serde_json::to_value(ArrowReaderOptions::default()).unwrap()["data_file_concurrency_limit"],
+ serde_json::Value::Null
+ );
+ }
- pub(crate) fn preload_column_index(&self) -> bool {
- self.preload_column_index
+ #[test]
+ fn partial_parquet_options_preserve_defaults() {
+ let options: ArrowReaderOptions = serde_json::from_str(
+
r#"{"parquet_options":{"range_fetch_concurrency":2,"future_option":true}}"#,
+ )
+ .unwrap();
+ assert_eq!(
+ options,
+ ArrowReaderOptions::builder()
+ .with_parquet_options(
+ ParquetReadOptions::builder()
+ .with_range_fetch_concurrency(2)
+ .build()
+ )
+ .build()
+ );
}
- pub(crate) fn preload_offset_index(&self) -> bool {
- self.preload_offset_index
+ #[test]
+ fn parquet_override_preserves_other_reader_settings() {
+ let parquet = ParquetReadOptions::builder()
+ .with_metadata_size_hint(4096)
+ .with_range_fetch_concurrency(2)
+ .build();
+ let reader = ArrowReaderBuilder::new(FileIO::new_with_memory(),
test_runtime())
+
.with_options(ArrowReaderOptions::builder().with_batch_size(17).build())
+ .with_parquet_options(parquet.clone())
+ .build();
+ assert_eq!(
+ (reader.batch_size, reader.parquet_read_options),
+ (Some(17), parquet)
+ );
}
- pub(crate) fn preload_page_index(&self) -> bool {
- self.preload_page_index
+ #[test]
+ fn options_restore_local_defaults_or_apply_explicit_settings() {
Review Comment:
Ditto, one test function should have only one test case.
##########
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 {
+ 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, Serialize, Deserialize, TypedBuilder)]
+#[cfg_attr(test, derive(PartialEq))]
+#[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,
- /// Whether to preload the offset index when reading Parquet metadata.
- #[builder(default = true)]
- pub(crate) preload_offset_index: bool,
- /// Whether to preload the page index when reading Parquet metadata.
- #[builder(default = false)]
- pub(crate) preload_page_index: bool,
+ range_fetch_concurrency: usize,
+}
+
+impl Default for ParquetReadOptions {
+ fn default() -> Self {
+ Self::builder().build()
+ }
}
impl ParquetReadOptions {
- pub(crate) fn metadata_size_hint(&self) -> Option<usize> {
+ /// Returns the configured metadata size hint.
+ pub fn metadata_size_hint(&self) -> usize {
self.metadata_size_hint
}
- pub(crate) fn range_coalesce_bytes(&self) -> u64 {
+ /// Returns the byte-gap threshold for coalescing read requests.
+ pub fn range_coalesce_bytes(&self) -> u64 {
self.range_coalesce_bytes
}
- pub(crate) fn range_fetch_concurrency(&self) -> usize {
+ /// Returns the maximum number of concurrent range requests.
+ pub fn range_fetch_concurrency(&self) -> usize {
self.range_fetch_concurrency
}
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use futures::{TryStreamExt, stream};
+
+ use super::*;
+ use crate::arrow::ArrowReaderBuilder;
+ 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();
+ assert_eq!(reader.concurrency_limit_data_files, 1);
+ 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_data_file_concurrency_preserves_explicit_options() {
+ for (configured, expected) in [(None, 7), (Some(3), 3), (Some(0), 0)] {
+ let options: ArrowReaderOptions =
serde_json::from_value(serde_json::json!({
+ "batch_size": 17,
+ "data_file_concurrency_limit": configured,
+ "row_selection_enabled": true,
+ }))
+ .unwrap();
+ let expected = ArrowReaderOptions::builder()
+ .with_batch_size(17)
+ .with_data_file_concurrency_limit(expected)
+ .with_row_selection_enabled(true)
+ .build();
+ assert_eq!(
+ options.with_default_data_file_concurrency_limit(7),
+ expected
+ );
+ }
+ }
+
+ #[test]
+ fn missing_options_use_defaults_and_unknown_options_are_ignored() {
+ let decoded: ArrowReaderOptions =
+ serde_json::from_str(r#"{"future_option":true}"#).unwrap();
+ assert_eq!(decoded, ArrowReaderOptions::default());
+ let decoded: ArrowReaderOptions =
serde_json::from_str(r#"{"batch_size":17}"#).unwrap();
+ assert_eq!(
+ decoded,
+ ArrowReaderOptions::builder().with_batch_size(17).build()
+ );
+ assert_eq!(
+
serde_json::to_value(ArrowReaderOptions::default()).unwrap()["data_file_concurrency_limit"],
+ serde_json::Value::Null
+ );
+ }
- pub(crate) fn preload_column_index(&self) -> bool {
- self.preload_column_index
+ #[test]
+ fn partial_parquet_options_preserve_defaults() {
+ let options: ArrowReaderOptions = serde_json::from_str(
+
r#"{"parquet_options":{"range_fetch_concurrency":2,"future_option":true}}"#,
+ )
+ .unwrap();
+ assert_eq!(
+ options,
+ ArrowReaderOptions::builder()
+ .with_parquet_options(
+ ParquetReadOptions::builder()
+ .with_range_fetch_concurrency(2)
+ .build()
+ )
+ .build()
+ );
}
- pub(crate) fn preload_offset_index(&self) -> bool {
- self.preload_offset_index
+ #[test]
+ fn parquet_override_preserves_other_reader_settings() {
Review Comment:
This is useless, remove it.
##########
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 {
+ 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, Serialize, Deserialize, TypedBuilder)]
+#[cfg_attr(test, derive(PartialEq))]
+#[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,
- /// Whether to preload the offset index when reading Parquet metadata.
- #[builder(default = true)]
- pub(crate) preload_offset_index: bool,
- /// Whether to preload the page index when reading Parquet metadata.
- #[builder(default = false)]
- pub(crate) preload_page_index: bool,
+ range_fetch_concurrency: usize,
+}
+
+impl Default for ParquetReadOptions {
+ fn default() -> Self {
+ Self::builder().build()
+ }
}
impl ParquetReadOptions {
- pub(crate) fn metadata_size_hint(&self) -> Option<usize> {
+ /// Returns the configured metadata size hint.
+ pub fn metadata_size_hint(&self) -> usize {
self.metadata_size_hint
}
- pub(crate) fn range_coalesce_bytes(&self) -> u64 {
+ /// Returns the byte-gap threshold for coalescing read requests.
+ pub fn range_coalesce_bytes(&self) -> u64 {
self.range_coalesce_bytes
}
- pub(crate) fn range_fetch_concurrency(&self) -> usize {
+ /// Returns the maximum number of concurrent range requests.
+ pub fn range_fetch_concurrency(&self) -> usize {
self.range_fetch_concurrency
}
+}
+
+#[cfg(test)]
+mod tests {
+ use std::time::Duration;
+
+ use futures::{TryStreamExt, stream};
+
+ use super::*;
+ use crate::arrow::ArrowReaderBuilder;
+ 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();
+ assert_eq!(reader.concurrency_limit_data_files, 1);
+ 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_data_file_concurrency_preserves_explicit_options() {
+ for (configured, expected) in [(None, 7), (Some(3), 3), (Some(0), 0)] {
+ let options: ArrowReaderOptions =
serde_json::from_value(serde_json::json!({
+ "batch_size": 17,
+ "data_file_concurrency_limit": configured,
+ "row_selection_enabled": true,
+ }))
+ .unwrap();
+ let expected = ArrowReaderOptions::builder()
+ .with_batch_size(17)
+ .with_data_file_concurrency_limit(expected)
+ .with_row_selection_enabled(true)
+ .build();
+ assert_eq!(
+ options.with_default_data_file_concurrency_limit(7),
+ expected
+ );
+ }
+ }
+
+ #[test]
+ fn missing_options_use_defaults_and_unknown_options_are_ignored() {
Review Comment:
One function should be one test case, don't mix them together.
--
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]
