mbutrovich opened a new pull request, #2349: URL: https://github.com/apache/iceberg-rust/pull/2349
## Which issue does this PR close? - Closes #. ## What changes are included in this PR? Add per-scan I/O metrics to `ArrowReader` so callers can observe bytes read from storage during a scan. **Motivation:** Downstream engines need per-scan byte counts to report I/O metrics in their UIs. For example, DataFusion Comet uses this to populate `bytes_scanned` on its Iceberg scan operator, which flows through to Spark UI's Executors page "Input" column via `TaskMetrics.inputMetrics.setBytesRead()`. This must be per-scan, not global storage-level metrics — concurrent scans against the same `FileIO` need independent counters. The approach matches DataFusion's pattern of wrapping `AsyncFileReader` with a counting layer, and is storage-backend agnostic (works with any `FileRead` implementation without changes to opendal, object_store, etc.). **New file: `crates/iceberg/src/arrow/scan_metrics.rs`** - `CountingFileRead`: wraps `FileRead` and increments a shared `AtomicU64` on each `read()` call. One `fetch_add(Relaxed)` per I/O request, no `Arc` clone per read. - `ScanMetrics`: public handle exposing `bytes_read()`. Designed as the extension point for future scan metrics (row groups pruned, deletes applied, etc.) without changing the return type again. **New method: `ArrowReader::read_with_metrics()`** - Returns `Result<(ArrowRecordBatchStream, ScanMetrics)>`. - Existing `read()` is unchanged — it delegates to `read_with_metrics()` and discards the metrics. No breaking change. - The counter is created fresh per call, not stored on `ArrowReader`, so cloned readers get independent metrics. **`open_parquet_file()` takes an optional counter** - `CountingFileRead` wraps before metadata loading so footer/index reads are counted. - Delete file loader passes `None` — delete file I/O is not counted, consistent with Iceberg Java's separate `totalDeleteFileSizeInBytes` tracking. **`FileScanTaskReader` struct (refactor)** - Extracted the 8 parameters of `process_file_scan_task` into a `Clone` struct with a `process(self, task)` method. Resolves a `clippy::too_many_arguments` violation that would recur with each new feature. Same per-task clone cost (actually one fewer `FileIO` clone since `self.file_io` is moved instead of pre-cloned). **Re-exports** - `ScanMetrics` re-exported from `iceberg::arrow` and `iceberg::scan`. ## Are these changes tested? `test_scan_metrics_bytes_read` in `reader.rs`: writes a Parquet file, reads it through `ArrowReader::read_with_metrics()`, asserts `bytes_read() == 0` before stream consumption and `bytes_read() > 0` after. All 40 existing reader tests and 34 scan tests pass unchanged. -- 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]
