Part 4: read_table.c uses transform in ffi

ffi/examples/read-table/arrow.c

@@ -66,7 +66,7 @@ static GArrowRecordBatch* add_partition_columns(
     char* col = partition_cols->cols[i];
     guint pos = cols + i;
     KernelStringSlice key = { col, strlen(col) };
-    char* partition_val = get_from_map(partition_values, key, allocate_string);
+    char* partition_val = get_from_string_map(partition_values, key, allocate_string);
     print_diag(
       "  Adding partition column '%s' with value '%s' at column %u\n",
       col,

ffi/examples/read-table/read_table.c

@@ -28,7 +28,7 @@ void print_partition_info(struct EngineContext* context, const CStringMap* parti
   for (uintptr_t i = 0; i < context->partition_cols->len; i++) {
     char* col = context->partition_cols->cols[i];
     KernelStringSlice key = { col, strlen(col) };
-    char* partition_val = get_from_map(partition_values, key, allocate_string);
+    char* partition_val = get_from_string_map(partition_values, key, allocate_string);
     if (partition_val) {
       print_diag("  partition '%s' here: %s\n", col, partition_val);
       free(partition_val);
@@ -87,14 +87,15 @@ void scan_row_callback(
 void do_visit_scan_data(
   void* engine_context,
   ExclusiveEngineData* engine_data,
-  KernelBoolSlice selection_vec)
+  KernelBoolSlice selection_vec,
+  const CTransformMap* transforms)
 {
   print_diag("\nScan iterator found some data to read\n  Of this data, here is "
              "a selection vector\n");
   print_selection_vector("    ", &selection_vec);
   // 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_data(engine_data, selection_vec, engine_context, scan_row_callback);
+  visit_scan_data(engine_data, selection_vec, transforms, engine_context, scan_row_callback);
   free_bool_slice(selection_vec);
   free_engine_data(engine_data);
 }

ffi/src/scan.rs

@@ -7,7 +7,7 @@ use delta_kernel::scan::state::{visit_scan_files, DvInfo, GlobalScanState};
 use delta_kernel::scan::{Scan, ScanData};
 use delta_kernel::schema::Schema;
 use delta_kernel::snapshot::Snapshot;
-use delta_kernel::{DeltaResult, Error};
+use delta_kernel::{DeltaResult, Error, ExpressionRef};
 use delta_kernel_ffi_macros::handle_descriptor;
 use tracing::debug;
 use url::Url;
@@ -211,6 +211,7 @@ pub unsafe extern "C" fn kernel_scan_data_next(
         engine_context: NullableCvoid,
         engine_data: Handle<ExclusiveEngineData>,
         selection_vector: KernelBoolSlice,
+        transforms: &CTransformMap,
     ),
 ) -> ExternResult<bool> {
     let data = unsafe { data.as_ref() };
@@ -224,15 +225,17 @@ fn kernel_scan_data_next_impl(
         engine_context: NullableCvoid,
         engine_data: Handle<ExclusiveEngineData>,
         selection_vector: KernelBoolSlice,
+        transforms: &CTransformMap,
     ),
 ) -> DeltaResult<bool> {
     let mut data = data
         .data
         .lock()
         .map_err(|_| Error::generic("poisoned mutex"))?;
-    if let Some((data, sel_vec)) = data.next().transpose()? {
+    if let Some((data, sel_vec, transforms)) = data.next().transpose()? {
         let bool_slice = KernelBoolSlice::from(sel_vec);
-        (engine_visitor)(engine_context, data.into(), bool_slice);
+        let transform_map = CTransformMap { transforms };
+        (engine_visitor)(engine_context, data.into(), bool_slice, &transform_map);
         Ok(true)
     } else {
         Ok(false)
@@ -281,7 +284,7 @@ pub struct CStringMap {
 /// # Safety
 ///
 /// The engine is responsible for providing a valid [`CStringMap`] pointer and [`KernelStringSlice`]
-pub unsafe extern "C" fn get_from_map(
+pub unsafe extern "C" fn get_from_string_map(
     map: &CStringMap,
     key: KernelStringSlice,
     allocate_fn: AllocateStringFn,
@@ -293,6 +296,32 @@ pub unsafe extern "C" fn get_from_map(
         .and_then(|v| allocate_fn(kernel_string_slice!(v)))
 }
 
+
+pub struct CTransformMap {
+    transforms: HashMap<usize, ExpressionRef>,
+}
+
+// #[no_mangle]
+// /// allow probing into a CStringMap. If the specified key is in the map, kernel will call
+// /// allocate_fn with the value associated with the key and return the value returned from that
+// /// function. If the key is not in the map, this will return NULL
+// ///
+// /// # Safety
+// ///
+// /// The engine is responsible for providing a valid [`CStringMap`] pointer and [`KernelStringSlice`]
+// pub unsafe extern "C" fn get_from_transform_map(
+//     map: &CTransformMap,
+//     key: usize,
+//     allocate_fn: AllocateStringFn,
+// ) -> NullableCvoid {
+//     // TODO: Return ExternResult to caller instead of panicking?
+//     let string_key = unsafe { TryFromStringSlice::try_from_slice(&key) };
+//     map.values
+//         .get(string_key.unwrap())
+//         .and_then(|v| allocate_fn(kernel_string_slice!(v)))
+// }
+
+
 /// Get a selection vector out of a [`DvInfo`] struct
 ///
 /// # Safety
@@ -355,6 +384,7 @@ fn rust_callback(
     size: i64,
     kernel_stats: Option<delta_kernel::scan::state::Stats>,
     dv_info: DvInfo,
+    _transform: Option<ExpressionRef>,
     partition_values: HashMap<String, String>,
 ) {
     let partition_map = CStringMap {
@@ -388,6 +418,7 @@ struct ContextWrapper {
 pub unsafe extern "C" fn visit_scan_data(
     data: Handle<ExclusiveEngineData>,
     selection_vec: KernelBoolSlice,
+    transforms: &CTransformMap,
     engine_context: NullableCvoid,
     callback: CScanCallback,
 ) {
@@ -398,5 +429,5 @@ pub unsafe extern "C" fn visit_scan_data(
         callback,
     };
     // TODO: return ExternResult to caller instead of panicking?
-    visit_scan_files(data, selection_vec, context_wrapper, rust_callback).unwrap();
+    visit_scan_files(data, selection_vec, &transforms.transforms, context_wrapper, rust_callback).unwrap();
 }

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

@@ -12,7 +12,7 @@ use delta_kernel::expressions::ColumnName;
 use delta_kernel::scan::state::{DvInfo, Stats};
 use delta_kernel::scan::ScanBuilder;
 use delta_kernel::schema::{ColumnNamesAndTypes, DataType};
-use delta_kernel::{DeltaResult, Error, Table};
+use delta_kernel::{DeltaResult, Error, ExpressionRef, Table};
 
 use std::collections::HashMap;
 use std::process::ExitCode;
@@ -163,6 +163,7 @@ fn print_scan_file(
     size: i64,
     stats: Option<Stats>,
     dv_info: DvInfo,
+    transform: Option<ExpressionRef>,
     partition_values: HashMap<String, String>,
 ) {
     let num_record_str = if let Some(s) = stats {
@@ -176,6 +177,7 @@ fn print_scan_file(
               Size (bytes):\t{size}\n  \
               Num Records:\t{num_record_str}\n  \
               Has DV?:\t{}\n  \
+              TransformExpr:\t {transform:?}\n \
               Part Vals:\t{partition_values:?}",
         dv_info.has_vector()
     );
@@ -209,10 +211,11 @@ fn try_main() -> DeltaResult<()> {
             let scan = ScanBuilder::new(snapshot).build()?;
             let scan_data = scan.scan_data(&engine)?;
             for res in scan_data {
-                let (data, vector) = res?;
+                let (data, vector, transforms) = res?;
                 delta_kernel::scan::state::visit_scan_files(
                     data.as_ref(),
                     &vector,
+                    &transforms,
                     (),
                     print_scan_file,
                 )?;

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

@@ -15,7 +15,7 @@ use delta_kernel::engine::sync::SyncEngine;
 use delta_kernel::scan::state::{DvInfo, GlobalScanState, Stats};
 use delta_kernel::scan::transform_to_logical;
 use delta_kernel::schema::Schema;
-use delta_kernel::{DeltaResult, Engine, EngineData, FileMeta, Table};
+use delta_kernel::{DeltaResult, Engine, EngineData, ExpressionRef, FileMeta, Table};
 
 use clap::{Parser, ValueEnum};
 use url::Url;
@@ -111,6 +111,7 @@ fn send_scan_file(
     size: i64,
     _stats: Option<Stats>,
     dv_info: DvInfo,
+    _transform: Option<ExpressionRef>,
     partition_values: HashMap<String, String>,
 ) {
     let scan_file = ScanFile {
@@ -210,10 +211,11 @@ fn try_main() -> DeltaResult<()> {
     drop(record_batch_tx);
 
     for res in scan_data {
-        let (data, vector) = res?;
+        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,
         )?;

kernel/src/engine_data.rs

@@ -129,7 +129,9 @@ pub trait TypedGetData<'a, T> {
     fn get_opt(&'a self, row_index: usize, field_name: &str) -> DeltaResult<Option<T>>;
     fn get(&'a self, row_index: usize, field_name: &str) -> DeltaResult<T> {
         let val = self.get_opt(row_index, field_name)?;
-        val.ok_or_else(|| Error::MissingData(format!("Data missing for field {field_name}")))
+        val.ok_or_else(|| {
+            Error::MissingData(format!("Data missing for field {field_name}")).with_backtrace()
+        })
     }
 }