This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-11569-53e7e369fa3db5cb560bf2a92748840e19b7d726 in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
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)]
