Merge branch 'main' into feat/nullable-values-in-tags

Author: nicklan

README.md

@@ -153,15 +153,7 @@ Some design principles which should be considered:
 - If using `emacs`, both [eglot](https://github.com/joaotavora/eglot) and
   [lsp-mode](https://github.com/emacs-lsp/lsp-mode) provide excellent integration with
   `rust-analyzer`. [rustic](https://github.com/brotzeit/rustic) is a nice mode as well.
-- When also developing in VS Code it's sometimes convenient to configure rust-analyzer in
-  `.vscode/settings.json`.
-
-```json
-{
-  "editor.formatOnSave": true,
-  "rust-analyzer.cargo.features": ["default-engine"]
-}
-```
+- When also developing in VS Code it's convenient to add rust-analyzer to your workspace.
 
 - The crate's documentation can be easily reviewed with: `cargo docs --open`
 - Code coverage is available on codecov via [cargo-llvm-cov]. See their docs for instructions to install/run locally.

acceptance/src/data.rs

@@ -1,9 +1,7 @@
 use std::{path::Path, sync::Arc};
 
 use delta_kernel::arrow::array::{Array, RecordBatch};
-use delta_kernel::arrow::compute::{
-    concat_batches, filter_record_batch, lexsort_to_indices, take, SortColumn,
-};
+use delta_kernel::arrow::compute::{concat_batches, lexsort_to_indices, take, SortColumn};
 use delta_kernel::arrow::datatypes::{DataType, Schema};
 
 use delta_kernel::parquet::arrow::async_reader::{
@@ -120,23 +118,16 @@ pub async fn assert_scan_metadata(
     let mut schema = None;
     let batches: Vec<RecordBatch> = scan
         .execute(engine)?
-        .map(|scan_result| -> DeltaResult<_> {
-            let scan_result = scan_result?;
-            let mask = scan_result.full_mask();
-            let data = scan_result.raw_data?;
-            let record_batch: RecordBatch = data
+        .map(|data| -> DeltaResult<_> {
+            let record_batch: RecordBatch = data?
                 .into_any()
                 .downcast::<ArrowEngineData>()
                 .unwrap()
                 .into();
             if schema.is_none() {
                 schema = Some(record_batch.schema());
             }
-            if let Some(mask) = mask {
-                Ok(filter_record_batch(&record_batch, &mask.into())?)
-            } else {
-                Ok(record_batch)
-            }
+            Ok(record_batch)
         })
         .try_collect()?;
     let all_data = concat_batches(&schema.unwrap(), batches.iter()).map_err(Error::from)?;

ffi/src/scan.rs

@@ -368,7 +368,7 @@ pub unsafe extern "C" fn visit_string_map(
 /// Transformation expressions that need to be applied to each row `i` in ScanMetadata. You can use
 /// [`get_transform_for_row`] to get the transform for a particular row. If that returns an
 /// associated expression, it _must_ be applied to the data read from the file specified by the
-/// row. The resultant schema for this expression is guaranteed to be `Scan.schema()`. If
+/// row. The resultant schema for this expression is guaranteed to be [`scan_logical_schema()`]. If
 /// `get_transform_for_row` returns `NULL` no expression need be applied and the data read from disk
 /// is already in the correct logical state.
 ///

ffi/src/table_changes.rs

@@ -4,12 +4,11 @@ use std::sync::Arc;
 use std::sync::Mutex;
 
 use delta_kernel::arrow::array::{Array, ArrayData, RecordBatch, StructArray};
-use delta_kernel::arrow::compute::filter_record_batch;
 use delta_kernel::arrow::ffi::to_ffi;
 use delta_kernel::engine::arrow_data::ArrowEngineData;
-use delta_kernel::scan::ScanResult;
 use delta_kernel::table_changes::scan::TableChangesScan;
 use delta_kernel::table_changes::TableChanges;
+use delta_kernel::EngineData;
 use delta_kernel::Error;
 use delta_kernel::{DeltaResult, Version};
 use delta_kernel_ffi_macros::handle_descriptor;
@@ -238,8 +237,10 @@ pub unsafe extern "C" fn table_changes_scan_physical_schema(
     table_changes_scan.physical_schema().clone().into()
 }
 
+type TableChangesData = Mutex<Box<dyn Iterator<Item = DeltaResult<Box<dyn EngineData>>> + Send>>;
+
 pub struct ScanTableChangesIterator {
-    data: Mutex<Box<dyn Iterator<Item = DeltaResult<ScanResult>> + Send>>,
+    data: TableChangesData,
     engine: Arc<dyn ExternEngine>,
 }
 
@@ -314,22 +315,16 @@ fn scan_table_changes_next_impl(data: &ScanTableChangesIterator) -> DeltaResult<
         .lock()
         .map_err(|_| Error::generic("poisoned scan table changes iterator mutex"))?;
 
-    let Some(scan_result) = data.next().transpose()? else {
+    let Some(data) = data.next().transpose()? else {
         return Ok(ArrowFFIData::empty());
     };
 
-    let mask = scan_result.full_mask();
-    let data = scan_result.raw_data?;
-    let mut record_batch: RecordBatch = data
+    let record_batch: RecordBatch = data
         .into_any()
         .downcast::<ArrowEngineData>()
         .map_err(|_| delta_kernel::Error::EngineDataType("ArrowEngineData".to_string()))?
         .into();
 
-    if let Some(mask) = mask {
-        record_batch = filter_record_batch(&record_batch, &mask.into())?;
-    }
-
     let batch_struct_array: StructArray = record_batch.into();
     let array_data: ArrayData = batch_struct_array.into_data();
     let (out_array, out_schema) = to_ffi(&array_data)?;
@@ -485,17 +480,7 @@ mod tests {
     ) -> DeltaResult<Vec<RecordBatch>> {
         let scan_results = scan.execute(engine)?;
         scan_results
-            .map(|scan_result| -> DeltaResult<_> {
-                let scan_result = scan_result?;
-                let mask = scan_result.full_mask();
-                let data = scan_result.raw_data?;
-                let record_batch = to_arrow(data)?;
-                if let Some(mask) = mask {
-                    Ok(filter_record_batch(&record_batch, &mask.into())?)
-                } else {
-                    Ok(record_batch)
-                }
-            })
+            .map(|data| -> DeltaResult<_> { to_arrow(data?) })
             .try_collect()
     }
 

kernel/examples/read-table-changes/src/main.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use clap::Parser;
 use common::{LocationArgs, ParseWithExamples};
 use delta_kernel::arrow::array::RecordBatch;
-use delta_kernel::arrow::{compute::filter_record_batch, util::pretty::print_batches};
+use delta_kernel::arrow::util::pretty::print_batches;
 use delta_kernel::engine::arrow_data::ArrowEngineData;
 use delta_kernel::table_changes::TableChanges;
 use delta_kernel::DeltaResult;
@@ -38,20 +38,13 @@ fn main() -> DeltaResult<()> {
     let table_changes_scan = table_changes.into_scan_builder().build()?;
     let batches: Vec<RecordBatch> = table_changes_scan
         .execute(Arc::new(engine))?
-        .map(|scan_result| -> DeltaResult<_> {
-            let scan_result = scan_result?;
-            let mask = scan_result.full_mask();
-            let data = scan_result.raw_data?;
-            let record_batch: RecordBatch = data
+        .map(|data| -> DeltaResult<_> {
+            let record_batch: RecordBatch = data?
                 .into_any()
                 .downcast::<ArrowEngineData>()
                 .map_err(|_| delta_kernel::Error::EngineDataType("ArrowEngineData".to_string()))?
                 .into();
-            if let Some(mask) = mask {
-                Ok(filter_record_batch(&record_batch, &mask.into())?)
-            } else {
-                Ok(record_batch)
-            }
+            Ok(record_batch)
         })
         .try_collect()?;
     print_batches(&batches)?;

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

@@ -1,7 +1,6 @@
 use std::process::ExitCode;
 use std::sync::Arc;
 
-use arrow::compute::filter_record_batch;
 use arrow::record_batch::RecordBatch;
 use arrow::util::pretty::print_batches;
 use common::{LocationArgs, ParseWithExamples, ScanArgs};
@@ -50,21 +49,14 @@ fn try_main() -> DeltaResult<()> {
     let mut rows_so_far = 0;
     let batches: Vec<RecordBatch> = scan
         .execute(Arc::new(engine))?
-        .map(|scan_result| -> DeltaResult<_> {
+        .map(|data| -> DeltaResult<_> {
             // extract the batches and filter them if they have deletion vectors
-            let scan_result = scan_result?;
-            let mask = scan_result.full_mask();
-            let data = scan_result.raw_data?;
-            let record_batch: RecordBatch = data
+            let record_batch: RecordBatch = data?
                 .into_any()
                 .downcast::<ArrowEngineData>()
                 .map_err(|_| delta_kernel::Error::EngineDataType("ArrowEngineData".to_string()))?
                 .into();
-            if let Some(mask) = mask {
-                Ok(filter_record_batch(&record_batch, &mask.into())?)
-            } else {
-                Ok(record_batch)
-            }
+            Ok(record_batch)
         })
         .scan(&mut rows_so_far, |rows_so_far, record_batch| {
             // handle truncation if we've specified a limit

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

@@ -320,24 +320,13 @@ async fn read_and_display_data(
 
     let batches: Vec<RecordBatch> = scan
         .execute(Arc::new(engine))?
-        .map(|scan_result| -> DeltaResult<_> {
-            let scan_result = scan_result?;
-            let mask = scan_result.full_mask();
-            let data = scan_result.raw_data?;
-            let record_batch: RecordBatch = data
+        .map(|data| -> DeltaResult<_> {
+            let record_batch: RecordBatch = data?
                 .into_any()
                 .downcast::<ArrowEngineData>()
                 .map_err(|_| Error::EngineDataType("ArrowEngineData".to_string()))?
                 .into();
-
-            if let Some(mask) = mask {
-                Ok(arrow::compute::filter_record_batch(
-                    &record_batch,
-                    &mask.into(),
-                )?)
-            } else {
-                Ok(record_batch)
-            }
+            Ok(record_batch)
         })
         .try_collect()?;
 

kernel/src/actions/crc.rs

@@ -150,7 +150,7 @@ mod tests {
     use crate::engine::sync::SyncEngine;
     use crate::schema::derive_macro_utils::ToDataType as _;
     use crate::schema::{ArrayType, DataType, StructField, StructType};
-    use crate::table_features::{ReaderFeature, WriterFeature};
+    use crate::table_features::TableFeature;
     use crate::utils::test_utils::string_array_to_engine_data;
     use crate::Engine;
 
@@ -252,8 +252,8 @@ mod tests {
         let expected_protocol = Protocol {
             min_reader_version: 3,
             min_writer_version: 7,
-            reader_features: Some(vec![ReaderFeature::ColumnMapping]),
-            writer_features: Some(vec![WriterFeature::ColumnMapping]),
+            reader_features: Some(vec![TableFeature::ColumnMapping]),
+            writer_features: Some(vec![TableFeature::ColumnMapping]),
         };
         let expected_metadata = Metadata {
             id: "testId".to_string(),