You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "HippoBaro (via GitHub)" <gi...@apache.org> on 2026/04/13 06:58:09 UTC
[PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
HippoBaro opened a new pull request, #9697:
URL: https://github.com/apache/arrow-rs/pull/9697
# Which issue does this PR close?
- Closes #9695 .
# Rationale for this change
The `PushDecoder` (introduced in #7997, #8080) is designed to decouple IO and CPU. It holds non-contiguous byte ranges, with a `NeedsData`/`push_range` protocol. However, it requires each logical read to be satisfied in full by a single physical buffer: `has_range`, `get_bytes`, and `Read::read` all searched for one buffer that entirely covered the requested range.
This assumption conflates two orthogonal IO strategies:
- Coalescing: the IO layer merges adjacent requested ranges into fewer, larger fetches.
- Prefetching: the IO layer pushes data ahead of what the decoder has requested. This is an inversion of control: the IO layer speculatively fills buffers at offsets not yet requested and for arbitrary buffer sizes.
These two strategies interact poorly with the current release mechanism (`clear_ranges`), which matches buffers by exact range equality:
- Coalescing is both rewarded and punished. It is load bearing because without it, the number of physical buffers scale with ranges requested, and `clear_ranges` performs an O(N×M) scan to remove consumed ranges, producing quadratic overhead on wide schemas. But it is also punished because a coalesced buffer never exactly matches any individual requested range, so `clear_ranges` silently skips it: the buffer leaks in `PushBuffers` until the decoder finishes or the caller manually calls `release_all_ranges` (#9624). This increases peak RSS proportionally to the amount of data coalesced ahead of the current row group.
- Prefetching is structurally impossible: speculatively pushed buffers will straddle future read boundaries, so the decoder cannot consume them, and `clear_ranges` cannot release them.
# What changes are included in this PR?
This commit makes `PushBuffers` boundary-agnostic, completing the prefetching story, and changes the internals to scale with buffer count instead of range count:
- Buffer stitching: `has_range`, `get_bytes`, and `Read::read` resolve logical ranges across multiple contiguous physical buffers via binary search, so the IO layer is free to push arbitrarily-sized parts without knowing future read boundaries. This is a nice improvement, because some IO layer can be made much more efficient when using uniform buffers and vectorized reads.
- Incremental release (`release_through`): replaces `clear_ranges` with a watermark-based release that drops all buffers below a byte offset, trimming straddling buffers via zero-copy `Bytes::slice`. The decoder calls this automatically at row-group boundaries.
# Are these changes tested?
Significant coverage added, all tests passing. Benchmark results (vs baseline in #9696):
```
push_decoder/1buf/1000ranges 321.9 µs (was 323.5 µs, −1%)
push_decoder/1buf/10000ranges 3.26 ms (was 3.25 ms, +0%)
push_decoder/1buf/100000ranges 34.9 ms (was 34.6 ms, +1%)
push_decoder/1buf/500000ranges 192.2 ms (was 185.3 ms, +4%)
push_decoder/Nbuf/1000ranges 363.9 µs (was 437.2 µs, −17%)
push_decoder/Nbuf/10000ranges 3.82 ms (was 10.7 ms, −64%)
push_decoder/Nbuf/100000ranges 42.1 ms (was 711.6 ms, −94%)
```
# Are there any user-facing changes?
`PushBuffers:: clear_all_ranges` marked as deprecated in favor of the newer `PushBuffers::clear_all`.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3083332852
##########
parquet/src/arrow/push_decoder/reader_builder/mod.rs:
##########
@@ -610,6 +617,12 @@ impl RowGroupReaderBuilder {
&mut self.buffers,
)?;
+ // All data for this row group has been extracted into the
+ // InMemoryRowGroup. Release physical buffers up to the end
+ // of this row group so streaming IO can reclaim memory.
+ self.buffers
Review Comment:
Thank to you both, I wasn't aware of this! This certainly changes things. My watermark approach was nice and optimal, but this makes it fundamentally flawed. Oh well, back to the drawing board!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4249093758
I've pushed two follow-up commits that address the feedback.
The watermark is gone, replaced by per-range release backed by a `BTreeMap` (as @alamb suggested). Row group column chunks are released individually when consumed or skipped — no monotonicity assumption, no restriction on file layout or access order. This fixes the reverse-scan regression @nathanb9 found.
On the prefetching side, incoming buffers are now filtered at push time against the column chunk byte ranges of the queued row groups, so data the decoder will never consume is discarded before it enters `PushBuffers`. The IO layer can push anything — coalesced, prefetched, the entire file — and the decoder sheds what it doesn't need, or releases it in row-group increments otherwise, regardless of how they are laid out in the file.
> I was thinking more about this change and I thought maybe we could take a step back and figure out what you are trying to accomplish
Interested to know your thoughts about this last version, but completely open to shift direction if you feels this is not the right way to go about it!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4310847578
@alamb @etseidl @nathanb9 I have a local WIP branch that I think gets us into a better place, but every approach I’ve tried so far adds a fair amount of complexity around deciding when a row group can be safely released.
> I also worry it will be complicated to track the exact ranges needed/not needed, and it adds a new non trivial constraint on the decoder to do range tracking.
I agree we should not try to be more granular than the row-group level right now, but even at that level the logic is still fairly subtle because row groups can span discontiguous byte ranges.
The core difficulty is that deciding whether a row group can be released depends on queued row groups, the remaining selection, the running offset/limit budget, and whether predicates require the decoder to stay conservative. Today that state is split across multiple components, which makes the release policy hard to centralize cleanly.
So, rather than keep layering that complexity directly into this PR, I started a separate refactor in #9804 to simplify that part of the decoder first. Once that work has been reviewed and merged, I plan to come back to this PR on top of it, which should make the buffer-release changes much simpler and easier to reason about.
Let me know if that plan sounds reasonable!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4239432848
I will review it -- I am currently working on reviewing
- https://github.com/apache/arrow-rs/pull/9696
Thanks @HippoBaro
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "etseidl (via GitHub)" <gi...@apache.org>.
etseidl commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3082362125
##########
parquet/src/arrow/push_decoder/reader_builder/mod.rs:
##########
@@ -610,6 +617,12 @@ impl RowGroupReaderBuilder {
&mut self.buffers,
)?;
+ // All data for this row group has been extracted into the
+ // InMemoryRowGroup. Release physical buffers up to the end
+ // of this row group so streaming IO can reclaim memory.
+ self.buffers
Review Comment:
> Also technically there is no reason that row groups have to be written in order (though most writers will do that) -- for example, you could have a file where the bytes for row group 0 are after the bytes for row group 1.
Indeed, coworkers of mine are using this property as a means to do deletions from parquet files. Rewrite a single row group, tack it onto the end of the file, and then modify the footer to point to the new row group and ignore the original.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4244021542
🤖 Arrow criterion benchmark running (GKE) | [trigger](https://github.com/apache/arrow-rs/pull/9697#issuecomment-4244008571)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4244008571-1224-mblm4 6.12.55+ #1 SMP Sun Feb 1 08:59:41 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing boundary_agnostic_pushbuffers (d2ea6c489ee6ed8e5c9c29ade673ba17801c85ac) to 711fac8 (merge-base) [diff](https://github.com/apache/arrow-rs/compare/711fac88104fc27d89faaa221345bc95686cf1e9..d2ea6c489ee6ed8e5c9c29ade673ba17801c85ac)
BENCH_NAME=push_decoder
BENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench push_decoder
BENCH_FILTER=
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4246048406
🤖 Arrow criterion benchmark running (GKE) | [trigger](https://github.com/apache/arrow-rs/pull/9697#issuecomment-4246036406)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB) | `Linux bench-c4246036406-1241-wplv5 6.12.55+ #1 SMP Sun Feb 1 08:59:41 UTC 2026 aarch64 GNU/Linux`
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
Comparing boundary_agnostic_pushbuffers (d2ea6c489ee6ed8e5c9c29ade673ba17801c85ac) to 711fac8 (merge-base) [diff](https://github.com/apache/arrow-rs/compare/711fac88104fc27d89faaa221345bc95686cf1e9..d2ea6c489ee6ed8e5c9c29ade673ba17801c85ac)
BENCH_NAME=arrow_reader
BENCH_COMMAND=cargo bench --features=arrow,async,test_common,experimental,object_store --bench arrow_reader
BENCH_FILTER=
Results will be posted here when complete
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4254648038
@alamb from the above PR you linked to:
> I also think we may need to adjust the Parquet PushDecoder somehow to communicate better its internal boundaries (e.g. so we can easily / efficiently pick off RowGroups without recreating it)
I see what you mean. Exposing internal boundaries to the caller is in tension with the IO decoupling I'm pursuing here (ideally the decoder shouldn't dictate how data is fetched or partitioned.) That said, take a look at the `RetentionSet` introduced in c02b044: if work is split across consumer threads, each could own its own `PushBuffers` with a `RetentionSet` scoped to its row groups. The IO layer pushes the same data to all of them, and each decoder admits only what it needs. No boundary negotiation required.
:2c:
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #9697: feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO
URL: https://github.com/apache/arrow-rs/pull/9697
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4239665193
Rebased onto `main` to resolve conflicts and pick up the new benchmarks. We can now run the fancy bot and check whether it shows the expected speedups.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3090919841
##########
parquet/src/util/push_buffers.rs:
##########
@@ -48,6 +59,13 @@ pub(crate) struct PushBuffers {
ranges: Vec<Range<u64>>,
/// The buffers of data that can be used to decode the Parquet file
buffers: Vec<Bytes>,
+ /// High-water mark set by [`Self::release_through`]. After a release,
+ /// no push, has_range, or read may target offsets below this value.
+ #[cfg(feature = "arrow")]
+ watermark: u64,
+ /// Whether `ranges`/`buffers` are sorted by range start.
+ /// Set to `false` on every `push_range`, restored lazily before reads.
+ sorted: bool,
Review Comment:
Latest version uses a Btree which takes care of the invariant for us, thanks!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3083355657
##########
parquet/src/file/metadata/mod.rs:
##########
@@ -713,6 +713,21 @@ impl RowGroupMetaData {
self.file_offset
}
+ /// Returns the byte offset just past the last column chunk in this row group.
Review Comment:
My reading of [the spec](https://github.com/apache/parquet-format/blob/master/README.md) makes me think the guarantee is at the column chunk level: "_Column chunks are composed of pages written back to back._" But there is no equivalent guarantee for column chunks within a row group. It explicitly says "_There is no physical structure that is guaranteed for a row group._"
Another reason, on top of your previous remark, that puts the final nail in the coffin of the watermark release mechanism.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "adriangbot (via GitHub)" <gi...@apache.org>.
adriangbot commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4244061157
🤖 Arrow criterion benchmark completed (GKE) | [trigger](https://github.com/apache/arrow-rs/pull/9697#issuecomment-4244008571)
**Instance:** `c4a-highmem-16` (12 vCPU / 65 GiB)
<details><summary>CPU Details (lscpu)</summary>
```
Architecture: aarch64
CPU op-mode(s): 64-bit
Byte Order: Little Endian
CPU(s): 16
On-line CPU(s) list: 0-15
Vendor ID: ARM
Model name: Neoverse-V2
Model: 1
Thread(s) per core: 1
Core(s) per cluster: 16
Socket(s): -
Cluster(s): 1
Stepping: r0p1
BogoMIPS: 2000.00
Flags: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma lrcpc dcpop sha3 sm3 sm4 asimddp sha512 sve asimdfhm dit uscat ilrcpc flagm sb paca pacg dcpodp sve2 sveaes svepmull svebitperm svesha3 svesm4 flagm2 frint svei8mm svebf16 i8mm bf16 dgh rng bti
L1d cache: 1 MiB (16 instances)
L1i cache: 1 MiB (16 instances)
L2 cache: 32 MiB (16 instances)
L3 cache: 80 MiB (1 instance)
NUMA node(s): 1
NUMA node0 CPU(s): 0-15
Vulnerability Gather data sampling: Not affected
Vulnerability Indirect target selection: Not affected
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Not affected
Vulnerability Reg file data sampling: Not affected
Vulnerability Retbleed: Not affected
Vulnerability Spec rstack overflow: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl
Vulnerability Spectre v1: Mitigation; __user pointer sanitization
Vulnerability Spectre v2: Mitigation; CSV2, BHB
Vulnerability Srbds: Not affected
Vulnerability Tsa: Not affected
Vulnerability Tsx async abort: Not affected
Vulnerability Vmscape: Not affected
```
</details>
<details><summary>Details</summary>
<p>
```
group boundary_agnostic_pushbuffers main
----- ----------------------------- ----
push_decoder/1buf/100000ranges 1.05 53.0±0.48ms ? ?/sec 1.00 50.7±0.64ms ? ?/sec
push_decoder/1buf/10000ranges 1.02 4.8±0.04ms ? ?/sec 1.00 4.7±0.05ms ? ?/sec
push_decoder/1buf/1000ranges 1.01 476.2±2.55µs ? ?/sec 1.00 469.4±2.66µs ? ?/sec
push_decoder/1buf/500000ranges 1.00 282.2±2.24ms ? ?/sec 1.06 298.0±43.35ms ? ?/sec
push_decoder/Nbuf/100000ranges 1.00 57.5±0.40ms ? ?/sec 14.52 834.3±76.53ms ? ?/sec
push_decoder/Nbuf/10000ranges 1.00 5.5±0.13ms ? ?/sec 2.25 12.4±0.03ms ? ?/sec
push_decoder/Nbuf/1000ranges 1.00 530.3±5.49µs ? ?/sec 1.10 582.9±2.43µs ? ?/sec
```
</p>
</details>
<details><summary>Resource Usage</summary>
**base (merge-base)**
| Metric | Value |
|--------|-------|
| Wall time | 183.4s |
| Peak memory | 5.8 GiB |
| Avg memory | 5.1 GiB |
| CPU user | 179.1s |
| CPU sys | 4.1s |
| Peak spill | 0 B |
**branch**
| Metric | Value |
|--------|-------|
| Wall time | 102.7s |
| Peak memory | 5.7 GiB |
| Avg memory | 4.7 GiB |
| CPU user | 101.4s |
| CPU sys | 1.3s |
| Peak spill | 0 B |
</details>
---
[File an issue](https://github.com/adriangb/datafusion-benchmarking/issues) against this benchmark runner
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4244008571
run benchmark push_decoder
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4246653775
I was thinking more about this change and I thought maybe we could take a step back and figure out what you are trying to accomplish
For example, if your usecase is to clear all IO after reading a row group's data, I wonder if we could you just call `clear_all_buffers` after reading the row group's data? One challenge if this is there is no way (currently) to know when the row group's data is no longer needed externally (as in there is no way to introspect the push decoder about when it is done reading a particular row group)
Recently, I have been working on something downstream in DataFusion where I would like a similar API
- https://github.com/apache/datafusion/issues/20529
One solution could be to make a way split off RemainingRowGroups perhaps into a different decoder 🤔
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4248720054
> So prefetching can't simply be "give a list of row groups of interest ahead of time."
Although come to think of it, we could give the IO layer a hint. Not a list of offsets we need right now (`DataRequest`) but a list of "regions" we "may" need.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4281589075
> This gives us fairly clean decoupling: the IO layer can do whatever it wants, but can't push unsolicited buffers, which seems like a reasonable constraint.
> WDYT @alamb?
I think pushing unsolicated buffers is important to support a prefetched usecase, though it is not clear if your proposal precludes that.
I also worry it will be complicated to track the exact ranges needed / not needed, and it adds a new non trivial constrant on the decoder to do range tracking.
I really liked the thoretical simplicity of your initial watermark and I feel like we should be able to leverage the fact that the biggest unit of buffering is a row group. As soon as the decoder is done with a row group, any data pre-fetched for it can be released.
Maybe we could add some way for the decoder to report at a higher granularity what might still be requested.
For example, maybe we could add an API ilke `PushDecoder::remaining_row_groups` -- that returns row groups that the decoder may still read in the future.
and the I/O subsystem can handle the mapping of whatever I/O prefetching it has done to those row groups, and when a row group is no longer needed it can cancel I/O and/or flush or whatever 🤔 And with the ability to clear the buffer sin the push decoder you can control memory very fine grained
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "nathanb9 (via GitHub)" <gi...@apache.org>.
nathanb9 commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3075611132
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -366,13 +369,31 @@ impl ParquetPushDecoder {
self.state.buffered_bytes()
}
- /// Clear any staged byte ranges currently buffered for future decode work.
+ /// Release all staged byte ranges currently buffered for future decode
+ /// work.
///
- /// This clears byte ranges still owned by the decoder's internal
+ /// This releases byte ranges still owned by the decoder's internal
/// `PushBuffers`. It does not affect any data that has already been handed
/// off to an active [`ParquetRecordBatchReader`].
+ pub fn release_all(&mut self) {
+ self.state.release_all();
+ }
+
+ /// Use [`Self::release_all`] instead.
+ #[deprecated(since = "58.1.0", note = "Use `release_all` instead")]
pub fn clear_all_ranges(&mut self) {
Review Comment:
my first commit here deprecated so soon 😆 😢 lol
##########
parquet/src/file/metadata/push_decoder.rs:
##########
@@ -361,6 +360,7 @@ impl ParquetMetaDataPushDecoder {
/// Try to decode the metadata from the pushed data, returning the
/// decoded metadata or an error if not enough data is available.
pub fn try_decode(&mut self) -> Result<DecodeResult<ParquetMetaData>> {
+ self.buffers.ensure_sorted();
Review Comment:
Don't need this because you later call the push buffers's `get_bytes` which also ensures of the order. But since you've made it short circuit anyway maybe its worth leaving for easier readability
##########
parquet/src/util/push_buffers.rs:
##########
@@ -82,83 +100,166 @@ impl PushBuffers {
file_len,
ranges: Vec::new(),
buffers: Vec::new(),
+ #[cfg(feature = "arrow")]
+ watermark: 0,
+ sorted: true,
}
}
- /// Push all the ranges and buffers
+ /// Restore the sort invariant on `ranges`/`buffers`.
+ ///
+ /// Because IO completions are expected to generally arrive in-order,
+ /// `push_range` appends without sorting. We instead delay sorting until
+ /// conumption to amortize its cost, if necessary.
Review Comment:
```suggestion
/// consumption to amortize its cost, if necessary.
```
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3081487227
##########
parquet/src/util/push_buffers.rs:
##########
@@ -48,6 +59,13 @@ pub(crate) struct PushBuffers {
ranges: Vec<Range<u64>>,
/// The buffers of data that can be used to decode the Parquet file
buffers: Vec<Bytes>,
+ /// High-water mark set by [`Self::release_through`]. After a release,
+ /// no push, has_range, or read may target offsets below this value.
Review Comment:
it is probably good to point outwhat "may not" means (like does the code panic if it is tried?)
##########
parquet/src/file/metadata/mod.rs:
##########
@@ -713,6 +713,21 @@ impl RowGroupMetaData {
self.file_offset
}
+ /// Returns the byte offset just past the last column chunk in this row group.
Review Comment:
I think practically speaking most parquet files will have the column chunks for one row group written contiguously in the file, but I am not sure the spec requires this. I do think it effectively requires all pages for a column to be in a contiguous range
##########
parquet/src/util/push_buffers.rs:
##########
@@ -82,83 +100,166 @@ impl PushBuffers {
file_len,
ranges: Vec::new(),
buffers: Vec::new(),
+ #[cfg(feature = "arrow")]
+ watermark: 0,
+ sorted: true,
}
}
- /// Push all the ranges and buffers
+ /// Restore the sort invariant on `ranges`/`buffers`.
+ ///
+ /// Because IO completions are expected to generally arrive in-order,
+ /// `push_range` appends without sorting. We instead delay sorting until
+ /// conumption to amortize its cost, if necessary.
+ ///
+ /// This method must be called before any read-side operation that relies on
+ /// binary search (`has_range`, `get_bytes`, `release_through`,
+ /// `Read::read`). Callers that hold `&mut PushBuffers` should call this
+ /// once before lending `&PushBuffers` to read-side code.
+ pub fn ensure_sorted(&mut self) {
+ if self.sorted {
+ return;
+ }
+
+ // Insertion sort: zero-allocation and linear on nearly-sorted input
Review Comment:
this is n^2 on reverse sorted input though, right?
##########
parquet/src/util/push_buffers.rs:
##########
@@ -48,6 +59,13 @@ pub(crate) struct PushBuffers {
ranges: Vec<Range<u64>>,
/// The buffers of data that can be used to decode the Parquet file
buffers: Vec<Bytes>,
+ /// High-water mark set by [`Self::release_through`]. After a release,
+ /// no push, has_range, or read may target offsets below this value.
+ #[cfg(feature = "arrow")]
+ watermark: u64,
+ /// Whether `ranges`/`buffers` are sorted by range start.
+ /// Set to `false` on every `push_range`, restored lazily before reads.
+ sorted: bool,
Review Comment:
What if we encoded the sort invariant in the type system rather than relying on the flag to be set correctly? Something like
```rust
enum Buffers {
Sorted {
ranges: Vec<Range<u64>>,
buffers: Vec<Bytes>,
}
UnSorted {
ranges: Vec<Range<u64>>,
buffers: Vec<Bytes>,
}
}
```
Maybe it is overly complicated but it make it much clearer that all paths correctly update the sorting
##########
parquet/src/util/push_buffers.rs:
##########
@@ -48,6 +59,13 @@ pub(crate) struct PushBuffers {
ranges: Vec<Range<u64>>,
Review Comment:
If the goal is to keep a list sorted by start range, did you consider using a `BTreeSet`? You could then define some sort of wrapper over Range/Bytes liie
```rust
struct RangeAndData {
range: Range<u64>,
buffer: Bytes
}
impl PartialOrd for RangeAndData {
// define comparison from start range
}
pub(crate) struct PushBuffers {
...
buffers: BtreeSet<RangeAndData>,
}
```
That would probably simplify the accounting significantly
##########
parquet/src/util/push_buffers.rs:
##########
@@ -82,83 +100,166 @@ impl PushBuffers {
file_len,
ranges: Vec::new(),
buffers: Vec::new(),
+ #[cfg(feature = "arrow")]
+ watermark: 0,
+ sorted: true,
}
}
- /// Push all the ranges and buffers
+ /// Restore the sort invariant on `ranges`/`buffers`.
+ ///
+ /// Because IO completions are expected to generally arrive in-order,
+ /// `push_range` appends without sorting. We instead delay sorting until
+ /// conumption to amortize its cost, if necessary.
+ ///
+ /// This method must be called before any read-side operation that relies on
+ /// binary search (`has_range`, `get_bytes`, `release_through`,
+ /// `Read::read`). Callers that hold `&mut PushBuffers` should call this
+ /// once before lending `&PushBuffers` to read-side code.
+ pub fn ensure_sorted(&mut self) {
Review Comment:
does it need to be pub?
##########
parquet/src/arrow/push_decoder/reader_builder/mod.rs:
##########
@@ -212,9 +212,15 @@ impl RowGroupReaderBuilder {
self.buffers.buffered_bytes()
}
- /// Clear any staged ranges currently buffered for future decode work.
- pub fn clear_all_ranges(&mut self) {
- self.buffers.clear_all_ranges();
+ /// Release all staged ranges currently buffered for future decode work.
Review Comment:
I think the clear_all_ranges API was introduced in https://github.com/apache/arrow-rs/pull/9673 by @nathanb9 (and since we haven't released that yet) this change isn't a breaking API
##########
parquet/src/util/push_buffers.rs:
##########
@@ -26,18 +26,29 @@ use std::ops::Range;
/// This is the in-memory buffer for the ParquetDecoder and ParquetMetadataDecoders
///
/// Features:
-/// 1. Zero copy
-/// 2. non contiguous ranges of bytes
+/// 1. Non-contiguous ranges of bytes
+/// 2. Stitching: reads that span multiple contiguous physical buffers are
+/// resolved transparently. When a single buffer covers the request, the
+/// result is zero-copy ([`Bytes::slice`]). When multiple buffers must be
+/// stitched, the data is copied into a new allocation.
///
-/// # Non Coalescing
+/// # No Coalescing
///
-/// This buffer does not coalesce (merging adjacent ranges of bytes into a
-/// single range). Coalescing at this level would require copying the data but
-/// the caller may already have the needed data in a single buffer which would
-/// require no copying.
+/// This buffer does not coalesce (merging adjacent ranges of bytes into a ingle
+/// range). The IO layer is free to push arbitrarily-sized buffers; they will be
+/// stitched on read if needed. Coalescing is left to the IO layer because it
+/// would require an extra copy here, and because the optimal coalescing
+/// strategy depends on the workload and storage medium (e.g. spinning disk,
+/// NVMe, blob storage,) context that only the IO layer has.
///
-/// Thus, the implementation defers to the caller to coalesce subsequent requests
-/// if desired.
+/// # No Speculative Prefetching
Review Comment:
👍
##########
parquet/src/arrow/push_decoder/reader_builder/mod.rs:
##########
@@ -610,6 +617,12 @@ impl RowGroupReaderBuilder {
&mut self.buffers,
)?;
+ // All data for this row group has been extracted into the
+ // InMemoryRowGroup. Release physical buffers up to the end
+ // of this row group so streaming IO can reclaim memory.
+ self.buffers
Review Comment:
You can configure the parquet reader to read row groups in some arbitrary order [with_row_groups](https://docs.rs/parquet/58.0.0/parquet/arrow/arrow_reader/struct.ArrowReaderBuilder.html#method.with_row_groups)
Also technically there is no reason that row groups have to be written in order (though most writers will do that) -- for example, you could have a file where the bytes for row group 0 are after the bytes for row group 1.
So I think assuming that the reader will never want any bytes prior to the current row group should be reconsidered.
Can we instead perhaps release data for the start/end of the row group? rather than just a one sided range?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "etseidl (via GitHub)" <gi...@apache.org>.
etseidl commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3077030332
##########
parquet/src/util/push_buffers.rs:
##########
@@ -26,18 +26,29 @@ use std::ops::Range;
/// This is the in-memory buffer for the ParquetDecoder and ParquetMetadataDecoders
///
/// Features:
-/// 1. Zero copy
-/// 2. non contiguous ranges of bytes
+/// 1. Non-contiguous ranges of bytes
+/// 2. Stitching: reads that span multiple contiguous physical buffers are
+/// resolved transparently. When a single buffer covers the request, the
+/// result is zero-copy ([`Bytes::slice`]). When multiple buffers must be
+/// stitched, the data is copied into a new allocation.
///
-/// # Non Coalescing
+/// # No Coalescing
///
-/// This buffer does not coalesce (merging adjacent ranges of bytes into a
-/// single range). Coalescing at this level would require copying the data but
-/// the caller may already have the needed data in a single buffer which would
-/// require no copying.
+/// This buffer does not coalesce (merging adjacent ranges of bytes into a ingle
Review Comment:
```suggestion
/// This buffer does not coalesce (merging adjacent ranges of bytes into a single
```
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4234548385
@alamb This PR changes code you previously wrote, and I’d value your take on the direction.
Beyond fixing the quadratic complexity issue, it also pushes `PushBuffers`—pun intended—further along the path you laid out. More broadly, better support for I/O coalescing and speculative prefetching would be very valuable for us, and this feels like a step in that direction.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4239701379
Rebased onto `main` to resolve conflicts and pick up the new benchmarks (#9696). We can now run the fancy bot and check whether it shows the expected speedups.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4302955775
Just posting a quick update. I am still working on this and should be able to share the code by the end of the week. Sorry for the delay!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3083361668
##########
parquet/src/util/push_buffers.rs:
##########
@@ -82,83 +100,166 @@ impl PushBuffers {
file_len,
ranges: Vec::new(),
buffers: Vec::new(),
+ #[cfg(feature = "arrow")]
+ watermark: 0,
+ sorted: true,
}
}
- /// Push all the ranges and buffers
+ /// Restore the sort invariant on `ranges`/`buffers`.
+ ///
+ /// Because IO completions are expected to generally arrive in-order,
+ /// `push_range` appends without sorting. We instead delay sorting until
+ /// conumption to amortize its cost, if necessary.
+ ///
+ /// This method must be called before any read-side operation that relies on
+ /// binary search (`has_range`, `get_bytes`, `release_through`,
+ /// `Read::read`). Callers that hold `&mut PushBuffers` should call this
+ /// once before lending `&PushBuffers` to read-side code.
+ pub fn ensure_sorted(&mut self) {
+ if self.sorted {
+ return;
+ }
+
+ // Insertion sort: zero-allocation and linear on nearly-sorted input
Review Comment:
Yes, this was fine under my mistaken assumption that files were always laid out such that reading them back would be sequential. In that case, most reads would happen at the edge, and inserting buffers here would mostly amount to inserting into an ordered, or nearly ordered, vec.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3080451632
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -366,13 +369,31 @@ impl ParquetPushDecoder {
self.state.buffered_bytes()
}
- /// Clear any staged byte ranges currently buffered for future decode work.
+ /// Release all staged byte ranges currently buffered for future decode
+ /// work.
///
- /// This clears byte ranges still owned by the decoder's internal
+ /// This releases byte ranges still owned by the decoder's internal
/// `PushBuffers`. It does not affect any data that has already been handed
/// off to an active [`ParquetRecordBatchReader`].
+ pub fn release_all(&mut self) {
+ self.state.release_all();
+ }
+
+ /// Use [`Self::release_all`] instead.
+ #[deprecated(since = "58.1.0", note = "Use `release_all` instead")]
pub fn clear_all_ranges(&mut self) {
Review Comment:
Honestly, we can keep the symbol—it’s just a rename. I’m trying not to expose the concept of “ranges” on the release side, since the watermark mechanism supersedes it.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4316891164
> So, rather than keep layering that complexity directly into this PR, I started a separate refactor in https://github.com/apache/arrow-rs/pull/9804 to simplify that part of the decoder first.
Makes sense -- I'll check it out
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4307912706
Thank you @HippoBaro -- I appreciate the effort
Marking as draft as I think this PR is no longer waiting on feedback and I am trying to make it easier to find PRs in need of review. Please mark it as ready for review when it is ready for another look
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "nathanb9 (via GitHub)" <gi...@apache.org>.
nathanb9 commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3075604567
##########
parquet/src/file/metadata/push_decoder.rs:
##########
@@ -361,6 +360,7 @@ impl ParquetMetaDataPushDecoder {
/// Try to decode the metadata from the pushed data, returning the
/// decoded metadata or an error if not enough data is available.
pub fn try_decode(&mut self) -> Result<DecodeResult<ParquetMetaData>> {
+ self.buffers.ensure_sorted();
Review Comment:
Don't need this because you later call the push buffers's `get_bytes` which also ensures of the order. But since you've made it short circuit anyway maybe its worth leaving for easier readability
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on code in PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#discussion_r3083357502
##########
parquet/src/util/push_buffers.rs:
##########
@@ -82,83 +100,166 @@ impl PushBuffers {
file_len,
ranges: Vec::new(),
buffers: Vec::new(),
+ #[cfg(feature = "arrow")]
+ watermark: 0,
+ sorted: true,
}
}
- /// Push all the ranges and buffers
+ /// Restore the sort invariant on `ranges`/`buffers`.
+ ///
+ /// Because IO completions are expected to generally arrive in-order,
+ /// `push_range` appends without sorting. We instead delay sorting until
+ /// conumption to amortize its cost, if necessary.
+ ///
+ /// This method must be called before any read-side operation that relies on
+ /// binary search (`has_range`, `get_bytes`, `release_through`,
+ /// `Read::read`). Callers that hold `&mut PushBuffers` should call this
+ /// once before lending `&PushBuffers` to read-side code.
+ pub fn ensure_sorted(&mut self) {
Review Comment:
No, sorry about that
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4260624825
Actually, I dug around and I think we *DO* have the API that is needed:
https://docs.rs/parquet/58.0.0/parquet/arrow/push_decoder/struct.ParquetPushDecoder.html#method.try_next_reader
So the idea would be instead of calling [`try_decode`](https://docs.rs/parquet/58.0.0/parquet/arrow/push_decoder/struct.ParquetPushDecoder.html#method.try_decode) the caller could call `try_next_reader`
And then once a reader is returned, we could then clear out all the cached data (`clear_all_ranges`) and wait for the next request
What do you think @HippoBaro ?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4785344774
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4260500992
To summarize your use case, @HippoBaro, in my own words:
1. You want to control prefetching / coalescing of IO to optimize for your environment, especially object-store
access.
2. You want to know when previously prefetched data will no longer be needed so its resources can be freed.
3. For very wide schemas, you want buffer management to scale efficiently, not quadratically with the number of
columns.
What seems hard about the current setup is:
1. The push decoder knows which ranges it explicitly requested, so it can reason about those.
2. If an external system pushes data at a different granularity, such as speculative prefetch or coalesced
reads, the decoder does not really know what data may still be needed later versus what can safely be released.
My main concern with this PR is that it adds a specific IO buffer management policy for one usage pattern into `ParquetPushDecoder` . That may work well for your object-store case, but it may not generalize as well to other environments or IO strategies (I am thinking io_uring for example)
Thus my preference would be to keep buffer management policy above `ParquetPushDecoder`, and make sure arrow-rs exposes the primitives needed to support it.
For example:
```rust
let decoder = make_decoder();
// Plan and prefetch what is likely needed for row group 1
let row_group1_bytes = /* calculate likely bytes needed for row group 1 */;
let prefetched_data = fancy_prefetcher.get(row_group1_bytes);
decoder.push_data(prefetched_data);
decoder.decode(); // ideally does not need more data until row group 2
// Once row group 1 is fully consumed, release any staged buffered data
decoder.release_all();
// Move on to row group 2
let row_group2_bytes = /* calculate likely bytes needed for row group 2 */;
let prefetched_data = fancy_prefetcher.get(row_group2_bytes);
decoder.push_data(prefetched_data);
decoder.decode(); // ideally does not need more data until row group 3
```
From that perspective, I think we are close to having the right pieces already. What still seems missing is:
1. A reliable way to know when the push decoder has consumed everything it will need from previously pushed data.
2. (Maybe) an easier API to calculate likely byte ranges for a given set of row groups / columns. I think it can be derived from the metadata APIs.
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4260635590
We may need to add some API to the PushDecoder that says not to try and release used ranges though 🤔
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4266078901
> My main concern with this PR is that it adds a specific IO buffer management policy for one usage pattern into ParquetPushDecoder
Yeah, I agree with that. `RetentionSet` feels hacky and breaks with dynamic assignments.
The current code lets the IO layer push more data than was requested and then filters out what is not useful, but as you pointed out, that creates a lot of issues, especially once there are multiple destinations, such as with dynamic work-stealing row group consumption.
So instead, let’s assume the IO layer is responsible for owning prefetched data until it is actually requested. That does not solve the release problem so much as move it around.
> A reliable way to know when the push decoder has consumed everything it will need from previously pushed data.
I think that is partially solved if we use `try_next_reader`. Once you are done with the reader, you drop it, and the buffers go away, as you suggested. I like that. It does not help with speculatively fetched data still sitting in the IO layer, though. For that, we will still need some mechanism to tell the IO layer that certain ranges will never be consumed.
Maybe we could add a new `DecodeResult::SkipData(Vec<Range<u64>>)` that tells the caller: “I will never `NeedsData` this range, so you may free any buffered data for it and cancel associated in-flight IO.” We would return that when the decoder decides to skip a row group, for example.
If I put it all together, we get something like this:
```rust
let decoder = make_decoder();
// retrieve all row groups of interest with metadata
let ranges = predicted_row_group(&metadata);
io_layer.hint(ranges); // maybe start prefetching, maybe not
for rg_idx in row_groups {
match decoder.try_next_reader()? {
DecodeResult::Data(reader) => {
// All pushed buffers go away when the reader is dropped
consume(reader);
},
DecodeResult::NeedsData(ranges) => decoder.push_ranges(ranges, io_layer.fetch(&ranges))?,
DecodeResult::SkipData(ranges) => io_layer.release(ranges),
DecodeResult::Finished => break,
}
}
```
This gives us fairly clean decoupling: the IO layer can do whatever it wants, but **can't push unsolicited buffers**, which seems like a reasonable constraint.
WDYT @alamb?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4244963210
@nathanb9 Thank you for the feedback; I'll see what I can do there
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "nathanb9 (via GitHub)" <gi...@apache.org>.
nathanb9 commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4243147914
Nice, @HippoBaro. One finding: this change introduces a regression in reverse row-group scans, which surfaced to me in a Datafusion test:
https://github.com/apache/datafusion/blob/9dab336ee08eb778e23cd04acdd05f93fd0196f4/datafusion/datasource-parquet/src/opener.rs#L2471-L2477
```rust
// Test reverse scan
let opener = make_opener(true);
let stream = opener.open(file.clone()).unwrap().await.unwrap();
let reverse_values = collect_int32_values(stream).await;
// The forward scan should return data in the order written
assert_eq!(forward_values, vec![1, 2, 3, 4, 5, 6, 7, 8, 9]);
// With reverse scan, row groups are reversed, so we expect:
// Row group 3 (7,8,9), then row group 2 (4,5,6), then row group 1 (1,2,3)
assert_eq!(reverse_values, vec![7, 8, 9, 4, 5, 6, 1, 2, 3]);
```
result:`panic...buffers.rs:193:9:
has_range(57..110) below watermark 163`
Basically, the incremental release watermark invariant conflicts with reverse row-group traversal. From what I understand, incremental release decodes a row group, marks its ending byte offset as the watermark, and then expects subsequent ranges to stay at or above that offset. So not able to do reverse traversal and safely release in this way.
could potentially support a symmetric reverse version of the same monotonic watermark tracking you've introduced here?
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4246019988
> push_decoder/Nbuf/100000ranges 1.00 57.5±0.40ms ? ?/sec 14.52 834.3±76.53ms ? ?/sec
That is certainly nice looking
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4246036406
run benchmark arrow_reader
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat(parquet): make `PushBuffers` boundary-agnostic for prefetch IO [arrow-rs]
Posted by "HippoBaro (via GitHub)" <gi...@apache.org>.
HippoBaro commented on PR #9697:
URL: https://github.com/apache/arrow-rs/pull/9697#issuecomment-4248638163
@alamb @nathanb9 Thank you both for your feedback — it's been very helpful.
The watermark mechanism is fundamentally flawed. I worked under the assumption that all Parquet files are laid out like mine are, but the spec makes no such guarantee: "There is no physical structure that is guaranteed for a row group." Column chunks within a row group need not be contiguous or ordered, so a monotonic watermark won't be correct in general. @nathanb9's reverse-scan regression is one concrete consequence, but it's not the only one. The deletion trick is actually quite neat.
> I was thinking more about this change and I thought maybe we could take a step back and figure out what you are trying to accomplish
Fair point. I should have been clearer in the original description. Here are the high-level goals driving this work:
* **Linear scaling with column count.** My schemas are very high cardinality. Loads of columns. `PushBuffers` must scale linearly (ideally sub-linearly) with the number of columns. The current implementation scales with the number of requested ranges from the reader, which can reach millions on adversarial inputs.
* **Total decoupling from the IO layer.** The decoder should own all logic related to *what* to read; the IO layer should be exclusively concerned with *how* best to exploit the characteristics of the underlying storage medium along whatever axis the user cares about (throughput, latency, cost). The current design leaks information across that boundary: buffers have to align with logical range requests or they leak, and can't straddle request boundaries.
* **Coalescing should not be load bearing.** `PushBuffers` should scale sub-linearly to the number of buffers it's given. Coalescing should be decided based on the characteristics of storage, not because the code can't deal with too many small buffers.
* **First-class support for prefetching.** My data lives in blob storage. The economics of blob storage incentivize pulling as much as possible: you pay the toll per GET request. If we need two row groups separated by 100 MiB of pruned row groups, it is most likely cheapest to stream all of it and discard than to seek. There are also cases where the reader doesn't know which row groups to pull next, because that depends on the content of the current row group — i.e., there are data dependencies. So prefetching can't simply be "give a list of row groups of interest ahead of time."
* **No leaks.** Whatever the IO layer does (coalescing, prefetch, etc.), regardless of access pattern or pruning ratio, we should never leak buffers. This doesn't mean every byte must be freed at the earliest opportunity, but at row-group granularity, everything should be released when we move on to the next. A single decoder should be able to stream-decode TBs worth of data, and the user should have RSS guarantees derived from the max row-group size of the data they are dealing with (and modulo how aggressive prefetching is allowed to be, which they can easily control).
> if your usecase is to clear all IO after reading a row group's data, I wonder if we could you just call clear_all_buffers after reading the row group's data?
The problem is that by the time I finish decoding a row group, I only know I'm done with *that* row group. The IO layer may have already prefetched data I'm about to need. Calling `release_all` would throw away that prefetched work.
The watermark was a nice idea for this: if we know that after reading byte offset N we'll never read anything below N, we can prefetch freely above N and safely release everything below. But since we can't guarantee monotonic access, that doesn't hold.
I'll push a revised version that addresses all the feedback shortly!
--
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: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org