Introduce TableChangesScan::execute and ScanFileReader

ffi/src/error.rs

@@ -50,6 +50,7 @@ pub enum KernelError {
     UnsupportedError,
     ParseIntervalError,
     ChangeDataFeedUnsupported,
+    ChangeDataFeedIncompatibleSchema,
 }
 
 impl From<Error> for KernelError {
@@ -104,6 +105,9 @@ impl From<Error> for KernelError {
             Error::Unsupported(_) => KernelError::UnsupportedError,
             Error::ParseIntervalError(_) => KernelError::ParseIntervalError,
             Error::ChangeDataFeedUnsupported(_) => KernelError::ChangeDataFeedUnsupported,
+            Error::ChangeDataFeedIncompatibleSchema(_, _) => {
+                KernelError::ChangeDataFeedIncompatibleSchema
+            }
         }
     }
 }

kernel/examples/change_data_feed/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "change_data_feed"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+arrow-array = { workspace = true }
+arrow-schema = { workspace = true }
+clap = { version = "4.5", features = ["derive"] }
+delta_kernel = { path = "../../../kernel", features = [
+  "cloud",
+  "default-engine",
+  "developer-visibility",
+  "sync-engine"
+] }
+env_logger = "0.11.3"
+url = "2"
+itertools = "0.13"
+arrow = { workspace = true, features = ["prettyprint"] }

kernel/examples/change_data_feed/src/main.rs

@@ -0,0 +1,52 @@
+use std::sync::Arc;
+
+use arrow::{compute::filter_record_batch, util::pretty::print_batches};
+use arrow_array::RecordBatch;
+use clap::Parser;
+use delta_kernel::{
+    engine::{arrow_data::ArrowEngineData, sync::SyncEngine},
+    DeltaResult, Table,
+};
+use itertools::Itertools;
+
+#[derive(Parser)]
+#[command(author, version, about, long_about = None)]
+#[command(propagate_version = true)]
+struct Cli {
+    /// Path to the table to inspect
+    path: String,
+
+    start_version: u64,
+
+    end_version: Option<u64>,
+}
+
+fn main() -> DeltaResult<()> {
+    let cli = Cli::parse();
+    let table = Table::try_from_uri(cli.path)?;
+    let engine = Arc::new(SyncEngine::new());
+    let table_changes = table.table_changes(engine.as_ref(), cli.start_version, cli.end_version)?;
+
+    let x = table_changes.into_scan_builder().build()?;
+    let batches: Vec<RecordBatch> = x
+        .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
+                .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)
+            }
+        })
+        .try_collect()?;
+    print_batches(&batches)?;
+
+    Ok(())
+}

kernel/src/actions/deletion_vector.rs

@@ -1,10 +1,9 @@
 //! Code relating to parsing and using deletion vectors
 
-use std::io::{Cursor, Read};
-use std::sync::Arc;
-
 use bytes::Bytes;
 use roaring::RoaringTreemap;
+use std::io::{Cursor, Read};
+use std::sync::Arc;
 use url::Url;
 
 use delta_kernel_derive::Schema;
@@ -13,6 +12,7 @@ use crate::utils::require;
 use crate::{DeltaResult, Error, FileSystemClient};
 
 #[derive(Debug, Clone, PartialEq, Eq, Schema)]
