blackmwk commented on code in PR #3514:
URL: https://github.com/apache/iceberg-rust/pull/3514#discussion_r4478314477
##########
crates/iceberg/src/arrow/reader/options.rs:
##########
@@ -15,70 +15,283 @@
// 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_data_file_concurrency_limit_if_unset(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 with_parquet_options(mut self, value: ParquetReadOptions) ->
Self {
+ self.parquet_options = value;
+ self
+ }
+}
+
/// 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,
- /// 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,
+ #[builder(default = true, setter(skip))]
+ preload_column_index: bool,
+ #[builder(default = true, setter(skip))]
+ preload_offset_index: bool,
+ // Execution derives page-index preloading per file from predicates and
deletes.
+ #[builder(default = false, setter(skip))]
+ preload_page_index: bool,
+}
+
+impl Default for ParquetReadOptions {
+ fn default() -> Self {
+ Self::builder().build()
+ }
}
impl ParquetReadOptions {
- pub(crate) fn metadata_size_hint(&self) -> Option<usize> {
+ pub(super) fn with_preload_page_index(mut self, enabled: bool) -> Self {
+ self.preload_page_index = enabled;
+ self
+ }
+
+ pub(super) fn preload_column_index(&self) -> bool {
+ self.preload_column_index
+ }
+
+ pub(super) fn preload_offset_index(&self) -> bool {
+ self.preload_offset_index
+ }
+
+ pub(super) fn preload_page_index(&self) -> bool {
Review Comment:
Since we are now make `ParquetReadOptions` pub, we can also make them pub.
--
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]
