Jefffrey commented on code in PR #10287:
URL: https://github.com/apache/arrow-rs/pull/10287#discussion_r3541701790
##########
arrow-array/src/array/byte_view_array.rs:
##########
@@ -556,67 +569,61 @@ impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
let data_blocks = vec![data_block];
(views_buf, data_blocks)
} else {
- // slow path, need to split into multiple buffers
-
- struct GcCopyGroup {
- total_buffer_bytes: usize,
- total_len: usize,
- }
-
- impl GcCopyGroup {
- fn new(total_buffer_bytes: u32, total_len: usize) -> Self {
- Self {
- total_buffer_bytes: total_buffer_bytes as usize,
- total_len,
- }
+ // slow path: the non-inline data does not fit in a single buffer
+ // (a buffer offset is a `u32`, so no buffer may exceed
`i32::MAX`),
+ // so it must be split across several buffers.
+ //
+ // Iterate over *all* views in order — this is essential for
+ // correctness, as gc must preserve the row count. Inline views are
+ // copied unchanged (they reference no buffer); non-inline views
are
+ // packed into the current output buffer until adding the next one
+ // would exceed `max_buffer_size`, at which point a new buffer is
+ // started.
+ //
+ // `total_large` is the *sum* of all non-inline bytes — it is what
+ // pushed us onto this path, so it can be many GiB. Size each
buffer
+ // reservation to the data still to be copied, capped at one
buffer's
+ // worth (`max_buffer_size`), so we neither reserve the whole
+ // multi-GiB total up front nor pad the final buffer out to a full
+ // `max_buffer_size` when only a little data is left.
+ let mut remaining_large = total_large;
+ let mut views_buf = Vec::with_capacity(len);
+ let mut data_blocks: Vec<Buffer> = Vec::new();
+ let mut data_buf: Vec<u8> =
Vec::with_capacity(remaining_large.min(max_buffer_size));
Review Comment:
the way of reserving this much capacity up front can lead to fragmentation
with unused capacity at the end; i guess this is why the initial approach
calculated groups to get exact sizes 🤔
##########
arrow-array/src/array/byte_view_array.rs:
##########
@@ -556,67 +569,61 @@ impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
let data_blocks = vec![data_block];
(views_buf, data_blocks)
} else {
- // slow path, need to split into multiple buffers
-
- struct GcCopyGroup {
- total_buffer_bytes: usize,
- total_len: usize,
- }
-
- impl GcCopyGroup {
- fn new(total_buffer_bytes: u32, total_len: usize) -> Self {
- Self {
- total_buffer_bytes: total_buffer_bytes as usize,
- total_len,
- }
+ // slow path: the non-inline data does not fit in a single buffer
+ // (a buffer offset is a `u32`, so no buffer may exceed
`i32::MAX`),
Review Comment:
technically i think buffers *can* exceed `i32::MAX` safely; if we fill
`i32::MAX - 1` of the buffer, then the last string is also a massive string
(e.g. `i32::MAX`) then its offset is still within `i32` range 🤔
that is, things are fine so long as the starting offset is within `i32`
range, even if its length causes the buffer to exceed `i32::MAX` in length
##########
arrow-array/src/array/byte_view_array.rs:
##########
@@ -556,67 +569,61 @@ impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
let data_blocks = vec![data_block];
(views_buf, data_blocks)
} else {
- // slow path, need to split into multiple buffers
-
- struct GcCopyGroup {
- total_buffer_bytes: usize,
- total_len: usize,
- }
-
- impl GcCopyGroup {
- fn new(total_buffer_bytes: u32, total_len: usize) -> Self {
- Self {
- total_buffer_bytes: total_buffer_bytes as usize,
- total_len,
- }
+ // slow path: the non-inline data does not fit in a single buffer
+ // (a buffer offset is a `u32`, so no buffer may exceed
`i32::MAX`),
+ // so it must be split across several buffers.
+ //
+ // Iterate over *all* views in order — this is essential for
+ // correctness, as gc must preserve the row count. Inline views are
+ // copied unchanged (they reference no buffer); non-inline views
are
+ // packed into the current output buffer until adding the next one
+ // would exceed `max_buffer_size`, at which point a new buffer is
+ // started.
+ //
+ // `total_large` is the *sum* of all non-inline bytes — it is what
+ // pushed us onto this path, so it can be many GiB. Size each
buffer
+ // reservation to the data still to be copied, capped at one
buffer's
+ // worth (`max_buffer_size`), so we neither reserve the whole
+ // multi-GiB total up front nor pad the final buffer out to a full
+ // `max_buffer_size` when only a little data is left.
+ let mut remaining_large = total_large;
+ let mut views_buf = Vec::with_capacity(len);
+ let mut data_blocks: Vec<Buffer> = Vec::new();
+ let mut data_buf: Vec<u8> =
Vec::with_capacity(remaining_large.min(max_buffer_size));
+
+ for i in 0..len {
+ // SAFETY: `i < len == self.len()`.
+ let view_len = *unsafe { self.views().get_unchecked(i) } as
u32;
+ let is_large = view_len > MAX_INLINE_VIEW_LEN;
+ // Only non-inline views consume space in the data buffer; if
the
+ // next one would overflow the current buffer's offset space,
+ // seal it and start a new one. The `!data_buf.is_empty()`
guard
+ // avoids emitting a leading empty buffer and guarantees
forward
+ // progress even for a single value larger than
`max_buffer_size`
Review Comment:
can we clean up these comments; it should not be possible for a single value
to be larger than `max_buffer_size` (unless we're doing testing where
`max_buffer_size < i32::MAX`)
##########
arrow-array/src/array/byte_view_array.rs:
##########
@@ -556,67 +569,61 @@ impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
let data_blocks = vec![data_block];
(views_buf, data_blocks)
} else {
- // slow path, need to split into multiple buffers
-
- struct GcCopyGroup {
- total_buffer_bytes: usize,
- total_len: usize,
- }
-
- impl GcCopyGroup {
- fn new(total_buffer_bytes: u32, total_len: usize) -> Self {
- Self {
- total_buffer_bytes: total_buffer_bytes as usize,
- total_len,
- }
+ // slow path: the non-inline data does not fit in a single buffer
+ // (a buffer offset is a `u32`, so no buffer may exceed
`i32::MAX`),
+ // so it must be split across several buffers.
+ //
+ // Iterate over *all* views in order — this is essential for
+ // correctness, as gc must preserve the row count. Inline views are
+ // copied unchanged (they reference no buffer); non-inline views
are
+ // packed into the current output buffer until adding the next one
+ // would exceed `max_buffer_size`, at which point a new buffer is
+ // started.
+ //
+ // `total_large` is the *sum* of all non-inline bytes — it is what
+ // pushed us onto this path, so it can be many GiB. Size each
buffer
+ // reservation to the data still to be copied, capped at one
buffer's
+ // worth (`max_buffer_size`), so we neither reserve the whole
+ // multi-GiB total up front nor pad the final buffer out to a full
+ // `max_buffer_size` when only a little data is left.
+ let mut remaining_large = total_large;
+ let mut views_buf = Vec::with_capacity(len);
+ let mut data_blocks: Vec<Buffer> = Vec::new();
+ let mut data_buf: Vec<u8> =
Vec::with_capacity(remaining_large.min(max_buffer_size));
+
+ for i in 0..len {
+ // SAFETY: `i < len == self.len()`.
+ let view_len = *unsafe { self.views().get_unchecked(i) } as
u32;
+ let is_large = view_len > MAX_INLINE_VIEW_LEN;
+ // Only non-inline views consume space in the data buffer; if
the
+ // next one would overflow the current buffer's offset space,
+ // seal it and start a new one. The `!data_buf.is_empty()`
guard
+ // avoids emitting a leading empty buffer and guarantees
forward
+ // progress even for a single value larger than
`max_buffer_size`
+ // — in that case `data_buf.len()` itself can exceed
+ // `max_buffer_size`, so the sum is computed in `u64` to avoid
+ // overflowing the `u32` offset space on the addition.
+ if is_large
+ && !data_buf.is_empty()
+ && data_buf.len() as u64 + view_len as u64 >
max_buffer_size as u64
+ {
+ data_blocks.push(Buffer::from_vec(std::mem::take(&mut
data_buf)));
+ data_buf.reserve(remaining_large.min(max_buffer_size));
}
- }
- let mut groups = Vec::new();
- let mut current_length = 0;
- let mut current_elements = 0;
-
- for view in self.views() {
- let len = *view as u32;
- if len > MAX_INLINE_VIEW_LEN {
- if current_length + len > i32::MAX as u32 {
- // Start a new group
- groups.push(GcCopyGroup::new(current_length,
current_elements));
- current_length = 0;
- current_elements = 0;
- }
- current_length += len;
- current_elements += 1;
+ // The buffer currently being filled becomes block
+ // `data_blocks.len()` once it is sealed, so that is the index
+ // the copied views must point at.
+ let buffer_idx = data_blocks.len() as i32;
+ // SAFETY: `i < self.len()` and every view refers to valid
data.
+ let new_view = unsafe { self.copy_view_to_buffer(i,
buffer_idx, &mut data_buf) };
+ views_buf.push(new_view);
+ if is_large {
+ remaining_large -= view_len as usize;
Review Comment:
we can probably push this `remaining_large` adjustment inside the same if
block above where we allocate a new `data_buf`; this way we just subtract based
on previous `data_buf` instead of on each iteration
##########
arrow-array/src/array/byte_view_array.rs:
##########
@@ -556,67 +569,47 @@ impl<T: ByteViewType + ?Sized> GenericByteViewArray<T> {
let data_blocks = vec![data_block];
(views_buf, data_blocks)
} else {
- // slow path, need to split into multiple buffers
-
- struct GcCopyGroup {
- total_buffer_bytes: usize,
- total_len: usize,
- }
-
- impl GcCopyGroup {
- fn new(total_buffer_bytes: u32, total_len: usize) -> Self {
- Self {
- total_buffer_bytes: total_buffer_bytes as usize,
- total_len,
- }
- }
- }
-
- let mut groups = Vec::new();
- let mut current_length = 0;
- let mut current_elements = 0;
-
- for view in self.views() {
- let len = *view as u32;
- if len > MAX_INLINE_VIEW_LEN {
- if current_length + len > i32::MAX as u32 {
- // Start a new group
- groups.push(GcCopyGroup::new(current_length,
current_elements));
- current_length = 0;
- current_elements = 0;
- }
- current_length += len;
- current_elements += 1;
+ // slow path: the non-inline data does not fit in a single buffer
+ // (a buffer offset is a `u32`, so no buffer may exceed
`i32::MAX`),
+ // so it must be split across several buffers.
+ //
+ // Iterate over *all* views in order — this is essential for
+ // correctness, as gc must preserve the row count. Inline views are
+ // copied unchanged (they reference no buffer); non-inline views
are
+ // packed into the current output buffer until adding the next one
+ // would exceed `max_buffer_size`, at which point a new buffer is
+ // started. Cap the per-buffer reservation at `max_buffer_size` so
a
+ // single huge value doesn't request an absurd allocation up front.
+ let buffer_cap = total_large.min(max_buffer_size);
+ let mut views_buf = Vec::with_capacity(len);
+ let mut data_blocks = Vec::new();
+ let mut data_buf: Vec<u8> = Vec::with_capacity(buffer_cap);
+ let mut current_buffer_idx: i32 = 0;
+
+ for i in 0..len {
+ // SAFETY: `i < len == self.len()`.
+ let view_len = *unsafe { self.views().get_unchecked(i) } as
u32;
+ // Only non-inline views consume space in the data buffer; if
the
+ // next one would overflow the current buffer's offset space,
+ // seal it and start a new one. The `!data_buf.is_empty()`
guard
+ // avoids emitting a leading empty buffer and guarantees
forward
+ // progress even for a single value larger than
`max_buffer_size`.
+ if view_len > MAX_INLINE_VIEW_LEN
+ && !data_buf.is_empty()
+ && data_buf.len() as u64 + view_len as u64 >
max_buffer_size as u64
Review Comment:
`data_buf.len()` cannot exceed `max_buffer_len` here, i think you mean
`data_buf.len() + view_len` can exceed 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