+#[cfg_attr(test, derive(serde::Serialize), serde(rename_all = "camelCase"))]
 pub struct DeletionVectorDescriptor {
     /// A single character to indicate how to access the DV. Legal options are: ['u', 'i', 'p'].
     pub storage_type: String,

kernel/src/actions/mod.rs

@@ -85,6 +85,7 @@ pub(crate) fn get_log_commit_info_schema() -> &'static SchemaRef {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Schema)]
+#[cfg_attr(test, derive(Serialize), serde(rename_all = "camelCase"))]
 pub struct Format {
     /// Name of the encoding for files in this table
     pub provider: String,
@@ -102,6 +103,7 @@ impl Default for Format {
 }
 
 #[derive(Debug, Default, Clone, PartialEq, Eq, Schema)]
+#[cfg_attr(test, derive(Serialize), serde(rename_all = "camelCase"))]
 pub struct Metadata {
     /// Unique identifier for this table
     pub id: String,
@@ -325,6 +327,7 @@ where
 #[derive(Debug, Clone, PartialEq, Eq, Schema)]
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 #[cfg_attr(not(feature = "developer-visibility"), visibility::make(pub(crate)))]
+#[cfg_attr(test, derive(Serialize, Default), serde(rename_all = "camelCase"))]
 struct CommitInfo {
     /// The time this logical file was created, as milliseconds since the epoch.
     /// Read: optional, write: required (that is, kernel always writes).
@@ -346,6 +349,7 @@ struct CommitInfo {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Schema)]
+#[cfg_attr(test, derive(Serialize, Default), serde(rename_all = "camelCase"))]
 pub struct Add {
     /// A relative path to a data file from the root of the table or an absolute path to a file
     /// that should be added to the table. The path is a URI as specified by
@@ -374,23 +378,29 @@ pub struct Add {
     /// Contains [statistics] (e.g., count, min/max values for columns) about the data in this logical file.
     ///
     /// [statistics]: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#Per-file-Statistics
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub stats: Option<String>,
 
     /// Map containing metadata about this logical file.
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub tags: Option<HashMap<String, String>>,
 
     /// Information about deletion vector (DV) associated with this add action
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub deletion_vector: Option<DeletionVectorDescriptor>,
 
     /// Default generated Row ID of the first row in the file. The default generated Row IDs
     /// of the other rows in the file can be reconstructed by adding the physical index of the
     /// row within the file to the base Row ID
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub base_row_id: Option<i64>,
 
     /// First commit version in which an add action with the same path was committed to the table.
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub default_row_commit_version: Option<i64>,
 
     /// The name of the clustering implementation
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub clustering_provider: Option<String>,
 }
 
@@ -403,6 +413,7 @@ impl Add {
 #[derive(Debug, Clone, PartialEq, Eq, Schema)]
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 #[cfg_attr(not(feature = "developer-visibility"), visibility::make(pub(crate)))]
+#[cfg_attr(test, derive(Serialize, Default), serde(rename_all = "camelCase"))]
 struct Remove {
     /// A relative path to a data file from the root of the table or an absolute path to a file
     /// that should be added to the table. The path is a URI as specified by
@@ -412,39 +423,48 @@ struct Remove {
     pub(crate) path: String,
 
     /// The time this logical file was created, as milliseconds since the epoch.
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) deletion_timestamp: Option<i64>,
 
     /// When `false` the logical file must already be present in the table or the records
     /// in the added file must be contained in one or more remove actions in the same version.
     pub(crate) data_change: bool,
 
     /// When true the fields `partition_values`, `size`, and `tags` are present
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) extended_file_metadata: Option<bool>,
 
     /// A map from partition column to value for this logical file.
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) partition_values: Option<HashMap<String, String>>,
 
     /// The size of this data file in bytes
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) size: Option<i64>,
 
     /// Map containing metadata about this logical file.
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) tags: Option<HashMap<String, String>>,
 
     /// Information about deletion vector (DV) associated with this add action
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) deletion_vector: Option<DeletionVectorDescriptor>,
 
     /// Default generated Row ID of the first row in the file. The default generated Row IDs
     /// of the other rows in the file can be reconstructed by adding the physical index of the
     /// row within the file to the base Row ID
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) base_row_id: Option<i64>,
 
     /// First commit version in which an add action with the same path was committed to the table.
