feat: better ways to use deletion vectors

ffi/src/lib.rs

@@ -127,6 +127,7 @@ 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 {
+
     /// 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::drop_bool_slice] exactly once.
@@ -136,6 +137,12 @@ mod private {
         len: usize,
     }
 
+    #[repr(C)]
+    pub struct KernelRowIndexArray {
+        ptr: *mut u64,
+        len: usize,
+    }
+
     impl KernelBoolSlice {
         /// Creates an empty slice.
         pub fn empty() -> KernelBoolSlice {
@@ -189,13 +196,50 @@ mod private {
     /// memory, but must only free it by calling [super::drop_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 {}
+    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 been
+        /// already been consumed by a previous call to this method.
+        pub unsafe fn into_vec(self) -> Vec<u64> {
+            if self.ptr.is_null() {
+                Default::default()
+            } else {
+                Vec::from_raw_parts(self.ptr, self.len, self.len)
+            }
+        }
+
+        /// Creates an empty slice.
+        pub fn empty() -> KernelRowIndexArray {
+            Self {
+                ptr: std::ptr::null_mut(),
+                len: 0,
+            }
+        }
+    }
+
+    impl From<Vec<u64>> for KernelRowIndexArray {
+        fn from(mut vec: Vec<u64>) -> Self {
+            vec.shrink_to_fit();
+            let len = vec.len();
+            let boxed = vec.into_boxed_slice();
+            let ptr = Box::into_raw(boxed).cast();
+            KernelRowIndexArray { ptr, len }
+        }
+    }
 }
 pub use private::KernelBoolSlice;
+pub use private::KernelRowIndexArray;
 
 /// # Safety
 ///
@@ -206,6 +250,14 @@ pub unsafe extern "C" fn drop_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();
+}
+
 #[repr(C)]
 #[derive(Debug)]
 pub enum KernelError {

ffi/src/scan.rs

@@ -14,8 +14,8 @@ use url::Url;
 
 use crate::{
     unwrap_kernel_expression, AllocateStringFn, EnginePredicate, ExternEngine, ExternResult,
-    IntoExternResult, KernelBoolSlice, KernelExpressionVisitorState, KernelStringSlice,
-    NullableCvoid, SharedExternEngine, SharedSnapshot, TryFromStringSlice,
+    IntoExternResult, KernelBoolSlice, KernelExpressionVisitorState, KernelRowIndexArray,
+    KernelStringSlice, NullableCvoid, SharedExternEngine, SharedSnapshot, TryFromStringSlice,
 };
 
 use super::handle::Handle;
@@ -330,6 +330,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 to_deletion_vector(
+        &self,
+        fs_client: Arc<dyn FileSystemClient>,
+        parent: &Url,
+    ) -> DeltaResult<DeletionVector> {
+        DeletionVector::new(fs_client, parent, self)
+    }
 }
 
 enum Endian {
@@ -223,15 +232,37 @@ pub(crate) fn treemap_to_bools(treemap: RoaringTreemap) -> Vec<bool> {
     }
 }
 
+#[derive(Debug, Clone)]
+#[repr(C)]
+pub struct DeletionVector {
+    inner: RoaringTreemap,
+}
+
+impl DeletionVector {
+    pub fn new(
+        fs_client: Arc<dyn FileSystemClient>,
+        parent: &Url,
+        descriptor: &DeletionVectorDescriptor,
+    ) -> DeltaResult<Self> {
+        let inner = descriptor.read(fs_client, parent)?;
+        Ok(Self { inner })
+    }
+
+    pub fn row_indexes(self) -> Vec<u64> {
+        self.inner.into_iter().collect::<Vec<u64>>()
+    }
+}
+
 #[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 {
@@ -358,4 +389,17 @@ 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 dv = example.to_deletion_vector(fs_client, &parent).unwrap();
+        let row_idx = dv.row_indexes();
+
+        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

@@ -48,6 +48,24 @@ impl DvInfo {
             .transpose()?;
         Ok(dv_treemap.map(treemap_to_bools))
     }
+
+    pub fn get_row_indexes(
+        &self,
+        engine: &dyn Engine,
+        table_root: &url::Url,
+    ) -> DeltaResult<Option<Vec<u64>>> {
+        let indexes = self
+            .deletion_vector
+            .as_ref()
+            .map(|dv_descriptor| {
+                let fs_client = engine.get_file_system_client();
+                dv_descriptor.to_deletion_vector(fs_client, table_root)
+            })
+            .transpose()?
+            .map(|dv| dv.row_indexes());
+
+        Ok(indexes)
+    }
 }
 
 /// Request that the kernel call a callback on each valid file that needs to be read for the