refactor!: update ScanMetadata to struct with new FilteredEngineData type (#768)

## What changes are proposed in this pull request?

1. Updated `ScanMetata` from typed tuple to struct. ScanMetadata is now
a struct with fields:
- filtered_data: A `FilteredEngineData` instance.
- transforms: A vector of transformations to be applied to the data read
from the files

2. Introduction of `FilteredEngineData` type:
Couples `EngineData` with a selection vector indicating which rows to
process.
This type is returned from the`scan_metadata` API and the incoming
`checkpoint` API

3. Updates `visit_scan_files` parameters to accept `ScanMetadata` to
avoid de-structuring.

4. Corresponding FFI changes for `visit_scan_files` to accept
`ScanMetadata` param

All current tests pass.

Author: sebastiantia

ffi/examples/read-table/read_table.c

@@ -86,20 +86,25 @@ void scan_row_callback(
 // For each chunk of scan metadata (which may contain multiple files to scan), kernel will call this
 // function (named do_visit_scan_metadata to avoid conflict with visit_scan_metadata exported by
 // kernel)
-void do_visit_scan_metadata(
-  void* engine_context,
-  ExclusiveEngineData* engine_data,
-  KernelBoolSlice selection_vec,
-  const CTransforms* transforms)
-{
+void do_visit_scan_metadata(void* engine_context, HandleSharedScanMetadata scan_metadata) {
   print_diag("\nScan iterator found some data to read\n  Of this data, here is "
              "a selection vector\n");
-  print_selection_vector("    ", &selection_vec);
+  struct EngineContext* context = engine_context;
+
+  ExternResultKernelBoolSlice selection_vector_res =
+    selection_vector_from_scan_metadata(scan_metadata, context->engine);
+  if (selection_vector_res.tag != OkKernelBoolSlice) {
+    printf("Could not get selection vector from kernel\n");
+    exit(-1);
+  }
+  KernelBoolSlice selection_vector = selection_vector_res.ok;
+  print_selection_vector("    ", &selection_vector);
+
   // Ask kernel to iterate each individual file and call us back with extracted metadata
   print_diag("Asking kernel to call us back for each scan row (file to read)\n");
-  visit_scan_metadata(engine_data, selection_vec, transforms, engine_context, scan_row_callback);
-  free_bool_slice(selection_vec);
-  free_engine_data(engine_data);
+  visit_scan_metadata(scan_metadata, engine_context, scan_row_callback);
+  free_bool_slice(selection_vector);
+  free_scan_metadata(scan_metadata);
 }
 
 // Called for each element of the partition StringSliceIterator. We just turn the slice into a

ffi/src/scan.rs

@@ -3,7 +3,7 @@
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
 
-use delta_kernel::scan::state::{visit_scan_files, DvInfo, GlobalScanState};
+use delta_kernel::scan::state::{DvInfo, GlobalScanState};
 use delta_kernel::scan::{Scan, ScanMetadata};
 use delta_kernel::snapshot::Snapshot;
 use delta_kernel::{DeltaResult, Error, Expression, ExpressionRef};
@@ -16,9 +16,9 @@ use crate::expressions::engine::{
 };
 use crate::expressions::SharedExpression;
 use crate::{
-    kernel_string_slice, AllocateStringFn, ExclusiveEngineData, ExternEngine, ExternResult,
-    IntoExternResult, KernelBoolSlice, KernelRowIndexArray, KernelStringSlice, NullableCvoid,
-    SharedExternEngine, SharedSchema, SharedSnapshot, TryFromStringSlice,
+    kernel_string_slice, AllocateStringFn, ExternEngine, ExternResult, IntoExternResult,
+    KernelBoolSlice, KernelRowIndexArray, KernelStringSlice, NullableCvoid, SharedExternEngine,
+    SharedSchema, SharedSnapshot, TryFromStringSlice,
 };
 
 use super::handle::Handle;
@@ -29,6 +29,38 @@ use super::handle::Handle;
 #[handle_descriptor(target=Scan, mutable=false, sized=true)]
 pub struct SharedScan;
 
+#[handle_descriptor(target=ScanMetadata, mutable=false, sized=true)]
+pub struct SharedScanMetadata;
+
+/// Drop a `SharedScanMetadata`.
+///
+/// # Safety
+///
+/// Caller is responsible for passing a valid scan data handle.
+#[no_mangle]
+pub unsafe extern "C" fn free_scan_metadata(scan_metadata: Handle<SharedScanMetadata>) {
+    scan_metadata.drop_handle();
+}
+
+/// Get a selection vector out of a [`SharedScanMetadata`] struct
+///
+/// # Safety
+/// Engine is responsible for providing valid pointers for each argument
+#[no_mangle]
+pub unsafe extern "C" fn selection_vector_from_scan_metadata(
+    scan_metadata: Handle<SharedScanMetadata>,
+    engine: Handle<SharedExternEngine>,
+) -> ExternResult<KernelBoolSlice> {
+    let scan_metadata = unsafe { scan_metadata.as_ref() };
+    selection_vector_from_scan_metadata_impl(scan_metadata).into_extern_result(&engine.as_ref())
+}
+
+fn selection_vector_from_scan_metadata_impl(
+    scan_metadata: &ScanMetadata,
+) -> DeltaResult<KernelBoolSlice> {
+    Ok(scan_metadata.scan_files.selection_vector.clone().into())
+}
+
 /// Drops a scan.
 ///
 /// # Safety
@@ -175,9 +207,10 @@ fn scan_metadata_iter_init_impl(
     Ok(Arc::new(data).into())
 }
 
-/// Call the provided `engine_visitor` on the next scan metadata item. The visitor will be provided
-/// with a selection vector and engine data. It is the responsibility of the _engine_ to free these
-/// when it is finished by calling [`free_bool_slice`] and [`free_engine_data`] respectively.
+/// Call the provided `engine_visitor` on the next scan metadata item. The visitor will be provided with
+/// a [`SharedScanMetadata`], which contains the actual scan files and the associated selection vector. It is the
+/// responsibility of the _engine_ to free the associated resources after use by calling
+/// [`free_engine_data`] and [`free_bool_slice`] respectively.
 ///
 /// # Safety
 ///
@@ -192,9 +225,7 @@ pub unsafe extern "C" fn scan_metadata_next(
     engine_context: NullableCvoid,
     engine_visitor: extern "C" fn(
         engine_context: NullableCvoid,
-        engine_data: Handle<ExclusiveEngineData>,
-        selection_vector: KernelBoolSlice,
-        transforms: &CTransforms,
+        scan_metadata: Handle<SharedScanMetadata>,
     ),
 ) -> ExternResult<bool> {
     let data = unsafe { data.as_ref() };
@@ -206,19 +237,15 @@ fn scan_metadata_next_impl(
     engine_context: NullableCvoid,
     engine_visitor: extern "C" fn(
         engine_context: NullableCvoid,
-        engine_data: Handle<ExclusiveEngineData>,
-        selection_vector: KernelBoolSlice,
-        transforms: &CTransforms,
+        scan_metadata: Handle<SharedScanMetadata>,
     ),
 ) -> DeltaResult<bool> {
     let mut data = data
         .data
         .lock()
         .map_err(|_| Error::generic("poisoned mutex"))?;
-    if let Some((data, sel_vec, transforms)) = data.next().transpose()? {
-        let bool_slice = KernelBoolSlice::from(sel_vec);
-        let transform_map = CTransforms { transforms };
-        (engine_visitor)(engine_context, data.into(), bool_slice, &transform_map);
+    if let Some(scan_metadata) = data.next().transpose()? {
+        (engine_visitor)(engine_context, Arc::new(scan_metadata).into());
         Ok(true)
     } else {
         Ok(false)
@@ -421,31 +448,24 @@ struct ContextWrapper {
 }
 
 /// Shim for ffi to call visit_scan_metadata. This will generally be called when iterating through scan
-/// data which provides the data handle and selection vector as each element in the iterator.
+/// data which provides the [`SharedScanMetadata`] as each element in the iterator.
 ///
 /// # Safety
-/// engine is responsible for passing a valid [`ExclusiveEngineData`] and selection vector.
+/// engine is responsible for passing a valid [`SharedScanMetadata`].
 #[no_mangle]
 pub unsafe extern "C" fn visit_scan_metadata(
-    data: Handle<ExclusiveEngineData>,
-    selection_vec: KernelBoolSlice,
-    transforms: &CTransforms,
+    scan_metadata: Handle<SharedScanMetadata>,
     engine_context: NullableCvoid,
     callback: CScanCallback,
 ) {
-    let selection_vec = unsafe { selection_vec.as_ref() };
-    let data = unsafe { data.as_ref() };
+    let scan_metadata = unsafe { scan_metadata.as_ref() };
     let context_wrapper = ContextWrapper {
         engine_context,
         callback,
     };
+
     // TODO: return ExternResult to caller instead of panicking?
-    visit_scan_files(
-        data,
-        selection_vec,
-        &transforms.transforms,
-        context_wrapper,
-        rust_callback,
-    )
-    .unwrap();
+    scan_metadata
+        .visit_scan_files(context_wrapper, rust_callback)
+        .unwrap();
 }

kernel/examples/inspect-table/src/main.rs

@@ -209,16 +209,10 @@ fn try_main() -> DeltaResult<()> {
         }
         Commands::ScanMetadata => {
             let scan = ScanBuilder::new(snapshot).build()?;
-            let scan_metadata = scan.scan_metadata(&engine)?;
-            for res in scan_metadata {
-                let (data, vector, transforms) = res?;
-                delta_kernel::scan::state::visit_scan_files(
-                    data.as_ref(),
-                    &vector,
-                    &transforms,
-                    (),
-                    print_scan_file,
-                )?;
+            let scan_metadata_iter = scan.scan_metadata(&engine)?;
+            for res in scan_metadata_iter {
+                let scan_metadata = res?;
+                scan_metadata.visit_scan_files((), print_scan_file)?;
             }
         }
         Commands::Actions { oldest_first } => {

kernel/examples/read-table-multi-threaded/src/main.rs

@@ -210,14 +210,8 @@ fn try_main() -> DeltaResult<()> {
     drop(record_batch_tx);
 
     for res in scan_metadata {
-        let (data, vector, transforms) = res?;
-        scan_file_tx = delta_kernel::scan::state::visit_scan_files(
-            data.as_ref(),
-            &vector,
-            &transforms,
-            scan_file_tx,
-            send_scan_file,
-        )?;
+        let scan_metadata = res?;
+        scan_file_tx = scan_metadata.visit_scan_files(scan_file_tx, send_scan_file)?;
     }
 
     // have sent all scan files, drop this so threads will exit when there's no more work

kernel/src/engine_data.rs

@@ -7,6 +7,19 @@ use tracing::debug;
 
 use std::collections::HashMap;
 
+/// Engine data paired with a selection vector indicating which rows are logically selected.
+///
+/// A value of `true` in the selection vector means the corresponding row is selected (i.e., not deleted),
+/// while `false` means the row is logically deleted and should be ignored.
+///
+/// Interpreting unselected (`false`) rows will result in incorrect/undefined behavior.
+pub struct FilteredEngineData {
+    // The underlying engine data
+    pub data: Box<dyn EngineData>,
+    // The selection vector where `true` marks rows to include in results
+    pub selection_vector: Vec<bool>,
+}
+
 /// a trait that an engine exposes to give access to a list
 pub trait EngineList {
     /// Return the length of the list at the specified row_index in the raw data

kernel/src/scan/log_replay.rs

@@ -364,7 +364,7 @@ impl LogReplayProcessor for ScanLogReplayProcessor {
 
         // TODO: Teach expression eval to respect the selection vector we just computed so carefully!
         let result = self.add_transform.evaluate(actions_batch)?;
-        Ok((
+        Ok(ScanMetadata::new(
             result,
             visitor.selection_vector,
             visitor.row_transform_exprs,
@@ -470,8 +470,11 @@ mod tests {
             None,
         );
         for res in iter {
-            let (_batch, _sel, transforms) = res.unwrap();
-            assert!(transforms.is_empty(), "Should have no transforms");
+            let scan_metadata = res.unwrap();
+            assert!(
+                scan_metadata.scan_file_transforms.is_empty(),
+                "Should have no transforms"
+            );
         }
     }
 
@@ -516,7 +519,8 @@ mod tests {
         }
 
         for res in iter {
-            let (_batch, _sel, transforms) = res.unwrap();
+            let scan_metadata = res.unwrap();
+            let transforms = scan_metadata.scan_file_transforms;
             // in this case we have a metadata action first and protocol 3rd, so we expect 4 items,
             // the first and 3rd being a `None`
             assert_eq!(transforms.len(), 4, "Should have 4 transforms");