ARROW-10111: [Rust] Added new crate with code that consumes C Data interface

rust/arrow/src/array/ffi.rs

@@ -0,0 +1,120 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Contains functionality to load an ArrayData from the C Data Interface
+
+use std::convert::TryFrom;
+
+use crate::{
+    error::{ArrowError, Result},
+    ffi,
+};
+
+use super::ArrayData;
+
+impl TryFrom<ffi::ArrowArray> for ArrayData {
+    type Error = ArrowError;
+
+    fn try_from(value: ffi::ArrowArray) -> Result<Self> {
+        let data_type = value.data_type()?;
+        let len = value.len();
+        let offset = value.offset();
+        let null_count = value.null_count();
+        let buffers = value.buffers()?;
+        let null_bit_buffer = value.null_bit_buffer();
+
+        // todo: no child data yet...
+        Ok(ArrayData::new(
+            data_type,
+            len,
+            Some(null_count),
+            null_bit_buffer,
+            offset,
+            buffers,
+            vec![],
+        ))
+    }
+}
+
+impl TryFrom<ArrayData> for ffi::ArrowArray {
+    type Error = ArrowError;
+
+    fn try_from(value: ArrayData) -> Result<Self> {
+        let len = value.len;
+        let offset = value.offset as usize;
+        let null_count = value.null_count;
+        let buffers = value.buffers().to_vec();
+        let null_buffer = value.null_buffer().cloned();
+
+        // todo: no child data yet...
+        ffi::ArrowArray::try_new(
+            value.data_type(),
+            len,
+            null_count,
+            null_buffer,
+            offset,
+            buffers,
+            vec![],
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::error::Result;
+    use crate::{
+        array::{Array, ArrayData, Int64Array, UInt32Array, UInt64Array},
+        ffi::ArrowArray,
+    };
+    use std::convert::TryFrom;
+
+    fn test_round_trip(expected: &ArrayData) -> Result<()> {
+        // create a `ArrowArray` from the data.
+        // here we take ownership (increase ref count)
+        let d1 = ArrowArray::try_from(expected.clone())?;
+
+        // here we export the array as 2 pointers. We have no control over ownership if it was not for
+        // the release mechanism.
+        let (array, schema) = ArrowArray::to_raw(d1);
+
+        // simulate an external consumer by being the consumer
+        let d1 = unsafe { ArrowArray::try_from_raw(array, schema) }?;
+
+        let result = &ArrayData::try_from(d1)?;
+
+        assert_eq!(result, expected);
+        Ok(())
+    }
+
+    #[test]
+    fn test_u32() -> Result<()> {
+        let data = UInt32Array::from(vec![2]).data();
+        test_round_trip(data.as_ref())
+    }
+
+    #[test]
+    fn test_u64() -> Result<()> {
+        let data = UInt64Array::from(vec![2]).data();
+        test_round_trip(data.as_ref())
+    }
+
+    #[test]
+    fn test_i64() -> Result<()> {
+        let data = Int64Array::from(vec![2]).data();
+        test_round_trip(data.as_ref())
+    }
+}

rust/arrow/src/array/mod.rs

@@ -254,3 +254,5 @@ pub use self::cast::{
     as_boolean_array, as_dictionary_array, as_null_array, as_primitive_array,
     as_string_array,
 };
+
+pub mod ffi;

rust/arrow/src/buffer.rs

@@ -21,13 +21,17 @@
 #[cfg(feature = "simd")]
 use packed_simd::u8x64;
 
+use crate::{
+    bytes::{Bytes, Deallocation},
+    ffi,
+};
+
 use std::cmp;
 use std::convert::AsRef;
-use std::fmt::{Debug, Formatter};
+use std::fmt::Debug;
 use std::io::{Error as IoError, ErrorKind, Result as IoResult, Write};
 use std::mem;
 use std::ops::{BitAnd, BitOr, Not};
-use std::ptr::NonNull;
 use std::slice::{from_raw_parts, from_raw_parts_mut};
 use std::sync::Arc;
 
@@ -38,92 +42,16 @@ use crate::util::bit_util;
 #[cfg(feature = "simd")]
 use std::borrow::BorrowMut;
 
-/// Buffer is a contiguous memory region of fixed size and is aligned at a 64-byte
-/// boundary. Buffer is immutable.
-#[derive(PartialEq, Debug)]
+/// [`Buffer`] is an immutable contiguous memory region.
+#[derive(PartialEq, Clone, Debug)]
 pub struct Buffer {
     /// Reference-counted pointer to the internal byte buffer.
-    data: Arc<BufferData>,
+    data: Arc<Bytes>,
 
     /// The offset into the buffer.
     offset: usize,
 }
 
-struct BufferData {
-    /// The raw pointer into the buffer bytes
-    ptr: *const u8,
-
-    /// The length (num of bytes) of the buffer. The region `[0, len)` of the buffer
-    /// is occupied with meaningful data, while the rest `[len, capacity)` is the
-    /// unoccupied region.
-    len: usize,
-
-    /// Whether this piece of memory is owned by this object
-    owned: bool,
-
-    /// The capacity (num of bytes) of the buffer
-    /// Invariant: len <= capacity
-    capacity: usize,
-}
-
-impl PartialEq for BufferData {
-    fn eq(&self, other: &BufferData) -> bool {
-        if self.capacity != other.capacity {
-            return false;
-        }
-
-        self.data() == other.data()
-    }
-}
-
-/// Release the underlying memory when the current buffer goes out of scope
-impl Drop for BufferData {
-    fn drop(&mut self) {
-        if self.is_allocated() && self.owned {
-            unsafe { memory::free_aligned(self.ptr as *mut u8, self.capacity) };
-        }
-    }
-}
-
-impl Debug for BufferData {
-    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {
-        write!(
-            f,
-            "BufferData {{ ptr: {:?}, len: {}, capacity: {}, data: ",
-            self.ptr, self.len, self.capacity
-        )?;
-
-        f.debug_list().entries(self.data().iter()).finish()?;
-
-        write!(f, " }}")
-    }
-}
-
-impl BufferData {
-    fn data(&self) -> &[u8] {
-        if !self.is_allocated() {
-            &[]
-        } else {
-            unsafe { std::slice::from_raw_parts(self.ptr, self.len) }
-        }
-    }
-
-    fn is_zst(&self) -> bool {
-        self.ptr == BUFFER_INIT.as_ptr() as _
-    }
-
-    fn is_allocated(&self) -> bool {
-        !(self.is_zst() || self.ptr.is_null())
-    }
-}
-
-///
-/// SAFETY: (vcq):
-/// As you can see this is global and lives as long as the program lives.
-/// This is used for lazy allocation in the further steps of buffer allocations.
-/// Pointer below is well-aligned, only used for ZSTs and discarded afterwards.
-const BUFFER_INIT: NonNull<u8> = NonNull::dangling();
-
 impl Buffer {
     /// Creates a buffer from an existing memory region (must already be byte-aligned), this
     /// `Buffer` will free this piece of memory when dropped.
@@ -139,7 +67,7 @@ impl Buffer {
     /// This function is unsafe as there is no guarantee that the given pointer is valid for `len`
     /// bytes. If the `ptr` and `capacity` come from a `Buffer`, then this is guaranteed.
     pub unsafe fn from_raw_parts(ptr: *const u8, len: usize, capacity: usize) -> Self {
-        Buffer::build_with_arguments(ptr, len, capacity, true)
+        Buffer::build_with_arguments(ptr, len, Deallocation::Native(capacity))
     }
 
     /// Creates a buffer from an existing memory region (must already be byte-aligned), this
@@ -149,70 +77,52 @@ impl Buffer {
     ///
     /// * `ptr` - Pointer to raw parts
     /// * `len` - Length of raw parts in **bytes**
-    /// * `capacity` - Total allocated memory for the pointer `ptr`, in **bytes**
+    /// * `data` - An [ffi::FFI_ArrowArray] with the data
     ///
     /// # Safety
     ///
     /// This function is unsafe as there is no guarantee that the given pointer is valid for `len`
-    /// bytes. If the `ptr` and `capacity` come from a `Buffer`, then this is guaranteed.
-    pub unsafe fn from_unowned(ptr: *const u8, len: usize, capacity: usize) -> Self {
-        Buffer::build_with_arguments(ptr, len, capacity, false)
+    /// bytes and that the foreign deallocator frees the region.
+    pub unsafe fn from_unowned(
+        ptr: *const u8,
+        len: usize,
+        data: ffi::FFI_ArrowArray,
+    ) -> Self {
+        Buffer::build_with_arguments(ptr, len, Deallocation::Foreign(data))
     }
 
-    /// Creates a buffer from an existing memory region (must already be byte-aligned).
-    ///
-    /// # Arguments
-    ///
-    /// * `ptr` - Pointer to raw parts
-    /// * `len` - Length of raw parts in bytes
-    /// * `capacity` - Total allocated memory for the pointer `ptr`, in **bytes**
-    /// * `owned` - Whether the raw parts is owned by this `Buffer`. If true, this `Buffer` will
-    /// free this memory when dropped, otherwise it will skip freeing the raw parts.
-    ///
-    /// # Safety
-    ///
-    /// This function is unsafe as there is no guarantee that the given pointer is valid for `len`
-    /// bytes. If the `ptr` and `capacity` come from a `Buffer`, then this is guaranteed.
+    /// Auxiliary method to create a new Buffer
     unsafe fn build_with_arguments(
         ptr: *const u8,
         len: usize,
-        capacity: usize,
-        owned: bool,
+        deallocation: Deallocation,
     ) -> Self {
-        assert!(
-            memory::is_aligned(ptr, memory::ALIGNMENT),
-            "memory not aligned"
-        );
-        let buf_data = BufferData {
-            ptr,
-            len,
-            capacity,
-            owned,
-        };
+        let bytes = Bytes::new(ptr, len, deallocation);
         Buffer {
-            data: Arc::new(buf_data),
+            data: Arc::new(bytes),
             offset: 0,
         }
     }
 
     /// Returns the number of bytes in the buffer
     pub fn len(&self) -> usize {
-        self.data.len - self.offset
+        self.data.len() - self.offset
     }
 
-    /// Returns the capacity of this buffer
+    /// Returns the capacity of this buffer.
+    /// For exernally owned buffers, this returns zero
     pub fn capacity(&self) -> usize {
-        self.data.capacity
+        self.data.capacity()
     }
 
     /// Returns whether the buffer is empty.
     pub fn is_empty(&self) -> bool {
-        self.data.len - self.offset == 0
+        self.data.len() - self.offset == 0
     }
 
     /// Returns the byte slice stored in this buffer
     pub fn data(&self) -> &[u8] {
-        &self.data.data()[self.offset..]
+        &self.data.as_slice()[self.offset..]
     }
 
     /// Returns a slice of this buffer, starting from `offset`.
@@ -232,7 +142,7 @@ impl Buffer {
     /// Note that this should be used cautiously, and the returned pointer should not be
     /// stored anywhere, to avoid dangling pointers.
     pub fn raw_data(&self) -> *const u8 {
-        unsafe { self.data.ptr.add(self.offset) }
+        unsafe { self.data.raw_data().add(self.offset) }
     }
 
     /// View buffer as typed slice.
@@ -253,20 +163,6 @@ impl Buffer {
             self.len() / mem::size_of::<T>(),
         )
     }
-
-    /// Returns an empty buffer.
-    pub fn empty() -> Self {
-        unsafe { Self::from_raw_parts(BUFFER_INIT.as_ptr() as _, 0, 0) }
-    }
-}
-
-impl Clone for Buffer {
-    fn clone(&self) -> Buffer {
-        Buffer {
-            data: self.data.clone(),
-            offset: self.offset,
-        }
-    }
 }
 
 /// Creating a `Buffer` instance by copying the memory from a `AsRef<[u8]>` into a newly
@@ -667,11 +563,8 @@ impl MutableBuffer {
 
     /// Freezes this buffer and return an immutable version of it.
     pub fn freeze(self) -> Buffer {
-        let buffer_data = BufferData {
-            ptr: self.data,
-            len: self.len,
-            capacity: self.capacity,
-            owned: true,
+        let buffer_data = unsafe {
+            Bytes::new(self.data, self.len, Deallocation::Native(self.capacity))
         };
         std::mem::forget(self);
         Buffer {