Repository navigation
fix(arrow): decode delete files on the CPU runtime - #3587
alexanderbianchi wants to merge 1 commit into
Conversation
| let (parquet_file_reader, arrow_metadata) = ArrowReader::open_parquet_file( | ||
| data_file_path, | ||
| &self.file_io, | ||
| &self.runtime, |
There was a problem hiding this comment.
heads up, #3433 (mine) adds a metadata() call in parquet_to_batch_stream when the delete file size is unknown. with this change that call would run on the cpu runtime. whichever of the two lands second should route it through io too, I can do it on my side if this goes first.
| let parquet_file = file_io.new_input(data_file_path)?; | ||
| let counting_reader = | ||
| CountingFileRead::new(parquet_file.reader().await?, Arc::clone(bytes_read)); | ||
| let reader = RuntimeFileRead::open(file_io, data_file_path, runtime.io()).await?; |
There was a problem hiding this comment.
since open_parquet_file is shared, data file reads also go through RuntimeFileRead with this PR, not only after #3588. page reads made while the scan stream is polled already get dispatched to io here. fine by me, maybe worth a line in the description.
| return Self::load_deletion_vector(task, basic_delete_file_loader).await; | ||
| let task = task.clone(); | ||
| return io | ||
| .spawn(async move { |
There was a problem hiding this comment.
delete_loading_requires_cpu_runtime uses setup(), which only has parquet delete files, so nothing checks that this DV read actually runs on io. a split-runtime test with a puffin DV and a thread name check like the one in runtime_file_read.rs would cover it.
Which issue does this PR close?
Related to #3276 and #3277. No issue is closed by this draft.
What changes are included in this PR?
Delete-file loading currently consumes Parquet streams on the I/O runtime, so footer parsing, decompression, page decoding, and delete-filter construction compete with storage requests. Run the caching delete loader on the configured CPU runtime instead.
Add a private runtime-dispatched file reader that opens storage readers and executes byte-range reads on the I/O runtime. Awaiting these reads resumes Parquet processing on its caller's runtime. Dropping a read cancels its spawned storage request. The shared Parquet opening path uses this adapter; data-file decoding placement is addressed in dependent draft #3588.
Puffin deletion-vector blob loading stays on I/O workers; vector deserialization and validation run in the CPU parse phase. Existing delete caching and concurrency limits are preserved.
This adds task dispatch per storage read. No throughput benchmark has been run.
Are these changes tested?
cargo test -p iceberg --lib arrow::: 285 passed.cargo clippy -p iceberg --lib --tests -- -D warnings: passed.cargo fmt --all -- --check: passed.AI Disclosure
Implementation, tests, and this description were prepared with OpenAI Codex assistance.