feat: better ways to use deletion vectors

ffi/src/lib.rs

@@ -128,20 +128,30 @@ pub type AllocateStringFn = extern "C" fn(kernel_str: KernelStringSlice) -> Null
 // Put KernelBoolSlice in a sub-module, with non-public members, so rust code cannot instantiate it
 // directly. It can only be created by converting `From<Vec<bool>>`.
 mod private {
+    use std::ptr::NonNull;
+
     /// Represents an owned slice of boolean values allocated by the kernel. Any time the engine
     /// receives a `KernelBoolSlice` as a return value from a kernel method, engine is responsible
     /// to free that slice, by calling [super::free_bool_slice] exactly once.
     #[repr(C)]
     pub struct KernelBoolSlice {
-        ptr: *mut bool,
+        ptr: NonNull<bool>,
+        len: usize,
+    }
+
+    /// An owned slice of u64 row indexes allocated by the kernel. The engine is responsible for
+    /// freeing this slice by calling [super::free_row_indexes] once.
+    #[repr(C)]
+    pub struct KernelRowIndexArray {
+        ptr: NonNull<u64>,
         len: usize,
     }
 
     impl KernelBoolSlice {
         /// Creates an empty slice.
         pub fn empty() -> KernelBoolSlice {
             KernelBoolSlice {
-                ptr: std::ptr::null_mut(),
+                ptr: NonNull::dangling(),
                 len: 0,
             }
         }
@@ -153,10 +163,10 @@ mod private {
         /// The slice must have been originally created `From<Vec<bool>>`, and must not have been
         /// already been consumed by a previous call to this method.
         pub unsafe fn as_ref(&self) -> &[bool] {
-            if self.ptr.is_null() {
+            if self.len == 0 {
                 Default::default()
             } else {
-                unsafe { std::slice::from_raw_parts(self.ptr, self.len) }
+                unsafe { std::slice::from_raw_parts(self.ptr.as_ptr(), self.len) }
             }
         }
 
@@ -167,10 +177,10 @@ mod private {
         /// The slice must have been originally created `From<Vec<bool>>`, and must not have been
         /// already been consumed by a previous call to this method.
         pub unsafe fn into_vec(self) -> Vec<bool> {
-            if self.ptr.is_null() {
+            if self.len == 0 {
                 Default::default()
             } else {
-                Vec::from_raw_parts(self.ptr, self.len, self.len)
+                Vec::from_raw_parts(self.ptr.as_ptr(), self.len, self.len)
             }
         }
     }
@@ -179,8 +189,12 @@ mod private {
         fn from(val: Vec<bool>) -> Self {
             let len = val.len();
             let boxed = val.into_boxed_slice();
-            let ptr = Box::into_raw(boxed).cast();
-            KernelBoolSlice { ptr, len }
+            let ptr = Box::leak(boxed).as_mut_ptr();
+            if let Some(ptr) = NonNull::new(ptr) {
+                KernelBoolSlice { ptr, len }
+            } else {
+                KernelBoolSlice::empty()
+            }
         }
     }
 
@@ -190,13 +204,57 @@ mod private {
     /// memory, but must only free it by calling [super::free_bool_slice]. Since the global
     /// allocator is threadsafe, it doesn't matter which engine thread invokes that method.
     unsafe impl Send for KernelBoolSlice {}
+    unsafe impl Send for KernelRowIndexArray {}
 
     /// # Safety
     ///
     /// If engine chooses to leverage concurrency, engine is responsible to prevent data races.
     unsafe impl Sync for KernelBoolSlice {}
+    /// # Safety
+    ///
+    /// If engine chooses to leverage concurrency, engine is responsible to prevent data races.
+    /// Same contract as KernelBoolSlice above
+    unsafe impl Sync for KernelRowIndexArray {}
+
+    impl KernelRowIndexArray {
+        /// Converts this slice back into a `Vec<u64>`.
+        ///
+        /// # Safety
+        ///
+        /// The slice must have been originally created `From<Vec<u64>>`, and must not have
+        /// already been consumed by a previous call to this method.
+        pub unsafe fn into_vec(self) -> Vec<u64> {
+            if self.len == 0 {
+                Default::default()
+            } else {
+                Vec::from_raw_parts(self.ptr.as_ptr(), self.len, self.len)
+            }
+        }
+
+        /// Creates an empty slice.
+        pub fn empty() -> KernelRowIndexArray {
+            Self {
+                ptr: NonNull::dangling(),
+                len: 0,
+            }
+        }
+    }
+
+    impl From<Vec<u64>> for KernelRowIndexArray {
+        fn from(vec: Vec<u64>) -> Self {
+            let len = vec.len();
+            let boxed = vec.into_boxed_slice();
+            let ptr = Box::leak(boxed).as_mut_ptr();
+            if let Some(ptr) = NonNull::new(ptr) {
+                KernelRowIndexArray { ptr, len }
+            } else {
+                KernelRowIndexArray::empty()
+            }
+        }
+    }
 }
 pub use private::KernelBoolSlice;
+pub use private::KernelRowIndexArray;
 
 /// # Safety
 ///
@@ -207,6 +265,14 @@ pub unsafe extern "C" fn free_bool_slice(slice: KernelBoolSlice) {
     debug!("Dropping bool slice. It is {vec:#?}");
 }
 
+/// # Safety
+///
+/// Caller is responsible for passing a valid handle.
+#[no_mangle]
+pub unsafe extern "C" fn free_row_indexes(slice: KernelRowIndexArray) {
+    let _ = slice.into_vec();
+}
+
 // TODO: Do we want this handle at all? Perhaps we should just _always_ pass raw *mut c_void pointers
 // that are the engine data? Even if we want the type, should it be a shared handle instead?
 /// an opaque struct that encapsulates data read by an engine. this handle can be passed back into

ffi/src/scan.rs

@@ -16,8 +16,8 @@ use url::Url;
 use crate::{
     unwrap_kernel_expression, AllocateStringFn, EnginePredicate, ExclusiveEngineData, ExternEngine,
     ExternResult, IntoExternResult, KernelBoolSlice, KernelExpressionVisitorState,
-    KernelStringSlice, NullableCvoid, SharedExternEngine, SharedSnapshot, StringIter,
-    StringSliceIterator, TryFromStringSlice,
+    KernelRowIndexArray, KernelStringSlice, NullableCvoid, SharedExternEngine, SharedSnapshot,
+    StringIter, StringSliceIterator, TryFromStringSlice,
 };
 
 use super::handle::Handle;
@@ -391,6 +391,33 @@ fn selection_vector_from_dv_impl(
     }
 }
 
+/// Get a vector of row indexes out of a [`DvInfo`] struct
+///
+/// # Safety
+/// Engine is responsible for providing valid pointers for each argument
+#[no_mangle]
+pub unsafe extern "C" fn row_indexes_from_dv(
+    dv_info: &DvInfo,
+    engine: Handle<SharedExternEngine>,
+    state: Handle<SharedGlobalScanState>,
+) -> ExternResult<KernelRowIndexArray> {
+    let state = unsafe { state.as_ref() };
+    let engine = unsafe { engine.as_ref() };
+    row_indexes_from_dv_impl(dv_info, engine, state).into_extern_result(&engine)
+}
+
+fn row_indexes_from_dv_impl(
+    dv_info: &DvInfo,
+    extern_engine: &dyn ExternEngine,
+    state: &GlobalScanState,
+) -> DeltaResult<KernelRowIndexArray> {
+    let root_url = Url::parse(&state.table_root)?;
+    match dv_info.get_row_indexes(extern_engine.engine().as_ref(), &root_url)? {
+        Some(v) => Ok(v.into()),
+        None => Ok(KernelRowIndexArray::empty()),
+    }
+}
+
 // Wrapper function that gets called by the kernel, transforms the arguments to make the ffi-able,
 // and then calls the ffi specified callback
 fn rust_callback(

kernel/src/actions/deletion_vector.rs

@@ -4,10 +4,11 @@ use std::io::{Cursor, Read};
 use std::sync::Arc;
 
 use bytes::Bytes;
-use delta_kernel_derive::Schema;
 use roaring::RoaringTreemap;
 use url::Url;
 
+use delta_kernel_derive::Schema;
+
 use crate::utils::require;
 use crate::{DeltaResult, Error, FileSystemClient};
 
@@ -165,6 +166,14 @@ impl DeletionVectorDescriptor {
             }
         }
     }
+
+    pub fn row_indexes(
+        &self,
+        fs_client: Arc<dyn FileSystemClient>,
+        parent: &Url,
+    ) -> DeltaResult<Vec<u64>> {
+        Ok(self.read(fs_client, parent)?.into_iter().collect())
+    }
 }
 
 enum Endian {
@@ -249,13 +258,14 @@ pub fn split_vector(
 
 #[cfg(test)]
 mod tests {
-    use roaring::RoaringTreemap;
     use std::path::PathBuf;
 
-    use super::*;
+    use roaring::RoaringTreemap;
+
     use crate::{engine::sync::SyncEngine, Engine};
 
     use super::DeletionVectorDescriptor;
+    use super::*;
 
     fn dv_relative() -> DeletionVectorDescriptor {
         DeletionVectorDescriptor {
@@ -382,4 +392,16 @@ mod tests {
         expected[4294967300] = false;
         assert_eq!(bools, expected);
     }
+
+    #[test]
+    fn test_dv_wrapper() {
+        let example = dv_inline();
+        let sync_engine = SyncEngine::new();
+        let fs_client = sync_engine.get_file_system_client();
+        let parent = Url::parse("http://not.used").unwrap();
+        let row_idx = example.row_indexes(fs_client, &parent).unwrap();
+
+        assert_eq!(row_idx.len(), 6);
+        assert_eq!(&row_idx, &[3, 4, 7, 11, 18, 29]);
+    }
 }

kernel/src/engine/sync/mod.rs

@@ -9,7 +9,7 @@ mod fs_client;
 pub(crate) mod json;
 mod parquet;
 
-/// This is a simple implemention of [`Engine`]. It only supports reading data from the local
+/// This is a simple implementation of [`Engine`]. It only supports reading data from the local
 /// filesystem, and internally represents data using `Arrow`.
 pub struct SyncEngine {
     fs_client: Arc<fs_client::SyncFilesystemClient>,

kernel/src/scan/state.rs

@@ -66,6 +66,21 @@ impl DvInfo {
             .transpose()?;
         Ok(dv_treemap.map(treemap_to_bools))
     }
+
+    /// Returns a vector of row indexes that should be *removed* from the result set
+    pub fn get_row_indexes(
+        &self,
+        engine: &dyn Engine,
+        table_root: &url::Url,
+    ) -> DeltaResult<Option<Vec<u64>>> {
+        self.deletion_vector
+            .as_ref()
+            .map(|dv| {
+                let fs_client = engine.get_file_system_client();
+                dv.row_indexes(fs_client, table_root)
+            })
+            .transpose()
+    }
 }
 
 pub type ScanCallback<T> = fn(