Implement Log Replay for Change Data Feed

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/src/actions/deletion_vector.rs

@@ -13,6 +13,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,
@@ -287,7 +289,7 @@ impl Protocol {
 }
 
 // given unparsed `table_features`, parse and check if they are subset of `supported_features`
-fn ensure_supported_features<T>(
+pub(crate) fn ensure_supported_features<T>(
     table_features: &[String],
     supported_features: &HashSet<T>,
 ) -> DeltaResult<()>
@@ -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

@@ -121,7 +121,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/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(crate) 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/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/log_replay.rs

@@ -180,7 +180,8 @@ 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(crate) 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(crate) mod data_skipping;
 pub mod log_replay;
 pub mod state;