laskoviymishka commented on code in PR #3514:
URL: https://github.com/apache/iceberg-rust/pull/3514#discussion_r4398430375
##########
crates/iceberg/src/arrow/reader/mod.rs:
##########
@@ -133,33 +127,53 @@ impl ArrowReaderBuilder {
///
/// 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.options
+ .parquet_options_mut()
+ .set_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;
+ self.options
+ .parquet_options_mut()
+ .set_range_fetch_concurrency(range_fetch_concurrency);
+ self
+ }
+
+ /// 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
+ }
+
+ /// 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());
Review Comment:
How would you prefer to handle `data_file_concurrency_limit: Some(0)`? As I
read [futures-util
0.3.32](https://github.com/rust-lang/futures-rs/blob/d9bba94c239daa1175a5bb2958f37a5c72db3f6a/futures-util/src/stream/try_stream/try_buffer_unordered.rs#L49-L73),
`try_buffer_unordered(0)` never polls the input and returns `Pending` with
nothing to wake it, so the read hangs. `main` has the same hang through the
`usize` setters, but now a deserialized payload can reach it too. A `.max(1)`
here, an error, or a `NonZeroUsize` field would all work, and a follow-up issue
is fine.
##########
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.
Review Comment:
Could the row-group and row-selection rationale from the removed scan
setters
([`main`](https://github.com/apache/iceberg-rust/blob/ca21f1587c6caaeb88dd3a9e1cc245747e1aba08/crates/iceberg/src/scan/mod.rs#L278-L308))
move somewhere rustdoc renders, like the public getters or the struct doc?
These field docs sit on private fields and the typed builder is
`#[doc(hidden)]`, so nothing rendered says why row selection is off by default
("parsing the page index outweighs any gains"). The bloom-filter note did
survive on `ArrowReaderBuilder::with_bloom_filter_enabled`.
##########
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:
What is each of these needed for? None came up in the threads: public `Copy`
here (removing it later breaks callers,
[public-api.txt](https://github.com/apache/iceberg-rust/blob/76c26a97e379899211794bc86694e42c8356a7f4/crates/iceberg/public-api.txt#L120)),
`PartialEq` on both types (only asserts use it), the `*Builder` re-exports in
[reader/mod.rs](https://github.com/apache/iceberg-rust/blob/76c26a97e379899211794bc86694e42c8356a7f4/crates/iceberg/src/arrow/reader/mod.rs#L46-L48),
and two `Option`-setter styles (`with_batch_size(Some(17))` vs
`with_metadata_size_hint(4096)`). If @blackmwk is fine with the shape, I am too.
##########
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.
Review Comment:
Could the rustdoc say this is a serialized format and what we promise about
it? As far as I can see, the only serde users in the repo are tests
([options.rs](https://github.com/apache/iceberg-rust/blob/76c26a97e379899211794bc86694e42c8356a7f4/crates/iceberg/src/arrow/reader/options.rs#L216-L250),
[scan/mod.rs](https://github.com/apache/iceberg-rust/blob/76c26a97e379899211794bc86694e42c8356a7f4/crates/iceberg/src/scan/mod.rs#L1963-L1973)).
So this doc is the only place downstream learns the field names are a wire
contract, and why the type exists.
```suggestion
/// Configures Arrow reads independently of file planning and local I/O
resources.
///
/// The type is serializable so callers can keep it next to planned
`FileScanTask`s,
/// drop the scan, and build an `ArrowReader` later. Field names are part of
the
/// serialized format: missing fields deserialize to their defaults and
unknown
/// fields are ignored.
```
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -2016,17 +1925,76 @@ mod tests {
}
#[tokio::test]
- async fn test_open_parquet_no_deletions_by_separate_reader() {
+ async fn reader_options_preserve_reads_after_dropping_scan() {
let mut fixture = TableTestFixture::new();
- fixture.setup_manifest_files().await;
-
- // Create table scan for current snapshot and plan files
- let table_scan = fixture
+ fixture.setup_multi_row_group_manifest(&[150, 299]).await;
+ let scan = fixture
.table
.scan()
- .with_row_selection_enabled(true)
+ .select(["y"])
+ .with_filter(Reference::new("y").greater_than(Datum::long(1100)))
.build()
.unwrap();
+ let options = ArrowReaderOptions::builder()
+ .with_batch_size(Some(17))
+ .with_data_file_concurrency_limit(Some(1))
+ .with_parquet_options(
+ ParquetReadOptions::builder()
+ .with_metadata_size_hint(4096)
+ .with_range_fetch_concurrency(2)
+ .build(),
+ )
+ .with_row_selection_enabled(true)
+ .build();
+ let expected: Vec<_> = scan
+ .to_arrow(options.clone())
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let tasks: Vec<_> = scan
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let serialized = serde_json::to_value(&options).unwrap();
+ assert_eq!(
+ serialized,
+ serde_json::json!({
+ "batch_size": 17, "data_file_concurrency_limit": 1,
+ "row_group_filtering_enabled": true, "row_selection_enabled":
true,
+ "bloom_filter_enabled": false,
+ "parquet_options": { "metadata_size_hint": 4096,
"range_coalesce_bytes": 1048576, "range_fetch_concurrency": 2 },
+ })
+ );
+ let options: ArrowReaderOptions =
serde_json::from_value(serialized).unwrap();
Review Comment:
Could we assert the round trip here? The flags don't show up in the read
output, so this test would still pass if deserialization dropped
`row_selection_enabled` or `bloom_filter_enabled`.
```suggestion
let deserialized: ArrowReaderOptions =
serde_json::from_value(serialized).unwrap();
assert_eq!(deserialized, options);
let options = deserialized;
```
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -259,13 +242,6 @@ impl<'a> TableScanBuilder<'a> {
pub fn with_concurrency_limit(mut self, limit: usize) -> Self {
Review Comment:
Could `to_arrow` keep the old contract by falling back to the scan limit
when `data_file_concurrency_limit` is unset? On `main` this method also bounded
data files
([L262](https://github.com/apache/iceberg-rust/blob/ca21f1587c6caaeb88dd3a9e1cc245747e1aba08/crates/iceberg/src/scan/mod.rs#L257-L264)),
so `.with_concurrency_limit(1)` plus `to_arrow(Default::default())` goes from
one in-flight data file to `available_parallelism()`. As I read
[`build()`](https://github.com/apache/iceberg-rust/blob/76c26a97e379899211794bc86694e42c8356a7f4/crates/iceberg/src/arrow/reader/mod.rs#L159-L172),
that value also sizes the delete-file loader.
If we keep the split, could you explain here why no existing caller relies
on that bound?
--
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]
