This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new ce110bba02 refactor(parquet): do not store empty pushed ranges in
PushBuffers (#11569)
ce110bba02 is described below
commit ce110bba0277405a7c2a5903468fb342634aefe3
Author: Adrian Garcia Badaracco <[email protected]>
AuthorDate: Thu Oct 15 23:07:36 2026 +0000
refactor(parquet): do not store empty pushed ranges in PushBuffers (#11569)
# Which issue does this PR close?
- Follow-up to https://github.com/apache/arrow-rs/pull/11459, as
suggested by @alamb in
https://github.com/apache/arrow-rs/pull/11459#pullrequestreview-5523546294
# Rationale for this change
In https://github.com/apache/arrow-rs/pull/11459, `PushBuffers` keeps an
`empty` counter of the empty ranges that `push_range` accepts. When the
counter is not zero, `release_ranges` scans all buffers to remove them.
This adds state and a slow path for ranges that contain no bytes.
It is not safe to only drop empty pushes. A lookup for an empty range
`n..n` succeeds only if a buffer touches offset `n`. If the decoder
requests such a range, the caller pushes `n..n`. If that push is
dropped, the decoder requests the same range again on each call, and the
decoder never makes progress.
# What changes are included in this PR?
- `has_range`, `get_bytes` and `Read` treat an empty request as always
available. `get_bytes(_, 0)` returns `Bytes::new()`.
- `push_range` does not store an empty (or inverted) range, because it
adds no bytes.
- Remove the `empty` counter, the full-scan path in `release_ranges`,
and the `empty` bookkeeping in `clear_ranges` and `clear_all_ranges`.
`release_ranges` now only looks at the buffers that overlap the released
ranges.
# Are these changes tested?
Yes.
- New unit test `empty_ranges` in `push_buffers.rs`: an empty request is
available with no buffers, an empty push is a no-op, and reads of real
data still work after an empty push.
- New test `needed_ranges_never_contains_an_empty_range` in the push
decoder (`reader_builder/data.rs`): a request with a zero-length range
never lists that range as needed, also after the caller pushes it.
- The fuzz test model now also skips empty pushes and expects empty
requests to be available. The invariant check asserts that no empty
buffer is stored.
# Are there any user-facing changes?
Yes, one small behaviour change. `ParquetPushDecoder` and
`ParquetMetaDataPushDecoder` no longer list zero-length ranges in
`DecodeResult::NeedsData`. Before this change, a zero-length range that
no pushed buffer touched was listed as needed. With this PR, an empty
push is not stored, so such a request would never be satisfied and the
decoder would loop forever. Because an empty range is now always
available, the decoders do not request it, and this risk is removed.
Callers that push empty ranges still get `Ok(())`. There are no public
API changes.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../src/arrow/push_decoder/reader_builder/data.rs | 34 ++++++++
parquet/src/util/push_buffers.rs | 91 ++++++++++++++++------
2 files changed, 100 insertions(+), 25 deletions(-)
diff --git a/parquet/src/arrow/push_decoder/reader_builder/data.rs
b/parquet/src/arrow/push_decoder/reader_builder/data.rs
index ab628bf285..305e3bf8e2 100644
--- a/parquet/src/arrow/push_decoder/reader_builder/data.rs
+++ b/parquet/src/arrow/push_decoder/reader_builder/data.rs
@@ -237,3 +237,37 @@ impl<'a> DataRequestBuilder<'a> {
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ /// A zero-length requested range (for example, a column chunk with no
+ /// bytes) must never be in `NeedsData`. Before, the decoder requested it
+ /// again after each push, because an empty push is not stored, and did
+ /// not make progress.
+ #[test]
+ fn needed_ranges_never_contains_an_empty_range() {
+ let request = DataRequest {
+ column_chunks: vec![None, None],
+ ranges: vec![100..100, 200..210],
+ page_start_offsets: None,
+ };
+ let mut buffers = PushBuffers::new(1000);
+ assert_eq!(request.needed_ranges(&buffers), vec![200..210]);
+
+ // A caller that pushes the empty range anyway does not change this.
+ buffers.push_range(100..100, Bytes::new()).unwrap();
+ assert_eq!(request.needed_ranges(&buffers), vec![200..210]);
+
+ buffers
+ .push_range(200..210, Bytes::from_static(b"0123456789"))
+ .unwrap();
+ assert!(request.needed_ranges(&buffers).is_empty());
+ let chunks = request.get_chunks(&buffers).unwrap();
+ assert_eq!(
+ chunks,
+ vec![Bytes::new(), Bytes::from_static(b"0123456789")]
+ );
+ }
+}
diff --git a/parquet/src/util/push_buffers.rs b/parquet/src/util/push_buffers.rs
index e1fa2bd049..0aade10616 100644
--- a/parquet/src/util/push_buffers.rs
+++ b/parquet/src/util/push_buffers.rs
@@ -79,10 +79,6 @@ pub struct PushBuffers {
/// The sum of the lengths of `ranges`, kept up to date so that
/// [`Self::buffered_bytes`] does not scan all ranges.
buffered_bytes: u64,
- /// The number of empty ranges in `ranges`. [`Self::push_range`] accepts
- /// them, and [`Self::release_ranges`] removes them. The count tells
- /// `release_ranges` when it must look for them in all buffers.
- empty: usize,
}
impl Display for PushBuffers {
@@ -124,7 +120,6 @@ impl PushBuffers {
max_len: 0,
max_len_count: 0,
buffered_bytes: 0,
- empty: 0,
}
}
@@ -154,6 +149,9 @@ impl PushBuffers {
/// Push a new range and its associated buffer
///
+ /// An empty range adds no bytes, so it is not stored. An empty range is
+ /// always available (see `has_range`).
+ ///
/// # Errors
/// Returns an error if the buffer's length does not match the range's
/// length, e.g. when a truncated (short) read is pushed.
@@ -168,6 +166,9 @@ impl PushBuffers {
range.end
));
}
+ if expected == 0 {
+ return Ok(());
+ }
// Insert after all buffers that start at or before `range.start`.
// Thus, ranges pushed in file order go at the end.
let idx = self.ranges.partition_point(|r| r.start <= range.start);
@@ -177,15 +178,16 @@ impl PushBuffers {
Ordering::Less => {}
}
self.buffered_bytes += expected;
- self.empty += usize::from(expected == 0);
self.ranges.insert(idx, range);
self.buffers.insert(idx, buffer);
Ok(())
}
/// Returns true if the Buffers contains data for the given range
+ ///
+ /// An empty range needs no bytes, so it is always available.
pub(crate) fn has_range(&self, range: &Range<u64>) -> bool {
- self.find(range.start, range.end).is_some()
+ range.is_empty() || self.find(range.start, range.end).is_some()
}
/// Returns the index of a buffer that contains all bytes of `start..end`,
@@ -287,7 +289,6 @@ impl PushBuffers {
if remove.next_if_eq(&idx).is_some() {
let len =
self.ranges[idx].end.saturating_sub(self.ranges[idx].start);
self.buffered_bytes -= len;
- self.empty -= usize::from(len == 0);
if len == self.max_len {
self.max_len_count -= 1;
}
@@ -429,15 +430,9 @@ impl PushBuffers {
}
}
}
- if emptied || self.empty > 0 {
- // Remove the released buffers. As before, this also removes empty
- // buffers that were pushed. The released buffers are all in
- // `first..=last`, so look at all buffers only if empty buffers
- // were pushed.
- let span = match self.empty {
- 0 => first..last + 1,
- _ => 0..self.ranges.len(),
- };
+ if emptied {
+ // Remove the released buffers. They are all in `first..=last`.
+ let span = first..last + 1;
let mut kept = span.start;
for idx in span.clone() {
if !self.ranges[idx].is_empty() {
@@ -448,7 +443,6 @@ impl PushBuffers {
}
self.ranges.drain(kept..span.end);
self.buffers.drain(kept..span.end);
- self.empty = 0;
}
// A trim moves the start of a buffer only past released bytes. Each
// other buffer that starts in those bytes is trimmed past them too,
@@ -495,7 +489,6 @@ impl PushBuffers {
self.max_len = 0;
self.max_len_count = 0;
self.buffered_bytes = 0;
- self.empty = 0;
}
/// Panics if `ranges`, `buffers` and `max_len` do not agree, or if the
@@ -524,8 +517,11 @@ impl PushBuffers {
self.max_len_count,
self.ranges.iter().filter(is_longest).count()
);
- let empty = self.ranges.iter().filter(|r| r.is_empty()).count();
- assert_eq!(self.empty, empty);
+ assert!(
+ self.ranges.iter().all(|r| !r.is_empty()),
+ "empty buffer: {:?}",
+ self.ranges
+ );
}
}
@@ -554,6 +550,9 @@ impl Length for PushBuffers {
/// less efficient implementation of Read for Buffers
impl std::io::Read for PushBuffers {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+ if buf.is_empty() {
+ return Ok(0);
+ }
// Find the range that contains the start offset
let found = self.find(self.offset, self.offset + buf.len() as u64);
if let Some(idx) = found {
@@ -582,6 +581,10 @@ impl ChunkReader for PushBuffers {
}
fn get_bytes(&self, start: u64, length: usize) -> Result<Bytes,
ParquetError> {
+ // An empty range needs no bytes (see `has_range`)
+ if length == 0 {
+ return Ok(Bytes::new());
+ }
// find the range that contains the start offset
if let Some(idx) = self.find(start, start + length as u64) {
// Found the range, figure out the starting offset in the buffer
@@ -860,7 +863,10 @@ mod tests {
0..4 => {
let range = random_range(&mut rng);
push(&mut buffers, range.clone());
- model.push(range);
+ // An empty push is not stored.
+ if !range.is_empty() {
+ model.push(range);
+ }
}
4..6 => {
// Clear some pushed ranges and some other ranges.
@@ -875,9 +881,10 @@ mod tests {
}
_ => {
let range = random_range(&mut rng);
- let expected = model
- .iter()
- .any(|r| r.start <= range.start && r.end >=
range.end);
+ let expected = range.is_empty()
+ || model
+ .iter()
+ .any(|r| r.start <= range.start && r.end >=
range.end);
assert_eq!(buffers.has_range(&range), expected, "seed
{seed} {range:?}");
if expected {
let len = (range.end - range.start) as usize;
@@ -943,6 +950,40 @@ mod tests {
parts
}
+ #[test]
+ #[expect(clippy::reversed_empty_ranges)]
+ fn empty_ranges() {
+ use std::io::Read;
+ // An empty range is available without any buffers.
+ let mut buffers = PushBuffers::new(100);
+ assert!(buffers.has_range(&(50..50)));
+ assert_eq!(buffers.get_bytes(50, 0).unwrap(), Bytes::new());
+ assert_eq!(buffers.get_read(50).unwrap().read(&mut []).unwrap(), 0);
+
+ // Pushing an empty or an inverted range is a no-op.
+ buffers.push_range(30..30, Bytes::new()).unwrap();
+ buffers.push_range(10..5, Bytes::new()).unwrap();
+ assert!(buffers.ranges.is_empty());
+ assert!(buffers.buffers.is_empty());
+ buffers.assert_invariants();
+
+ // Reads of real data still work after an empty push.
+ buffers
+ .push_range(20..25, Bytes::from_static(b"abcde"))
+ .unwrap();
+ buffers.push_range(22..22, Bytes::new()).unwrap();
+ buffers.assert_invariants();
+ assert_eq!(buffers.ranges, vec![20..25]);
+ assert!(buffers.has_range(&(21..24)));
+ assert_eq!(
+ buffers.get_bytes(21, 3).unwrap(),
+ Bytes::from_static(b"bcd")
+ );
+ let mut buf = [0; 2];
+ buffers.get_read(23).unwrap().read_exact(&mut buf).unwrap();
+ assert_eq!(&buf, b"de");
+ }
+
#[test]
#[cfg(feature = "arrow")]
#[expect(clippy::reversed_empty_ranges)]
