1use std::alloc::Layout;
19use std::mem;
20use std::ptr::NonNull;
21
22use crate::alloc::{ALIGNMENT, Deallocation};
23use crate::{
24 bytes::Bytes,
25 native::{ArrowNativeType, ToByteSlice},
26 util::bit_util,
27};
28
29#[cfg(feature = "pool")]
30use crate::pool::{MemoryPool, TrackedReservation};
31
32use super::Buffer;
33
34#[derive(Debug, Clone, PartialEq, Eq)]
36pub enum MutableBufferError {
37 LengthOverflow,
39 LayoutError,
41 AllocationError(Layout),
43}
44
45impl std::fmt::Display for MutableBufferError {
46 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
47 match self {
48 Self::LengthOverflow => write!(f, "buffer length overflow"),
49 Self::LayoutError => write!(f, "invalid allocation layout for requested capacity"),
50 Self::AllocationError(layout) => {
51 write!(f, "failed to allocate memory for layout {layout:?}")
52 }
53 }
54 }
55}
56
57impl std::error::Error for MutableBufferError {}
58
59#[derive(Debug)]
122pub struct MutableBuffer {
123 data: NonNull<u8>,
125 len: usize,
127 layout: Layout,
128
129 #[cfg(feature = "pool")]
131 reservation: TrackedReservation,
132}
133
134impl MutableBuffer {
135 #[inline]
143 pub fn new(capacity: usize) -> Self {
144 Self::try_with_capacity(capacity).unwrap_or_else(|e| panic!("{e}"))
145 }
146
147 #[inline]
154 pub fn with_capacity(capacity: usize) -> Self {
155 Self::try_with_capacity(capacity).unwrap_or_else(|e| panic!("{e}"))
156 }
157
158 #[inline]
160 pub fn try_with_capacity(capacity: usize) -> Result<Self, MutableBufferError> {
161 let capacity = capacity
162 .checked_next_multiple_of(64)
163 .ok_or(MutableBufferError::LayoutError)?;
164 let layout = Layout::from_size_align(capacity, ALIGNMENT)
165 .map_err(|_| MutableBufferError::LayoutError)?;
166 let data = match layout.size() {
167 0 => dangling_ptr(),
168 _ => {
169 let raw_ptr = unsafe { std::alloc::alloc(layout) };
171 match NonNull::new(raw_ptr) {
172 Some(ptr) => ptr,
173 None => return Err(MutableBufferError::AllocationError(layout)),
174 }
175 }
176 };
177 Ok(Self {
178 data,
179 len: 0,
180 layout,
181 #[cfg(feature = "pool")]
182 reservation: TrackedReservation::default(),
183 })
184 }
185
186 pub fn from_len_zeroed(len: usize) -> Self {
202 Self::try_from_len_zeroed(len).unwrap_or_else(|e| panic!("{e}"))
203 }
204
205 pub fn try_from_len_zeroed(len: usize) -> Result<Self, MutableBufferError> {
207 let layout =
208 Layout::from_size_align(len, ALIGNMENT).map_err(|_| MutableBufferError::LayoutError)?;
209 let data = match layout.size() {
210 0 => dangling_ptr(),
211 _ => {
212 let raw_ptr = unsafe { std::alloc::alloc_zeroed(layout) };
214 match NonNull::new(raw_ptr) {
215 Some(ptr) => ptr,
216 None => return Err(MutableBufferError::AllocationError(layout)),
217 }
218 }
219 };
220 Ok(Self {
221 data,
222 len,
223 layout,
224 #[cfg(feature = "pool")]
225 reservation: TrackedReservation::default(),
226 })
227 }
228
229 pub(crate) fn from_bytes(bytes: Bytes) -> Result<Self, Bytes> {
231 let layout = match bytes.deallocation() {
232 Deallocation::Standard(layout) => *layout,
233 Deallocation::Custom(..) => return Err(bytes),
234 };
235
236 let len = bytes.len();
237 let data = bytes.ptr();
238 #[cfg(feature = "pool")]
239 let reservation = bytes.reservation.take();
240
241 mem::forget(bytes);
242
243 Ok(Self {
244 data,
245 len,
246 layout,
247 #[cfg(feature = "pool")]
248 reservation,
249 })
250 }
251
252 pub fn new_null(len: usize) -> Self {
259 let num_bytes = bit_util::ceil(len, 8);
260 MutableBuffer::from_len_zeroed(num_bytes)
261 }
262
263 pub fn with_bitset(mut self, end: usize, val: bool) -> Self {
274 assert!(end <= self.layout.size());
275 let v = if val { 255 } else { 0 };
276 unsafe {
277 std::ptr::write_bytes(self.data.as_ptr(), v, end);
278 self.len = end;
279 }
280 self
281 }
282
283 pub fn set_null_bits(&mut self, start: usize, count: usize) {
293 assert!(
294 start.saturating_add(count) <= self.layout.size(),
295 "range start index {start} and count {count} out of bounds for \
296 buffer of length {}",
297 self.layout.size(),
298 );
299
300 unsafe {
302 std::ptr::write_bytes(self.data.as_ptr().add(start), 0, count);
303 }
304 }
305
306 #[inline]
308 pub fn try_reserve(&mut self, additional: usize) -> Result<(), MutableBufferError> {
309 let required_cap = self
310 .len
311 .checked_add(additional)
312 .ok_or(MutableBufferError::LengthOverflow)?;
313 if required_cap > self.layout.size() {
314 let new_capacity = required_cap
315 .checked_next_multiple_of(64)
316 .ok_or(MutableBufferError::LayoutError)?;
317 let new_capacity = std::cmp::max(new_capacity, self.layout.size().saturating_mul(2));
318 self.try_reallocate(new_capacity)?;
319 }
320 Ok(())
321 }
322 #[inline(always)]
341 pub fn reserve(&mut self, additional: usize) {
342 self.try_reserve(additional)
343 .unwrap_or_else(|e| panic!("{e}"))
344 }
345
346 pub fn try_repeat_slice_n_times<T: ArrowNativeType>(
348 &mut self,
349 slice_to_repeat: &[T],
350 repeat_count: usize,
351 ) -> Result<(), MutableBufferError> {
352 if repeat_count == 0 || slice_to_repeat.is_empty() {
353 return Ok(());
354 }
355 let bytes_per_copy = size_of_val(slice_to_repeat);
356 let total_bytes = repeat_count
357 .checked_mul(bytes_per_copy)
358 .ok_or(MutableBufferError::LengthOverflow)?;
359 self.len
360 .checked_add(total_bytes)
361 .ok_or(MutableBufferError::LengthOverflow)?;
362
363 self.try_reserve(total_bytes)?;
365
366 let length_before = self.len;
368
369 self.try_extend_from_slice(slice_to_repeat)?;
371
372 let mut already_repeated = 1usize;
374
375 while already_repeated < repeat_count {
377 let to_copy = already_repeated.min(repeat_count - already_repeated);
380 let byte_count = to_copy * bytes_per_copy;
381 unsafe {
382 let src = self.data.as_ptr().add(length_before).cast_const();
384 let dst = self.data.as_ptr().add(self.len);
386 std::ptr::copy_nonoverlapping(src, dst, byte_count);
388 }
389 self.len += byte_count;
391 already_repeated += to_copy;
392 }
393 Ok(())
394 }
395 pub fn repeat_slice_n_times<T: ArrowNativeType>(
414 &mut self,
415 slice_to_repeat: &[T],
416 repeat_count: usize,
417 ) {
418 self.try_repeat_slice_n_times(slice_to_repeat, repeat_count)
419 .unwrap_or_else(|e| panic!("{e}"))
420 }
421
422 #[cold]
423 fn try_reallocate(&mut self, capacity: usize) -> Result<(), MutableBufferError> {
424 let new_layout = Layout::from_size_align(capacity, self.layout.align())
425 .map_err(|_| MutableBufferError::LayoutError)?;
426
427 if new_layout.size() == 0 {
428 if self.layout.size() != 0 {
429 unsafe { std::alloc::dealloc(self.as_mut_ptr(), self.layout) };
431 self.layout = new_layout;
432 }
433 return Ok(());
434 }
435
436 let data = match self.layout.size() {
437 0 => unsafe { std::alloc::alloc(new_layout) },
439 _ => unsafe { std::alloc::realloc(self.as_mut_ptr(), self.layout, capacity) },
441 };
442 self.data = match NonNull::new(data) {
443 Some(ptr) => ptr,
444 None => return Err(MutableBufferError::AllocationError(new_layout)),
445 };
446 self.layout = new_layout;
447 #[cfg(feature = "pool")]
448 self.reservation.resize(self.layout.size());
449 Ok(())
450 }
451 #[inline(always)]
455 pub fn truncate(&mut self, len: usize) {
456 if len > self.len {
457 return;
458 }
459 self.len = len;
460 #[cfg(feature = "pool")]
461 self.reservation.resize(self.len);
462 }
463
464 #[inline]
466 pub fn try_resize(&mut self, new_len: usize, value: u8) -> Result<(), MutableBufferError> {
467 if new_len > self.len {
468 let diff = new_len - self.len;
469 self.try_reserve(diff)?;
470 unsafe { self.data.as_ptr().add(self.len).write_bytes(value, diff) };
473 }
474 self.len = new_len;
476 #[cfg(feature = "pool")]
477 self.reservation.resize(self.len);
478 Ok(())
479 }
480 #[inline(always)]
497 pub fn resize(&mut self, new_len: usize, value: u8) {
498 self.try_resize(new_len, value)
499 .unwrap_or_else(|e| panic!("{e}"))
500 }
501
502 pub fn try_shrink_to_fit(&mut self) -> Result<(), MutableBufferError> {
504 let new_capacity = self
505 .len
506 .checked_next_multiple_of(64)
507 .ok_or(MutableBufferError::LayoutError)?;
508 if new_capacity < self.layout.size() {
509 self.try_reallocate(new_capacity)?;
510 }
511 Ok(())
512 }
513 pub fn shrink_to_fit(&mut self) {
534 self.try_shrink_to_fit().unwrap_or_else(|e| panic!("{e}"))
535 }
536
537 #[inline]
539 pub const fn is_empty(&self) -> bool {
540 self.len == 0
541 }
542
543 #[inline]
546 pub const fn len(&self) -> usize {
547 self.len
548 }
549
550 #[inline]
554 pub const fn capacity(&self) -> usize {
555 self.layout.size()
556 }
557
558 pub fn clear(&mut self) {
560 self.len = 0;
561 #[cfg(feature = "pool")]
562 self.reservation.resize(self.len);
563 }
564
565 pub fn as_slice(&self) -> &[u8] {
567 self
568 }
569
570 pub fn as_slice_mut(&mut self) -> &mut [u8] {
572 self
573 }
574
575 #[inline]
578 pub const fn as_ptr(&self) -> *const u8 {
579 self.data.as_ptr()
580 }
581
582 #[inline]
585 pub fn as_mut_ptr(&mut self) -> *mut u8 {
586 self.data.as_ptr()
587 }
588
589 #[inline]
590 pub(super) fn into_buffer(self) -> Buffer {
591 let bytes = unsafe { Bytes::new(self.data, self.len, Deallocation::Standard(self.layout)) };
592 #[cfg(feature = "pool")]
593 bytes.reservation.replace(self.reservation.take());
594 std::mem::forget(self);
595 Buffer::from(bytes)
596 }
597
598 pub fn typed_data_mut<T: ArrowNativeType>(&mut self) -> &mut [T] {
605 let (prefix, offsets, suffix) = unsafe { self.as_slice_mut().align_to_mut::<T>() };
609 assert!(prefix.is_empty() && suffix.is_empty());
610 offsets
611 }
612
613 pub fn typed_data<T: ArrowNativeType>(&self) -> &[T] {
620 let (prefix, offsets, suffix) = unsafe { self.as_slice().align_to::<T>() };
624 assert!(prefix.is_empty() && suffix.is_empty());
625 offsets
626 }
627
628 #[inline]
630 pub fn try_extend_from_slice<T: ArrowNativeType>(
631 &mut self,
632 items: &[T],
633 ) -> Result<(), MutableBufferError> {
634 let additional = mem::size_of_val(items);
635 self.try_reserve(additional)?;
636 unsafe {
637 let src = items.as_ptr().cast::<u8>();
641 let dst = self.data.as_ptr().add(self.len);
642 std::ptr::copy_nonoverlapping(src, dst, additional);
643 }
644 self.len += additional;
645 Ok(())
646 }
647 #[inline]
661 pub fn extend_from_slice<T: ArrowNativeType>(&mut self, items: &[T]) {
662 self.try_extend_from_slice(items)
663 .unwrap_or_else(|e| panic!("{e}"))
664 }
665
666 #[inline]
680 pub fn push<T: ToByteSlice>(&mut self, item: T) {
681 let additional = std::mem::size_of::<T>();
682 self.reserve(additional);
683 unsafe {
684 let src = item.to_byte_slice().as_ptr();
685 let dst = self.data.as_ptr().add(self.len);
686 std::ptr::copy_nonoverlapping(src, dst, additional);
687 }
688 self.len += additional;
689 }
690
691 #[inline]
695 pub unsafe fn push_unchecked<T: ToByteSlice>(&mut self, item: T) {
696 let additional = std::mem::size_of::<T>();
697 let src = item.to_byte_slice().as_ptr();
698 let dst = unsafe { self.data.as_ptr().add(self.len) };
699 unsafe { std::ptr::copy_nonoverlapping(src, dst, additional) };
700 self.len += additional;
701 }
702
703 #[inline]
705 pub fn try_extend_zeros(&mut self, additional: usize) -> Result<(), MutableBufferError> {
706 let new_len = self
707 .len
708 .checked_add(additional)
709 .ok_or(MutableBufferError::LengthOverflow)?;
710 self.try_resize(new_len, 0)
711 }
712 #[inline]
719 pub fn extend_zeros(&mut self, additional: usize) {
720 self.try_extend_zeros(additional)
721 .unwrap_or_else(|e| panic!("{e}"))
722 }
723
724 #[inline]
731 pub unsafe fn set_len(&mut self, len: usize) {
732 assert!(len <= self.capacity());
733 self.len = len;
734 }
735
736 #[inline]
746 pub fn collect_bool<F: FnMut(usize) -> bool>(len: usize, f: F) -> Self {
747 Self::try_collect_bool(len, f).unwrap_or_else(|e| panic!("{e}"))
748 }
749
750 #[inline]
755 pub fn try_collect_bool<F: FnMut(usize) -> bool>(
756 len: usize,
757 mut f: F,
758 ) -> Result<Self, MutableBufferError> {
759 let words = bit_util::ceil(len, 64);
760 let layout = Layout::array::<u64>(words).map_err(|_| MutableBufferError::LayoutError)?;
761 let mut buffer: Vec<u64> = Vec::new();
762 buffer
763 .try_reserve(words)
764 .map_err(|_| MutableBufferError::AllocationError(layout))?;
765
766 let chunks = len / 64;
767 let remainder = len % 64;
768 buffer.extend((0..chunks).map(|chunk| {
769 let mut packed = 0;
770 for bit_idx in 0..64 {
771 let i = bit_idx + chunk * 64;
772 packed |= (f(i) as u64) << bit_idx;
773 }
774
775 packed
776 }));
777
778 if remainder != 0 {
779 let mut packed = 0;
780 for bit_idx in 0..remainder {
781 let i = bit_idx + chunks * 64;
782 packed |= (f(i) as u64) << bit_idx;
783 }
784
785 buffer.push(packed)
786 }
787
788 let mut buffer: MutableBuffer = buffer.into();
789 buffer.truncate(bit_util::ceil(len, 8));
790 Ok(buffer)
791 }
792
793 #[inline]
812 pub unsafe fn extend_bool_trusted_len<I: Iterator<Item = bool>>(
813 &mut self,
814 mut iter: I,
815 offset: usize,
816 ) {
817 let (lower, upper) = iter.size_hint();
818 let len = upper.expect("Iterator must have exact size_hint");
819 assert_eq!(lower, len, "Iterator must have exact size_hint");
820 debug_assert!(
821 offset <= self.len * 8,
822 "offset must be <= buffer length in bits"
823 );
824
825 if len == 0 {
826 return;
827 }
828
829 let start_len = offset;
830 let end_bit = start_len + len;
831
832 let new_len_bytes = bit_util::ceil(end_bit, 8);
834 if new_len_bytes > self.len {
835 self.reserve(new_len_bytes - self.len);
836 unsafe { self.set_len(new_len_bytes) };
838 }
839
840 let slice = self.as_slice_mut();
841
842 let mut bit_idx = start_len;
843
844 let misalignment = bit_idx & 63;
846 let prefix_bits = if misalignment == 0 {
847 0
848 } else {
849 (64 - misalignment).min(end_bit - bit_idx)
850 };
851
852 if prefix_bits != 0 {
853 let byte_start = bit_idx / 8;
854 let byte_end = bit_util::ceil(bit_idx + prefix_bits, 8);
855 let bit_offset = bit_idx % 8;
856
857 if bit_offset != 0 {
859 let keep_mask = (1u8 << bit_offset).wrapping_sub(1);
860 slice[byte_start] &= keep_mask;
861 }
862
863 let zero_from = if bit_offset == 0 {
865 byte_start
866 } else {
867 byte_start + 1
868 };
869 if byte_end > zero_from {
870 slice[zero_from..byte_end].fill(0);
871 }
872
873 for _ in 0..prefix_bits {
874 let v = iter.next().unwrap();
875 if v {
876 let byte_idx = bit_idx / 8;
877 let bit = bit_idx % 8;
878 slice[byte_idx] |= 1 << bit;
879 }
880 bit_idx += 1;
881 }
882 }
883
884 if bit_idx < end_bit {
885 debug_assert_eq!(bit_idx & 63, 0);
887 let remaining_bits = end_bit - bit_idx;
888 let chunks = remaining_bits / 64;
889
890 let words_start = bit_idx / 8;
891 let words_end = words_start + chunks * 8;
892 for dst in slice[words_start..words_end].as_chunks_mut::<8>().0 {
893 let mut packed: u64 = 0;
894 for i in 0..64 {
895 packed |= (iter.next().unwrap() as u64) << i;
896 }
897 dst.copy_from_slice(&packed.to_le_bytes());
898 bit_idx += 64;
899 }
900
901 let suffix_bits = end_bit - bit_idx;
903 if suffix_bits != 0 {
904 debug_assert_eq!(bit_idx % 8, 0);
905 let byte_start = bit_idx / 8;
906 let byte_end = bit_util::ceil(end_bit, 8);
907 slice[byte_start..byte_end].fill(0);
908
909 for _ in 0..suffix_bits {
910 let v = iter.next().unwrap();
911 if v {
912 let byte_idx = bit_idx / 8;
913 let bit = bit_idx % 8;
914 slice[byte_idx] |= 1 << bit;
915 }
916 bit_idx += 1;
917 }
918 }
919 }
920
921 let remainder = end_bit % 8;
923 if remainder != 0 {
924 let mask = (1u8 << remainder).wrapping_sub(1);
925 slice[bit_util::ceil(end_bit, 8) - 1] &= mask;
926 }
927
928 debug_assert_eq!(bit_idx, end_bit);
929 }
930
931 #[cfg(feature = "pool")]
938 pub fn claim(&self, pool: &dyn MemoryPool) {
939 self.reservation.claim(pool, self.capacity());
940 }
941}
942
943#[inline]
947pub(crate) fn dangling_ptr() -> NonNull<u8> {
948 #[cfg(miri)]
952 {
953 unsafe { NonNull::new_unchecked(std::ptr::without_provenance_mut(ALIGNMENT)) }
955 }
956 #[cfg(not(miri))]
957 {
958 unsafe { NonNull::new_unchecked(ALIGNMENT as *mut u8) }
959 }
960}
961
962impl<A: ArrowNativeType> Extend<A> for MutableBuffer {
963 #[inline]
964 fn extend<T: IntoIterator<Item = A>>(&mut self, iter: T) {
965 let iterator = iter.into_iter();
966 self.extend_from_iter(iterator)
967 }
968}
969
970impl<T: ArrowNativeType> From<Vec<T>> for MutableBuffer {
971 fn from(mut value: Vec<T>) -> Self {
972 let data = unsafe { NonNull::new_unchecked(value.as_mut_ptr().cast()) };
975 let len = value.len() * mem::size_of::<T>();
976 let layout = unsafe { Layout::array::<T>(value.capacity()).unwrap_unchecked() };
980 mem::forget(value);
981 Self {
982 data,
983 len,
984 layout,
985 #[cfg(feature = "pool")]
986 reservation: TrackedReservation::default(),
987 }
988 }
989}
990
991impl MutableBuffer {
992 #[inline]
993 pub(super) fn extend_from_iter<T: ArrowNativeType, I: Iterator<Item = T>>(
994 &mut self,
995 mut iterator: I,
996 ) {
997 let item_size = std::mem::size_of::<T>();
998 let (lower, _) = iterator.size_hint();
999 let additional = lower * item_size;
1000 self.reserve(additional);
1001
1002 let mut len = SetLenOnDrop::new(&mut self.len);
1004 let mut dst = unsafe { self.data.as_ptr().add(len.local_len) };
1005 let capacity = self.layout.size();
1006
1007 while len.local_len + item_size <= capacity {
1008 if let Some(item) = iterator.next() {
1009 unsafe {
1010 let src = item.to_byte_slice().as_ptr();
1011 std::ptr::copy_nonoverlapping(src, dst, item_size);
1012 dst = dst.add(item_size);
1013 }
1014 len.local_len += item_size;
1015 } else {
1016 break;
1017 }
1018 }
1019 drop(len);
1020
1021 iterator.for_each(|item| self.push(item));
1022 }
1023
1024 #[inline]
1049 pub unsafe fn from_trusted_len_iter<T: ArrowNativeType, I: Iterator<Item = T>>(
1050 iterator: I,
1051 ) -> Self {
1052 let item_size = std::mem::size_of::<T>();
1053 let (_, upper) = iterator.size_hint();
1054 let upper = upper.expect("from_trusted_len_iter requires an upper limit");
1055 let len = upper * item_size;
1056
1057 let mut buffer = MutableBuffer::new(len);
1058
1059 let mut dst = buffer.data.as_ptr();
1060 for item in iterator {
1061 let src = item.to_byte_slice().as_ptr();
1063 unsafe { std::ptr::copy_nonoverlapping(src, dst, item_size) };
1064 dst = unsafe { dst.add(item_size) };
1065 }
1066 assert_eq!(
1067 unsafe { dst.offset_from(buffer.data.as_ptr()) } as usize,
1068 len,
1069 "Trusted iterator length was not accurately reported"
1070 );
1071 buffer.len = len;
1072 buffer
1073 }
1074
1075 #[inline]
1099 pub unsafe fn from_trusted_len_iter_bool<I: Iterator<Item = bool>>(mut iterator: I) -> Self {
1100 let (_, upper) = iterator.size_hint();
1101 let len = upper.expect("from_trusted_len_iter requires an upper limit");
1102
1103 Self::collect_bool(len, |_| iterator.next().unwrap())
1104 }
1105
1106 #[inline]
1128 pub unsafe fn try_from_trusted_len_iter<
1129 E,
1130 T: ArrowNativeType,
1131 I: Iterator<Item = Result<T, E>>,
1132 >(
1133 iterator: I,
1134 ) -> Result<Self, E> {
1135 let item_size = std::mem::size_of::<T>();
1136 let (_, upper) = iterator.size_hint();
1137 let upper = upper.expect("try_from_trusted_len_iter requires an upper limit");
1138 let len = upper * item_size;
1139
1140 let mut buffer = MutableBuffer::new(len);
1141
1142 let mut dst = buffer.data.as_ptr();
1143 for item in iterator {
1144 let item = item?;
1145 let src = item.to_byte_slice().as_ptr();
1147 unsafe { std::ptr::copy_nonoverlapping(src, dst, item_size) };
1148 dst = unsafe { dst.add(item_size) };
1149 }
1150 unsafe fn finalize_buffer(dst: *mut u8, buffer: &mut MutableBuffer, len: usize) {
1153 unsafe {
1154 assert_eq!(
1155 dst.offset_from(buffer.data.as_ptr()) as usize,
1156 len,
1157 "Trusted iterator length was not accurately reported"
1158 );
1159 buffer.len = len;
1160 }
1161 }
1162 unsafe { finalize_buffer(dst, &mut buffer, len) };
1163 Ok(buffer)
1164 }
1165}
1166
1167impl Default for MutableBuffer {
1168 fn default() -> Self {
1169 Self::with_capacity(0)
1170 }
1171}
1172
1173impl std::ops::Deref for MutableBuffer {
1174 type Target = [u8];
1175
1176 fn deref(&self) -> &[u8] {
1177 unsafe { std::slice::from_raw_parts(self.as_ptr(), self.len) }
1178 }
1179}
1180
1181impl std::ops::DerefMut for MutableBuffer {
1182 fn deref_mut(&mut self) -> &mut [u8] {
1183 unsafe { std::slice::from_raw_parts_mut(self.as_mut_ptr(), self.len) }
1184 }
1185}
1186
1187impl AsRef<[u8]> for &MutableBuffer {
1188 fn as_ref(&self) -> &[u8] {
1189 self.as_slice()
1190 }
1191}
1192
1193impl Drop for MutableBuffer {
1194 fn drop(&mut self) {
1195 if self.layout.size() != 0 {
1196 unsafe { std::alloc::dealloc(self.data.as_ptr().cast(), self.layout) };
1198 }
1199 }
1200}
1201
1202impl PartialEq for MutableBuffer {
1203 fn eq(&self, other: &MutableBuffer) -> bool {
1204 if self.len != other.len {
1205 return false;
1206 }
1207 if self.layout != other.layout {
1208 return false;
1209 }
1210 self.as_slice() == other.as_slice()
1211 }
1212}
1213
1214unsafe impl Sync for MutableBuffer {}
1215unsafe impl Send for MutableBuffer {}
1216
1217struct SetLenOnDrop<'a> {
1218 len: &'a mut usize,
1219 local_len: usize,
1220}
1221
1222impl<'a> SetLenOnDrop<'a> {
1223 #[inline]
1224 fn new(len: &'a mut usize) -> Self {
1225 SetLenOnDrop {
1226 local_len: *len,
1227 len,
1228 }
1229 }
1230}
1231
1232impl Drop for SetLenOnDrop<'_> {
1233 #[inline]
1234 fn drop(&mut self) {
1235 *self.len = self.local_len;
1236 }
1237}
1238
1239impl std::iter::FromIterator<bool> for MutableBuffer {
1241 fn from_iter<I>(iter: I) -> Self
1242 where
1243 I: IntoIterator<Item = bool>,
1244 {
1245 let mut iterator = iter.into_iter();
1246 let mut result = {
1247 let byte_capacity: usize = iterator.size_hint().0.saturating_add(7) / 8;
1248 MutableBuffer::new(byte_capacity)
1249 };
1250
1251 loop {
1252 let mut exhausted = false;
1253 let mut byte_accum: u8 = 0;
1254 let mut mask: u8 = 1;
1255
1256 while mask != 0 {
1258 if let Some(value) = iterator.next() {
1259 byte_accum |= match value {
1260 true => mask,
1261 false => 0,
1262 };
1263 mask <<= 1;
1264 } else {
1265 exhausted = true;
1266 break;
1267 }
1268 }
1269
1270 if exhausted && mask == 1 {
1272 break;
1273 }
1274
1275 if result.len() == result.capacity() {
1277 let additional_byte_capacity = 1usize.saturating_add(
1279 iterator.size_hint().0.saturating_add(7) / 8, );
1281 result.reserve(additional_byte_capacity)
1282 }
1283
1284 unsafe { result.push_unchecked(byte_accum) };
1286 if exhausted {
1287 break;
1288 }
1289 }
1290 result
1291 }
1292}
1293
1294impl<T: ArrowNativeType> std::iter::FromIterator<T> for MutableBuffer {
1295 fn from_iter<I: IntoIterator<Item = T>>(iter: I) -> Self {
1296 let mut buffer = Self::default();
1297 buffer.extend_from_iter(iter.into_iter());
1298 buffer
1299 }
1300}
1301
1302#[cfg(test)]
1303mod tests {
1304 use super::*;
1305
1306 #[test]
1307 fn test_mutable_new() {
1308 let buf = MutableBuffer::new(63);
1309 assert_eq!(64, buf.capacity());
1310 assert_eq!(0, buf.len());
1311 assert!(buf.is_empty());
1312 }
1313
1314 #[test]
1315 fn test_mutable_default() {
1316 let buf = MutableBuffer::default();
1317 assert_eq!(0, buf.capacity());
1318 assert_eq!(0, buf.len());
1319 assert!(buf.is_empty());
1320
1321 let mut buf = MutableBuffer::default();
1322 buf.extend_from_slice(b"hello");
1323 assert_eq!(5, buf.len());
1324 assert_eq!(b"hello", buf.as_slice());
1325 }
1326
1327 #[test]
1328 fn test_mutable_extend_from_slice() {
1329 let mut buf = MutableBuffer::new(100);
1330 buf.extend_from_slice(b"hello");
1331 assert_eq!(5, buf.len());
1332 assert_eq!(b"hello", buf.as_slice());
1333
1334 buf.extend_from_slice(b" world");
1335 assert_eq!(11, buf.len());
1336 assert_eq!(b"hello world", buf.as_slice());
1337
1338 buf.clear();
1339 assert_eq!(0, buf.len());
1340 buf.extend_from_slice(b"hello arrow");
1341 assert_eq!(11, buf.len());
1342 assert_eq!(b"hello arrow", buf.as_slice());
1343 }
1344
1345 #[test]
1346 fn mutable_extend_from_iter() {
1347 let mut buf = MutableBuffer::new(0);
1348 buf.extend(vec![1u32, 2]);
1349 assert_eq!(8, buf.len());
1350 assert_eq!(&[1u8, 0, 0, 0, 2, 0, 0, 0], buf.as_slice());
1351
1352 buf.extend(vec![3u32, 4]);
1353 assert_eq!(16, buf.len());
1354 assert_eq!(
1355 &[1u8, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0, 4, 0, 0, 0],
1356 buf.as_slice()
1357 );
1358 }
1359
1360 #[test]
1361 fn mutable_extend_from_iter_unaligned_u64() {
1362 let mut buf = MutableBuffer::new(16);
1363 buf.push(1_u8);
1364 buf.extend([1_u64]);
1365 assert_eq!(9, buf.len());
1366 assert_eq!(&[1u8, 1u8, 0, 0, 0, 0, 0, 0, 0], buf.as_slice());
1367 }
1368
1369 #[test]
1370 fn mutable_extend_from_slice_unaligned_u64() {
1371 let mut buf = MutableBuffer::new(16);
1372 buf.extend_from_slice(&[1_u8]);
1373 buf.extend_from_slice(&[1_u64]);
1374 assert_eq!(9, buf.len());
1375 assert_eq!(&[1u8, 1u8, 0, 0, 0, 0, 0, 0, 0], buf.as_slice());
1376 }
1377
1378 #[test]
1379 fn mutable_push_unaligned_u64() {
1380 let mut buf = MutableBuffer::new(16);
1381 buf.push(1_u8);
1382 buf.push(1_u64);
1383 assert_eq!(9, buf.len());
1384 assert_eq!(&[1u8, 1u8, 0, 0, 0, 0, 0, 0, 0], buf.as_slice());
1385 }
1386
1387 #[test]
1388 fn mutable_push_unchecked_unaligned_u64() {
1389 let mut buf = MutableBuffer::new(16);
1390 unsafe {
1391 buf.push_unchecked(1_u8);
1392 buf.push_unchecked(1_u64);
1393 }
1394 assert_eq!(9, buf.len());
1395 assert_eq!(&[1u8, 1u8, 0, 0, 0, 0, 0, 0, 0], buf.as_slice());
1396 }
1397
1398 #[test]
1399 fn test_from_trusted_len_iter() {
1400 let iter = vec![1u32, 2].into_iter();
1401 let buf = unsafe { MutableBuffer::from_trusted_len_iter(iter) };
1402 assert_eq!(8, buf.len());
1403 assert_eq!(&[1u8, 0, 0, 0, 2, 0, 0, 0], buf.as_slice());
1404 }
1405
1406 #[test]
1407 fn test_mutable_reserve() {
1408 let mut buf = MutableBuffer::new(1);
1409 assert_eq!(64, buf.capacity());
1410
1411 buf.reserve(10);
1413 assert_eq!(64, buf.capacity());
1414
1415 buf.reserve(80);
1416 assert_eq!(128, buf.capacity());
1417
1418 buf.reserve(129);
1419 assert_eq!(256, buf.capacity());
1420 }
1421
1422 #[test]
1423 fn test_mutable_resize() {
1424 let mut buf = MutableBuffer::new(1);
1425 assert_eq!(64, buf.capacity());
1426 assert_eq!(0, buf.len());
1427
1428 buf.resize(20, 0);
1429 assert_eq!(64, buf.capacity());
1430 assert_eq!(20, buf.len());
1431
1432 buf.resize(10, 0);
1433 assert_eq!(64, buf.capacity());
1434 assert_eq!(10, buf.len());
1435
1436 buf.resize(100, 0);
1437 assert_eq!(128, buf.capacity());
1438 assert_eq!(100, buf.len());
1439
1440 buf.resize(30, 0);
1441 assert_eq!(128, buf.capacity());
1442 assert_eq!(30, buf.len());
1443
1444 buf.resize(0, 0);
1445 assert_eq!(128, buf.capacity());
1446 assert_eq!(0, buf.len());
1447 }
1448
1449 #[test]
1450 fn test_mutable_into() {
1451 let mut buf = MutableBuffer::new(1);
1452 buf.extend_from_slice(b"aaaa bbbb cccc dddd");
1453 assert_eq!(19, buf.len());
1454 assert_eq!(64, buf.capacity());
1455 assert_eq!(b"aaaa bbbb cccc dddd", buf.as_slice());
1456
1457 let immutable_buf: Buffer = buf.into();
1458 assert_eq!(19, immutable_buf.len());
1459 assert_eq!(64, immutable_buf.capacity());
1460 assert_eq!(b"aaaa bbbb cccc dddd", immutable_buf.as_slice());
1461 }
1462
1463 #[test]
1464 fn test_mutable_equal() {
1465 let mut buf = MutableBuffer::new(1);
1466 let mut buf2 = MutableBuffer::new(1);
1467
1468 buf.extend_from_slice(&[0xaa]);
1469 buf2.extend_from_slice(&[0xaa, 0xbb]);
1470 assert_ne!(buf, buf2);
1471
1472 buf.extend_from_slice(&[0xbb]);
1473 assert_eq!(buf, buf2);
1474
1475 buf2.reserve(65);
1476 assert_ne!(buf, buf2);
1477 }
1478
1479 #[test]
1480 fn test_mutable_shrink_to_fit() {
1481 let mut buffer = MutableBuffer::new(128);
1482 assert_eq!(buffer.capacity(), 128);
1483 buffer.push(1);
1484 buffer.push(2);
1485
1486 buffer.shrink_to_fit();
1487 assert!(buffer.capacity() >= 64 && buffer.capacity() < 128);
1488 }
1489
1490 #[test]
1491 fn test_mutable_set_null_bits() {
1492 let mut buffer = MutableBuffer::new(8).with_bitset(8, true);
1493
1494 for i in 0..=buffer.capacity() {
1495 buffer.set_null_bits(i, 0);
1496 assert_eq!(buffer[..8], [255; 8][..]);
1497 }
1498
1499 buffer.set_null_bits(1, 4);
1500 assert_eq!(buffer[..8], [255, 0, 0, 0, 0, 255, 255, 255][..]);
1501 }
1502
1503 #[test]
1504 #[should_panic = "out of bounds for buffer of length"]
1505 fn test_mutable_set_null_bits_oob() {
1506 let mut buffer = MutableBuffer::new(64);
1507 buffer.set_null_bits(1, buffer.capacity());
1508 }
1509
1510 #[test]
1511 #[should_panic = "out of bounds for buffer of length"]
1512 fn test_mutable_set_null_bits_oob_by_overflow() {
1513 let mut buffer = MutableBuffer::new(0);
1514 buffer.set_null_bits(1, usize::MAX);
1515 }
1516
1517 #[test]
1518 fn from_iter() {
1519 let buffer = [1u16, 2, 3, 4].into_iter().collect::<MutableBuffer>();
1520 assert_eq!(buffer.len(), 4 * mem::size_of::<u16>());
1521 assert_eq!(buffer.as_slice(), &[1, 0, 2, 0, 3, 0, 4, 0]);
1522 }
1523
1524 #[test]
1525 #[should_panic(expected = "invalid allocation layout for requested capacity")]
1526 fn test_with_capacity_panics_above_max_capacity() {
1527 let max_capacity = isize::MAX as usize - (isize::MAX as usize % ALIGNMENT);
1528 let _ = MutableBuffer::with_capacity(max_capacity + 1);
1529 }
1530
1531 #[cfg(feature = "pool")]
1532 mod pool_tests {
1533 use super::*;
1534 use crate::pool::{MemoryPool, TrackingMemoryPool};
1535
1536 #[test]
1537 fn test_reallocate_with_pool() {
1538 let pool = TrackingMemoryPool::default();
1539 let mut buffer = MutableBuffer::with_capacity(100);
1540 buffer.claim(&pool);
1541
1542 assert_eq!(buffer.capacity(), 128);
1544 assert_eq!(pool.used(), 128);
1545
1546 buffer.try_reallocate(200).unwrap();
1548
1549 assert_eq!(buffer.capacity(), 200);
1551 assert_eq!(pool.used(), 200);
1552
1553 buffer.try_reallocate(50).unwrap();
1555
1556 assert_eq!(buffer.capacity(), 50);
1558 assert_eq!(pool.used(), 50);
1559 }
1560
1561 #[test]
1562 fn test_truncate_with_pool() {
1563 let pool = TrackingMemoryPool::default();
1564 let mut buffer = MutableBuffer::with_capacity(100);
1565
1566 buffer.resize(80, 1);
1568 assert_eq!(buffer.len(), 80);
1569
1570 buffer.claim(&pool);
1571 assert_eq!(pool.used(), 128);
1572
1573 buffer.truncate(40);
1575 assert_eq!(buffer.len(), 40);
1576 assert_eq!(pool.used(), 40);
1577
1578 buffer.clear();
1580 assert_eq!(buffer.len(), 0);
1581 assert_eq!(pool.used(), 0);
1582 }
1583
1584 #[test]
1585 fn test_resize_with_pool() {
1586 let pool = TrackingMemoryPool::default();
1587 let mut buffer = MutableBuffer::with_capacity(100);
1588 buffer.claim(&pool);
1589
1590 assert_eq!(buffer.len(), 0);
1592 assert_eq!(pool.used(), 128);
1593
1594 buffer.resize(50, 1);
1596 assert_eq!(buffer.len(), 50);
1597 assert_eq!(pool.used(), 50);
1598
1599 buffer.resize(150, 1);
1601 assert_eq!(buffer.len(), 150);
1602 assert_eq!(buffer.capacity(), 256);
1603 assert_eq!(pool.used(), 150);
1604
1605 buffer.resize(30, 1);
1607 assert_eq!(buffer.len(), 30);
1608 assert_eq!(pool.used(), 30);
1609 }
1610
1611 #[test]
1612 fn test_buffer_lifecycle_with_pool() {
1613 let pool = TrackingMemoryPool::default();
1614
1615 let mut mutable = MutableBuffer::with_capacity(100);
1617 mutable.resize(80, 1);
1618 mutable.claim(&pool);
1619
1620 assert_eq!(pool.used(), 128);
1622
1623 let buffer = mutable.into_buffer();
1625
1626 assert_eq!(pool.used(), 128);
1628
1629 drop(buffer);
1631 assert_eq!(pool.used(), 0);
1632 }
1633 }
1634
1635 fn create_expected_repeated_slice<T: ArrowNativeType>(
1636 slice_to_repeat: &[T],
1637 repeat_count: usize,
1638 ) -> Buffer {
1639 let mut expected = MutableBuffer::new(size_of_val(slice_to_repeat) * repeat_count);
1640 for _ in 0..repeat_count {
1641 expected.extend_from_slice(slice_to_repeat);
1643 }
1644 expected.into()
1645 }
1646
1647 fn test_repeat_count<T: ArrowNativeType + PartialEq + std::fmt::Debug>(
1649 repeat_count: usize,
1650 test_data: &[T],
1651 ) {
1652 let mut buffer = MutableBuffer::new(0);
1653 buffer.repeat_slice_n_times(test_data, repeat_count);
1654
1655 let expected = create_expected_repeated_slice(test_data, repeat_count);
1656 let result: Buffer = buffer.into();
1657
1658 assert_eq!(
1659 result,
1660 expected,
1661 "Failed for repeat_count={}, slice_len={}",
1662 repeat_count,
1663 test_data.len()
1664 );
1665 }
1666
1667 #[test]
1668 fn test_repeat_slice_count_edge_cases() {
1669 test_repeat_count(100, &[] as &[i32]);
1671
1672 test_repeat_count(0, &[1i32, 2, 3]);
1674 }
1675
1676 #[test]
1677 #[should_panic(expected = "buffer length overflow")]
1678 fn test_repeat_slice_count_multiply_overflow() {
1679 let mut buffer = MutableBuffer::new(0);
1680 buffer.repeat_slice_n_times(&[0_u64], usize::MAX / mem::size_of::<u64>() + 1);
1681 }
1682
1683 #[test]
1684 #[should_panic(expected = "buffer length overflow")]
1685 fn test_repeat_slice_count_len_overflow() {
1686 let mut buffer = MutableBuffer::new(0);
1687 buffer.push(0_u8);
1688 buffer.repeat_slice_n_times(&[0_u8], usize::MAX);
1689 }
1690
1691 #[test]
1692 fn test_small_repeats_counts() {
1693 let data = &[1u8, 2, 3, 4, 5];
1695
1696 for _ in 1..=10 {
1697 test_repeat_count(2, data);
1698 }
1699 }
1700
1701 #[test]
1702 fn test_different_size_of_i32_repeat_slice() {
1703 let data: &[i32] = &[1, 2, 3];
1704 let data_with_single_item: &[i32] = &[42];
1705
1706 for data in &[data, data_with_single_item] {
1707 for item in 1..=9 {
1708 let base_repeat_count = 2_usize.pow(item);
1709 test_repeat_count(base_repeat_count - 1, data);
1710 test_repeat_count(base_repeat_count, data);
1711 test_repeat_count(base_repeat_count + 1, data);
1712 }
1713 }
1714 }
1715
1716 #[test]
1717 fn test_different_size_of_u8_repeat_slice() {
1718 let data: &[u8] = &[1, 2, 3];
1719 let data_with_single_item: &[u8] = &[10];
1720
1721 for data in &[data, data_with_single_item] {
1722 for item in 1..=9 {
1723 let base_repeat_count = 2_usize.pow(item);
1724 test_repeat_count(base_repeat_count - 1, data);
1725 test_repeat_count(base_repeat_count, data);
1726 test_repeat_count(base_repeat_count + 1, data);
1727 }
1728 }
1729 }
1730
1731 #[test]
1732 fn test_different_size_of_u16_repeat_slice() {
1733 let data: &[u16] = &[1, 2, 3];
1734 let data_with_single_item: &[u16] = &[10];
1735
1736 for data in &[data, data_with_single_item] {
1737 for item in 1..=9 {
1738 let base_repeat_count = 2_usize.pow(item);
1739 test_repeat_count(base_repeat_count - 1, data);
1740 test_repeat_count(base_repeat_count, data);
1741 test_repeat_count(base_repeat_count + 1, data);
1742 }
1743 }
1744 }
1745
1746 #[test]
1747 fn test_various_slice_lengths() {
1748 let repeat_count = 37; test_repeat_count(repeat_count, &[42i32]);
1753
1754 test_repeat_count(repeat_count, &[1i32, 2]);
1756 test_repeat_count(repeat_count, &[1i32, 2, 3]);
1757 test_repeat_count(repeat_count, &[1i32, 2, 3, 4]);
1758 test_repeat_count(repeat_count, &[1i32, 2, 3, 4, 5]);
1759
1760 let data_10: Vec<i32> = (0..10).collect();
1762 test_repeat_count(repeat_count, &data_10);
1763
1764 let data_100: Vec<i32> = (0..100).collect();
1765 test_repeat_count(repeat_count, &data_100);
1766
1767 let data_1000: Vec<i32> = (0..1000).collect();
1768 test_repeat_count(repeat_count, &data_1000);
1769 }
1770
1771 #[test]
1772 #[should_panic(expected = "invalid allocation layout for requested capacity")]
1773 fn test_mutable_new_capacity_overflow() {
1774 let _ = MutableBuffer::new(usize::MAX - 10);
1776 }
1777
1778 #[test]
1779 #[should_panic(expected = "buffer length overflow")]
1780 fn test_mutable_reserve_overflow() {
1781 let mut buf = MutableBuffer::new(1);
1783 buf.push(1u8);
1784 buf.reserve(usize::MAX);
1785 }
1786}