comphead commented on code in PR #3514:
URL: https://github.com/apache/iceberg-rust/pull/3514#discussion_r4353216260
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -602,23 +602,51 @@ impl TableScan {
/// Returns an [`ArrowRecordBatchStream`].
pub async fn to_arrow(&self) -> Result<ArrowRecordBatchStream> {
- let mut arrow_reader_builder =
- ArrowReaderBuilder::new(self.file_io.clone(), self.runtime.clone())
-
.with_data_file_concurrency_limit(self.concurrency_limit_data_files)
-
.with_row_group_filtering_enabled(self.row_group_filtering_enabled)
- .with_row_selection_enabled(self.row_selection_enabled)
- .with_bloom_filter_enabled(self.bloom_filter_enabled);
-
- if let Some(batch_size) = self.batch_size {
- arrow_reader_builder =
arrow_reader_builder.with_batch_size(batch_size);
- }
-
- arrow_reader_builder
+ self.reader_options()
+ .apply_to(ArrowReaderBuilder::new(
+ self.file_io.clone(),
+ self.runtime.clone(),
+ ))
.build()
.read(self.plan_files().await?)
.map(|result| result.stream())
}
+ /// Captures reader settings independently of this scan's planning state.
+ ///
+ /// Projection, predicates, and delete references are carried by the
planned
+ /// file tasks; manifest-planning concurrency is not a reader setting.
+ ///
+ /// ```no_run
+ /// # use futures::{TryStreamExt, stream};
+ /// # use iceberg::{Result, Runtime, arrow::ArrowReaderBuilder,
io::FileIO, table::Table};
+ /// # async fn read(table: &Table, local_io: FileIO, local_runtime:
Runtime) -> Result<()> {
+ /// let scan = table.scan().with_batch_size(Some(1024)).build()?;
+ /// let options = scan.reader_options();
+ /// let tasks: Vec<_> = scan.plan_files().await?.try_collect().await?;
+ /// drop(scan);
+ ///
+ /// let reader = options
+ /// .apply_to(ArrowReaderBuilder::new(local_io, local_runtime))
+ /// .build();
+ /// let batches: Vec<_> = reader
+ /// .read(Box::pin(stream::iter(tasks.into_iter().map(Ok))))?
+ /// .stream()
+ /// .try_collect()
+ /// .await?;
+ /// # Ok(())
+ /// # }
+ /// ```
+ pub fn reader_options(&self) -> TableScanReaderOptions {
Review Comment:
The same five settings now live in `TableScanBuilder`, `TableScan` (two
struct literals in `build()`), `ArrowReaderBuilder`, and this new type, and
`reader_options()` copies them one by one. If it stays simple, store a
`TableScanReaderOptions` in `TableScanBuilder` and `TableScan` with the
`with_*` setters writing into it. Then `reader_options()` returns
`self.reader_options` and both `TableScan { .. }` literals lose five fields
each. Skip if the setter churn outweighs that.
##########
crates/iceberg/src/arrow/reader/scan_options.rs:
##########
@@ -0,0 +1,116 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use serde::{Deserialize, Serialize};
+
+use super::ArrowReaderBuilder;
+
+/// Reader settings captured from a [`TableScan`](crate::scan::TableScan).
+///
+/// Contains no I/O handles, runtime, or planning state. Apply these options
to a
+/// locally constructed builder, then build one reader and clone it across
+/// partitions to share its delete-file cache.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+pub struct TableScanReaderOptions {
+ pub(crate) batch_size: Option<usize>,
+ pub(crate) data_file_concurrency_limit: usize,
+ pub(crate) row_group_filtering_enabled: bool,
+ pub(crate) row_selection_enabled: bool,
+ pub(crate) bloom_filter_enabled: bool,
+}
+
+impl TableScanReaderOptions {
+ /// The requested record batch size, or the reader default.
+ pub fn batch_size(&self) -> Option<usize> {
Review Comment:
The five getters have no caller in this PR and the description only needs
`apply_to` plus serde. Each public getter is API to keep stable. Consider
dropping them until a consumer needs one, which also shrinks `public-api.txt`.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -2015,6 +2043,64 @@ mod tests {
assert_eq!(int64_arr.value(0), 1);
}
+ #[tokio::test]
+ async fn reader_options_preserve_reads_after_dropping_scan() {
+ let mut fixture = TableTestFixture::new();
+ fixture.setup_multi_row_group_manifest(&[150, 299]).await;
+ for enabled in [false, true] {
Review Comment:
The `for enabled in [false, true]` loop does not add coverage for the flags.
Mutation check: deleting the `row_selection_enabled` and `bloom_filter_enabled`
assignments in `apply_to` fails only
`applies_scan_settings_and_preserves_local_io_tuning`. This test still passes
because the result equals `to_arrow()` either way. Consider a single iteration
and keep what only this test covers: the JSON shape, drop-scan, fresh local
I/O, and the batch size bound.
##########
crates/iceberg/src/arrow/reader/scan_options.rs:
##########
@@ -0,0 +1,116 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use serde::{Deserialize, Serialize};
+
+use super::ArrowReaderBuilder;
+
+/// Reader settings captured from a [`TableScan`](crate::scan::TableScan).
+///
+/// Contains no I/O handles, runtime, or planning state. Apply these options
to a
+/// locally constructed builder, then build one reader and clone it across
+/// partitions to share its delete-file cache.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
Review Comment:
On the serialized-option evolution you asked about: every field is required
on deserialize and the struct is `Copy + Eq`. A payload written by an older
planner will fail with `missing field` on a newer worker once a sixth setting
is added. A non-`Copy` field would also break the public API.
Consider `#[serde(default)]` (container level, with a `Default` that matches
the `ArrowReaderBuilder` defaults) so new fields can be added later. Unknown
fields from a newer planner are already ignored by serde.
Question (unverified): `data_file_concurrency_limit` defaults to
`available_parallelism()` of the planning host and is now serialized. A
coordinator with 64 cores would pass 64 to a 4-core worker unless the caller
overrides it. For comparison, Java's `SparkInputPartition` ships
`caseSensitive`, the expected schema, and `cacheDeleteFilesOnExecutors`, but no
batch size or concurrency. Should this field be `Option` so an unset value
keeps the local default, like `batch_size`?
--
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]
