blackmwk commented on code in PR #3514:
URL: https://github.com/apache/iceberg-rust/pull/3514#discussion_r4399007019
##########
crates/iceberg/src/arrow/reader/options.rs:
##########
@@ -15,70 +15,324 @@
// 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.
+///
+/// ```no_run
+/// # use iceberg::{Result, arrow::{ArrowReaderOptions, ParquetReadOptions},
table::Table};
+/// # async fn read(table: &Table) -> Result<()> {
+/// let options = ArrowReaderOptions::builder()
+/// .with_batch_size(Some(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.
+ batch_size: Option<usize>,
+ /// Maximum in-flight data files, or execution-host parallelism when unset.
+ data_file_concurrency_limit: Option<usize>,
Review Comment:
nit: Add a default =None and strip options attributes make api a little
easier to read.
##########
crates/iceberg/src/arrow/reader/options.rs:
##########
@@ -15,70 +15,324 @@
// 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.
+///
+/// ```no_run
+/// # use iceberg::{Result, arrow::{ArrowReaderOptions, ParquetReadOptions},
table::Table};
+/// # async fn read(table: &Table) -> Result<()> {
+/// let options = ArrowReaderOptions::builder()
+/// .with_batch_size(Some(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.
+ batch_size: Option<usize>,
+ /// Maximum in-flight data files, or execution-host parallelism when unset.
+ 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 {
+ /// Returns the configured batch size.
+ pub fn batch_size(&self) -> Option<usize> {
+ self.batch_size
+ }
+
+ pub(super) fn parquet_options_mut(&mut self) -> &mut ParquetReadOptions {
+ &mut self.parquet_options
+ }
+
+ pub(super) fn set_batch_size(&mut self, value: Option<usize>) {
+ self.batch_size = value;
+ }
+ /// Returns the configured data file concurrency limit.
+ pub fn data_file_concurrency_limit(&self) -> Option<usize> {
+ self.data_file_concurrency_limit
+ }
+
+ pub(super) fn set_data_file_concurrency_limit(&mut self, value:
Option<usize>) {
+ self.data_file_concurrency_limit = value;
+ }
+ /// Returns whether row-group statistics filtering is enabled.
+ pub fn row_group_filtering_enabled(&self) -> bool {
+ self.row_group_filtering_enabled
+ }
+
+ pub(super) fn set_row_group_filtering_enabled(&mut self, value: bool) {
+ self.row_group_filtering_enabled = value;
+ }
+ /// Returns whether page-index row selection is enabled.
+ pub fn row_selection_enabled(&self) -> bool {
+ self.row_selection_enabled
+ }
+
+ pub(super) fn set_row_selection_enabled(&mut self, value: bool) {
+ self.row_selection_enabled = value;
+ }
+ /// Returns whether bloom-filter pruning is enabled.
+ pub fn bloom_filter_enabled(&self) -> bool {
+ self.bloom_filter_enabled
+ }
+
+ pub(super) fn set_bloom_filter_enabled(&mut self, value: bool) {
+ self.bloom_filter_enabled = value;
+ }
+ /// 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, Copy, 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 = Some(DEFAULT_METADATA_SIZE_HINT),
setter(strip_option))]
+ metadata_size_hint: Option<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,
+ range_fetch_concurrency: usize,
/// Whether to preload the column index when reading Parquet metadata.
- #[builder(default = true)]
- pub(crate) preload_column_index: bool,
+ #[builder(default = true, setter(skip))]
Review Comment:
Why we need to skip serde and setter for following fields?
##########
crates/iceberg/src/arrow/reader/options.rs:
##########
@@ -15,70 +15,324 @@
// 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.
+///
+/// ```no_run
+/// # use iceberg::{Result, arrow::{ArrowReaderOptions, ParquetReadOptions},
table::Table};
+/// # async fn read(table: &Table) -> Result<()> {
+/// let options = ArrowReaderOptions::builder()
+/// .with_batch_size(Some(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.
+ batch_size: Option<usize>,
+ /// Maximum in-flight data files, or execution-host parallelism when unset.
+ 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 {
+ /// Returns the configured batch size.
+ pub fn batch_size(&self) -> Option<usize> {
+ self.batch_size
+ }
+
+ pub(super) fn parquet_options_mut(&mut self) -> &mut ParquetReadOptions {
+ &mut self.parquet_options
+ }
+
+ pub(super) fn set_batch_size(&mut self, value: Option<usize>) {
+ self.batch_size = value;
+ }
+ /// Returns the configured data file concurrency limit.
+ pub fn data_file_concurrency_limit(&self) -> Option<usize> {
+ self.data_file_concurrency_limit
+ }
+
+ pub(super) fn set_data_file_concurrency_limit(&mut self, value:
Option<usize>) {
+ self.data_file_concurrency_limit = value;
+ }
+ /// Returns whether row-group statistics filtering is enabled.
+ pub fn row_group_filtering_enabled(&self) -> bool {
+ self.row_group_filtering_enabled
+ }
+
+ pub(super) fn set_row_group_filtering_enabled(&mut self, value: bool) {
+ self.row_group_filtering_enabled = value;
+ }
+ /// Returns whether page-index row selection is enabled.
+ pub fn row_selection_enabled(&self) -> bool {
+ self.row_selection_enabled
+ }
+
+ pub(super) fn set_row_selection_enabled(&mut self, value: bool) {
+ self.row_selection_enabled = value;
+ }
+ /// Returns whether bloom-filter pruning is enabled.
+ pub fn bloom_filter_enabled(&self) -> bool {
+ self.bloom_filter_enabled
+ }
+
+ pub(super) fn set_bloom_filter_enabled(&mut self, value: bool) {
+ self.bloom_filter_enabled = value;
+ }
+ /// 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, Copy, 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 = Some(DEFAULT_METADATA_SIZE_HINT),
setter(strip_option))]
Review Comment:
This is odd, if it has a default value, then why to make it option? Just
remove the option.
##########
crates/iceberg/src/arrow/reader/options.rs:
##########
@@ -15,70 +15,324 @@
// 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.
+///
+/// ```no_run
+/// # use iceberg::{Result, arrow::{ArrowReaderOptions, ParquetReadOptions},
table::Table};
+/// # async fn read(table: &Table) -> Result<()> {
+/// let options = ArrowReaderOptions::builder()
+/// .with_batch_size(Some(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.
+ batch_size: Option<usize>,
+ /// Maximum in-flight data files, or execution-host parallelism when unset.
+ 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 {
+ /// Returns the configured batch size.
+ pub fn batch_size(&self) -> Option<usize> {
+ self.batch_size
+ }
+
+ pub(super) fn parquet_options_mut(&mut self) -> &mut ParquetReadOptions {
+ &mut self.parquet_options
+ }
+
+ pub(super) fn set_batch_size(&mut self, value: Option<usize>) {
+ self.batch_size = value;
+ }
+ /// Returns the configured data file concurrency limit.
+ pub fn data_file_concurrency_limit(&self) -> Option<usize> {
+ self.data_file_concurrency_limit
+ }
+
+ pub(super) fn set_data_file_concurrency_limit(&mut self, value:
Option<usize>) {
+ self.data_file_concurrency_limit = value;
+ }
+ /// Returns whether row-group statistics filtering is enabled.
+ pub fn row_group_filtering_enabled(&self) -> bool {
+ self.row_group_filtering_enabled
+ }
+
+ pub(super) fn set_row_group_filtering_enabled(&mut self, value: bool) {
+ self.row_group_filtering_enabled = value;
+ }
+ /// Returns whether page-index row selection is enabled.
+ pub fn row_selection_enabled(&self) -> bool {
+ self.row_selection_enabled
+ }
+
+ pub(super) fn set_row_selection_enabled(&mut self, value: bool) {
+ self.row_selection_enabled = value;
+ }
+ /// Returns whether bloom-filter pruning is enabled.
+ pub fn bloom_filter_enabled(&self) -> bool {
+ self.bloom_filter_enabled
+ }
+
+ pub(super) fn set_bloom_filter_enabled(&mut self, value: bool) {
+ self.bloom_filter_enabled = value;
+ }
+ /// 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, Copy, Debug, PartialEq, Serialize, Deserialize, TypedBuilder)]
Review Comment:
+1, I think we should unify the setter style for options, also I'm leaning
to the strip option.
##########
crates/iceberg/src/arrow/reader/mod.rs:
##########
@@ -52,55 +54,45 @@ 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 {
Review Comment:
I was planning to mark these apis as deprecated, but this forces use to keep
the modifier of the options struct, which I don't want to keep. How about we
remove these functions directly? cc @laskoviymishka @alexanderbianchi WDYT?
--
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]
