alamb commented on code in PR #11459:
URL: https://github.com/apache/arrow-rs/pull/11459#discussion_r4275336653
##########
parquet/src/util/push_buffers.rs:
##########
@@ -229,45 +250,78 @@ impl PushBuffers {
}
/// Clear any range and corresponding buffer that is exactly in the
ranges_to_clear
+ ///
+ /// A binary search finds the buffers that start at the start of each
Review Comment:
Maybe we could more concisely explain this is `O(ln(N))` in the number of
buffers, rather than `O(n)` 🤔
##########
parquet/src/util/push_buffers.rs:
##########
@@ -54,10 +56,13 @@ pub struct PushBuffers {
file_len: u64,
/// The ranges of data that are available for decoding (not adjusted for
/// offset), sorted by `start`
- ranges: Vec<Range<u64>>,
+ ///
+ /// A `VecDeque`, because a scan pushes buffers at the end and releases
+ /// them at the start. Both are then `O(1)` for each buffer.
Review Comment:
this is a good call and it will set us up well
##########
parquet/src/util/push_buffers.rs:
##########
@@ -66,9 +71,18 @@ pub struct PushBuffers {
/// when no earlier buffer can reach the requested range, so it checks one
/// or two buffers, not all of them. See [`Self::find`].
max_len: u64,
+ /// The number of ranges in `ranges` with the length `max_len`. `max_len`
+ /// can decrease only when this count goes to zero. Thus, a release does
+ /// not scan all ranges to update `max_len` while other ranges of that
+ /// length remain.
+ max_len_count: usize,
/// 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
Review Comment:
We could perhaps simplify the implementation if we simply strip out pushing
empty ranges on push and ignore them on release? maybe as a follow on PR
##########
parquet/src/util/push_buffers.rs:
##########
@@ -351,30 +430,49 @@ impl PushBuffers {
}
}
}
- if emptied {
+ if emptied || self.empty > 0 {
// Remove the released buffers. As before, this also removes empty
- // buffers that were pushed. `retain` visits each element exactly
- // once in the original order, so both calls keep the same indices.
- let mut keep = self.ranges.iter().map(|range| !range.is_empty());
- self.buffers.retain(|_| keep.next().unwrap());
- self.ranges.retain(|range| !range.is_empty());
- }
- for (range, buffer) in split {
- self.ranges.push(range);
- self.buffers.push(buffer);
+ // buffers that were pushed. The released buffers are all in
+ // `first..=last`, so look at all buffers only if empty buffers
+ // were pushed. `drain` moves the buffers on the shorter side of
+ // the removed ones, so a release at the start or at the end of
+ // the buffers moves none.
+ let span = match self.empty {
+ 0 => first..last + 1,
+ _ => 0..self.ranges.len(),
+ };
+ let mut kept = span.start;
+ for idx in span.clone() {
+ if !self.ranges[idx].is_empty() {
+ self.ranges.swap(kept, idx);
+ self.buffers.swap(kept, idx);
+ kept += 1;
+ }
+ }
+ self.ranges.drain(kept..span.end);
+ self.buffers.drain(kept..span.end);
+ self.empty = 0;
}
- // Split parts are appended, and if buffers overlap, the tail of a
- // trimmed buffer can start after the start of the next buffer. The
- // sort is stable.
- if !self.ranges.is_sorted_by_key(|range| range.start) {
+ // 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,
+ // or is empty and removed. Thus, the buffers are still sorted, except
+ // for the split parts, which are appended. Splits are rare, so sort
+ // all buffers then. The sort is stable.
Review Comment:
What values does the "the sort is stable" note add?
##########
parquet/src/util/push_buffers.rs:
##########
@@ -281,38 +335,63 @@ impl PushBuffers {
/// The buffers stay sorted by start, which lookups require (see
/// [`Self::find`]). If no buffer overlaps `ranges`, the buffers do not
/// change.
+ ///
+ /// A binary search finds the buffers that can overlap `ranges`, as in
Review Comment:
I think it would be more concise to document it is O(ln(N)) rather than
describe te internal implementation details
##########
parquet/src/util/push_buffers.rs:
##########
@@ -66,9 +71,18 @@ pub struct PushBuffers {
/// when no earlier buffer can reach the requested range, so it checks one
/// or two buffers, not all of them. See [`Self::find`].
max_len: u64,
+ /// The number of ranges in `ranges` with the length `max_len`. `max_len`
Review Comment:
maybe as a follow on `max_len` and `max_len_count` could be combined into a
single structure or something so they are guaranteed to stay in sync 🤔
##########
parquet/src/util/push_buffers.rs:
##########
@@ -351,30 +430,49 @@ impl PushBuffers {
}
}
}
- if emptied {
+ if emptied || self.empty > 0 {
// Remove the released buffers. As before, this also removes empty
- // buffers that were pushed. `retain` visits each element exactly
- // once in the original order, so both calls keep the same indices.
- let mut keep = self.ranges.iter().map(|range| !range.is_empty());
- self.buffers.retain(|_| keep.next().unwrap());
- self.ranges.retain(|range| !range.is_empty());
- }
- for (range, buffer) in split {
- self.ranges.push(range);
- self.buffers.push(buffer);
+ // buffers that were pushed. The released buffers are all in
+ // `first..=last`, so look at all buffers only if empty buffers
+ // were pushed. `drain` moves the buffers on the shorter side of
Review Comment:
This note about what buffers are drained seems to be an implementation
detail that isn't particularly relevant 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
