adriangb commented on code in PR #11235:
URL: https://github.com/apache/arrow-rs/pull/11235#discussion_r4214518940
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -527,6 +535,21 @@ impl ParquetPushDecoder {
/// Note this can be the entire file or just a part of it. If it is part
of the file,
/// the ranges should correspond to the data ranges requested by the
decoder.
///
+ /// When the decoder is done with a row group that it reads, it releases
Review Comment:
Fixed in
https://github.com/apache/arrow-rs/pull/11235/commits/14c209bd03863a0020d152f6e380fcba32fa6ae4
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -527,6 +535,21 @@ impl ParquetPushDecoder {
/// Note this can be the entire file or just a part of it. If it is part
of the file,
/// the ranges should correspond to the data ranges requested by the
decoder.
///
+ /// When the decoder is done with a row group that it reads, it releases
+ /// the pushed bytes of all column chunks of that row group, unless the
Review Comment:
Fixed in
https://github.com/apache/arrow-rs/pull/11235/commits/14c209bd03863a0020d152f6e380fcba32fa6ae4
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -527,6 +535,21 @@ impl ParquetPushDecoder {
/// Note this can be the entire file or just a part of it. If it is part
of the file,
/// the ranges should correspond to the data ranges requested by the
decoder.
///
+ /// When the decoder is done with a row group that it reads, it releases
+ /// the pushed bytes of all column chunks of that row group, unless the
+ /// scan reads the row group again. This is also true for bytes that the
Review Comment:
Fixed in
https://github.com/apache/arrow-rs/pull/11235/commits/14c209bd03863a0020d152f6e380fcba32fa6ae4
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -527,6 +535,21 @@ impl ParquetPushDecoder {
/// Note this can be the entire file or just a part of it. If it is part
of the file,
/// the ranges should correspond to the data ranges requested by the
decoder.
///
+ /// When the decoder is done with a row group that it reads, it releases
+ /// the pushed bytes of all column chunks of that row group, unless the
+ /// scan reads the row group again. This is also true for bytes that the
+ /// decoder did not request, and for bytes in a buffer that is larger than
+ /// a requested range. The decoder does not release the bytes of row
+ /// groups that it does not read (for example, row groups without selected
+ /// rows, or after the limit). Call [`Self::clear_all_ranges`] to release
+ /// them.
+ ///
+ /// Releasing bytes decreases [`Self::buffered_bytes`], but the memory of
+ /// a pushed [`Bytes`] is freed only after all zero-copy slices of it are
Review Comment:
Fixed in
https://github.com/apache/arrow-rs/pull/11235/commits/14c209bd03863a0020d152f6e380fcba32fa6ae4
##########
parquet/src/arrow/push_decoder/mod.rs:
##########
@@ -666,9 +689,10 @@ impl ParquetPushDecoder {
///
/// The decoder's buffered bytes are carried across the rebuild: bytes
/// already fetched for row groups the new configuration still reads are
- /// not re-requested. Bytes the new configuration no longer needs stay
- /// buffered until [`clear_all_ranges`](Self::clear_all_ranges) is called
- /// or the rebuilt decoder is dropped.
+ /// not re-requested. [`build`](ParquetPushDecoderBuilder::build) releases
+ /// the bytes that the new configuration does not read: the bytes outside
Review Comment:
Fixed in
https://github.com/apache/arrow-rs/pull/11235/commits/14c209bd03863a0020d152f6e380fcba32fa6ae4
##########
parquet/src/util/push_buffers.rs:
##########
@@ -180,15 +185,156 @@ impl PushBuffers {
new_buffers.push(buffer.clone());
}
}
+ self.buffered_bytes = new_ranges.iter().map(|r| r.end - r.start).sum();
self.ranges = new_ranges;
self.buffers = new_buffers;
}
+ /// Remove all buffered bytes in `ranges`, whatever the shape of the
Review Comment:
Fixed in
https://github.com/apache/arrow-rs/pull/11235/commits/14c209bd03863a0020d152f6e380fcba32fa6ae4
##########
parquet/src/util/push_buffers.rs:
##########
@@ -286,4 +552,137 @@ mod tests {
"Parquet error: Number of ranges (2) must match number of buffers
(1)"
);
}
+
+ /// The result of `release_ranges`, computed by rebuilding all buffers.
+ #[cfg(feature = "arrow")]
+ fn release_by_rebuild(
+ buffers: &PushBuffers,
+ release: &[Range<u64>],
+ ) -> Vec<(Range<u64>, Bytes)> {
+ let release = merge_ranges(release);
+ let mut parts = vec![];
+ for (range, buffer) in buffers.iter() {
+ let mut start = range.start;
+ let mut keep = |part: Range<u64>| {
+ let offset = |pos: u64| (pos - range.start) as usize;
+ parts.push((
+ part.clone(),
+ buffer.slice(offset(part.start)..offset(part.end)),
+ ));
+ };
+ for r in release
+ .iter()
+ .filter(|r| r.start < range.end && r.end > range.start)
+ {
+ if start < r.start {
+ keep(start..r.start);
+ }
+ start = start.max(r.end);
+ }
+ if start < range.end {
+ keep(start..range.end);
+ }
+ }
+ parts.sort_by_key(|(range, _)| range.start);
+ parts
+ }
+
+ #[test]
+ #[cfg(feature = "arrow")]
+ #[expect(clippy::reversed_empty_ranges)]
+ fn release_ranges_splits_and_removes_empty_buffers() {
+ let mut buffers = PushBuffers::new(100);
+ buffers
+ .push_range(0..20, Bytes::from_static(b"abcdefghijklmnopqrst"))
+ .unwrap();
+ // An empty and an inverted range, which `push_range` accepts.
+ buffers.push_range(30..30, Bytes::new()).unwrap();
+ buffers.push_range(10..5, Bytes::new()).unwrap();
+
+ // Split one buffer into three parts.
+ buffers.release_ranges(&[3..5, 8..10]);
+ assert_eq!(buffers.ranges, vec![0..3, 5..8, 10..20]);
+ assert_eq!(buffers.buffered_bytes(), 16);
+ assert_eq!(buffers.get_bytes(5, 3).unwrap(),
Bytes::from_static(b"fgh"));
+
+ // Release a whole buffer.
+ release(&mut buffers, 5..8);
+ assert_eq!(buffers.ranges, vec![0..3, 10..20]);
+ assert_eq!(buffers.buffered_bytes(), 13);
+ }
+
+ #[test]
+ #[cfg(feature = "arrow")]
+ fn release_ranges_matches_rebuild() {
Review Comment:
Fixed in
https://github.com/apache/arrow-rs/pull/11235/commits/14c209bd03863a0020d152f6e380fcba32fa6ae4
--
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]