+    #[cfg_attr(test, serde(skip_serializing_if = "Option::is_none"))]
     pub(crate) default_row_commit_version: Option<i64>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Schema)]
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 #[cfg_attr(not(feature = "developer-visibility"), visibility::make(pub(crate)))]
+#[cfg_attr(test, derive(Serialize, Default), serde(rename_all = "camelCase"))]
 struct Cdc {
     /// A relative path to a change data file from the root of the table or an absolute path to a
     /// change data file that should be added to the table. The path is a URI as specified by

kernel/src/actions/visitors.rs

@@ -122,7 +122,7 @@ struct ProtocolVisitor {
 
 impl ProtocolVisitor {
     #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
-    fn visit_protocol<'a>(
+    pub(crate) fn visit_protocol<'a>(
         row_index: usize,
         min_reader_version: i32,
         getters: &[&'a dyn GetData<'a>],

kernel/src/engine/sync/mod.rs

@@ -15,7 +15,7 @@ use tracing::debug;
 
 mod fs_client;
 pub(crate) mod json;
-mod parquet;
+pub(crate) mod parquet;
 
 /// This is a simple implementation of [`Engine`]. It only supports reading data from the local
 /// filesystem, and internally represents data using `Arrow`.

kernel/src/error.rs

@@ -6,7 +6,7 @@ use std::{
     str::Utf8Error,
 };
 
-use crate::schema::DataType;
+use crate::schema::{DataType, StructType};
 use crate::table_properties::ParseIntervalError;
 use crate::Version;
 
@@ -188,6 +188,9 @@ pub enum Error {
 
     #[error("Change data feed is unsupported for the table at version {0}")]
     ChangeDataFeedUnsupported(Version),
+
+    #[error("Change data feed encountered incompatible schema. Expected {0}, got {1}")]
+    ChangeDataFeedIncompatibleSchema(String, String),
 }
 
 // Convenience constructors for Error types that take a String argument
@@ -254,6 +257,12 @@ impl Error {
     pub fn change_data_feed_unsupported(version: impl Into<Version>) -> Self {
         Self::ChangeDataFeedUnsupported(version.into())
     }
+    pub fn change_data_feed_incompatible_schema(
+        expected: &StructType,
+        actual: &StructType,
+    ) -> Self {
+        Self::ChangeDataFeedIncompatibleSchema(format!("{:?}", expected), format!("{:?}", actual))
+    }
 
     // Capture a backtrace when the error is constructed.
     #[must_use]

kernel/src/expressions/scalars.rs

@@ -151,6 +151,13 @@ impl Scalar {
     pub fn is_null(&self) -> bool {
         matches!(self, Self::Null(_))
     }
+
+    pub fn timestamp_from_millis(millis: i64) -> DeltaResult<Self> {
+        let Some(timestamp) = DateTime::from_timestamp_millis(millis) else {
+            return Err(Error::generic("Failed to convert timestamp"));
+        };
+        Ok(Self::Timestamp(timestamp.timestamp_micros()))
+    }
 }
 
 impl Display for Scalar {

kernel/src/path.rs

@@ -14,7 +14,7 @@ const MULTIPART_PART_LEN: usize = 10;
 /// The number of characters in the uuid part of a uuid checkpoint
 const UUID_PART_LEN: usize = 36;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 #[cfg_attr(not(feature = "developer-visibility"), visibility::make(pub(crate)))]
 enum LogPathFileType {
@@ -37,7 +37,7 @@ enum LogPathFileType {
     Unknown,
 }
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
 #[cfg_attr(not(feature = "developer-visibility"), visibility::make(pub(crate)))]
 struct ParsedLogPath<Location: AsUrl = FileMeta> {

kernel/src/scan/data_skipping.rs

@@ -41,6 +41,7 @@ fn as_data_skipping_predicate(expr: &Expr, inverted: bool) -> Option<Expr> {
     DataSkippingPredicateCreator.eval_expr(expr, inverted)
 }
 
+#[derive(Clone)]
 pub(crate) struct DataSkippingFilter {
     stats_schema: SchemaRef,
     select_stats_evaluator: Arc<dyn ExpressionEvaluator>,

kernel/src/scan/log_replay.rs

@@ -180,7 +180,7 @@ pub(crate) static SCAN_ROW_SCHEMA: LazyLock<Arc<StructType>> = LazyLock::new(||
     ]))
 });
 
-static SCAN_ROW_DATATYPE: LazyLock<DataType> = LazyLock::new(|| SCAN_ROW_SCHEMA.clone().into());
+pub static SCAN_ROW_DATATYPE: LazyLock<DataType> = LazyLock::new(|| SCAN_ROW_SCHEMA.clone().into());
 
 fn get_add_transform_expr() -> Expression {
     Expression::Struct(vec![

kernel/src/scan/mod.rs

@@ -19,7 +19,7 @@ use crate::{DeltaResult, Engine, EngineData, Error, FileMeta};
 use self::log_replay::scan_action_iter;
 use self::state::GlobalScanState;
 
-mod data_skipping;
+pub mod data_skipping;
 pub mod log_replay;
 pub mod state;
 
@@ -124,7 +124,7 @@ pub struct ScanResult {
     pub raw_data: DeltaResult<Box<dyn EngineData>>,
     /// Raw row mask.
     // TODO(nick) this should be allocated by the engine
-    raw_mask: Option<Vec<bool>>,
+    pub(crate) raw_mask: Option<Vec<bool>>,
 }
 
 impl ScanResult {
@@ -159,7 +159,7 @@ impl ScanResult {
 /// store the name of the column, as that's all that's needed during the actual query. For
 /// `Partition` we store an index into the logical schema for this query since later we need the
 /// data type as well to materialize the partition column.
-#[derive(PartialEq, Debug)]
+#[derive(PartialEq, Debug, Clone)]
 pub enum ColumnType {
     // A column, selected from the data, as is
     Selected(String),
@@ -380,7 +380,7 @@ pub fn scan_row_schema() -> Schema {
     log_replay::SCAN_ROW_SCHEMA.as_ref().clone()
 }
 
-fn parse_partition_value(raw: Option<&String>, data_type: &DataType) -> DeltaResult<Scalar> {
+pub fn parse_partition_value(raw: Option<&String>, data_type: &DataType) -> DeltaResult<Scalar> {
     match (raw, data_type.as_primitive_opt()) {
         (Some(v), Some(primitive)) => primitive.parse_scalar(v),
         (Some(_), None) => Err(Error::generic(format!(