Add table changes constructor

ffi/src/lib.rs

@@ -383,6 +383,7 @@ pub enum KernelError {
     FileAlreadyExists,
     MissingCommitInfo,
     UnsupportedError,
+    TableChangesDisabled,
 }
 
 impl From<Error> for KernelError {
@@ -435,6 +436,7 @@ impl From<Error> for KernelError {
             Error::FileAlreadyExists(_) => KernelError::FileAlreadyExists,
             Error::MissingCommitInfo => KernelError::MissingCommitInfo,
             Error::Unsupported(_) => KernelError::UnsupportedError,
+            Error::TableChangesDisabled(_) => KernelError::TableChangesDisabled,
         }
     }
 }

kernel/src/error.rs

@@ -7,6 +7,7 @@ use std::{
 };
 
 use crate::schema::DataType;
+use crate::Version;
 
 /// A [`std::result::Result`] that has the kernel [`Error`] as the error variant
 pub type DeltaResult<T, E = Error> = std::result::Result<T, E>;
@@ -179,6 +180,9 @@ pub enum Error {
     /// Some functionality is currently unsupported
     #[error("Unsupported: {0}")]
     Unsupported(String),
+
+    #[error("Change data feed is unsupported for the table at version {0}")]
+    ChangeDataFeedUnsupported(Version),
 }
 
 // Convenience constructors for Error types that take a String argument
@@ -242,6 +246,9 @@ impl Error {
     pub fn unsupported(msg: impl ToString) -> Self {
         Self::Unsupported(msg.to_string())
     }
+    pub fn change_data_feed_unsupported(version: impl Into<Version>) -> Self {
+        Self::ChangeDataFeedUnsupported(version.into())
+    }
 
     // Capture a backtrace when the error is constructed.
     #[must_use]

kernel/src/lib.rs

@@ -64,6 +64,7 @@ pub mod engine_data;
 pub mod error;
 pub mod expressions;
 pub(crate) mod predicates;
+pub mod table_changes;
 pub mod table_features;
 
 #[cfg(feature = "developer-visibility")]

kernel/src/snapshot.rs

@@ -50,7 +50,7 @@ impl Snapshot {
     ///
     /// # Parameters
     ///
-    /// - `location`: url pointing at the table root (where `_delta_log` folder is located)
+    /// - `table_root`: url pointing at the table root (where `_delta_log` folder is located)
     /// - `engine`: Implementation of [`Engine`] apis.
     /// - `version`: target version of the [`Snapshot`]
     pub fn try_new(

kernel/src/table.rs

@@ -8,6 +8,7 @@ use std::path::PathBuf;
 use url::Url;
 
 use crate::snapshot::Snapshot;
+use crate::table_changes::TableChanges;
 use crate::transaction::Transaction;
 use crate::{DeltaResult, Engine, Error, Version};
 
@@ -80,6 +81,23 @@ impl Table {
         Snapshot::try_new(self.location.clone(), engine, version)
     }
 
+    /// Create a [`TableChanges`] to get a change data feed for the table between `start_version`,
+    /// and `end_version`. If no `end_version` is supplied, the latest version will be used as the
+    /// `end_version`.
+    pub fn table_changes(
+        &self,
+        engine: &dyn Engine,
+        start_version: Version,
+        end_version: impl Into<Option<Version>>,
+    ) -> DeltaResult<TableChanges> {
+        TableChanges::try_new(
+            self.location.clone(),
+            engine,
+            start_version,
+            end_version.into(),
+        )
+    }
+
     /// Create a new write transaction for this table.
     pub fn new_transaction(&self, engine: &dyn Engine) -> DeltaResult<Transaction> {
         Transaction::try_new(self.snapshot(engine, None)?)

kernel/src/table_changes/mod.rs

@@ -0,0 +1,209 @@
+//! Provides an API to read the table's change data feed between two versions.
+use std::sync::LazyLock;
+
+use url::Url;
+
+use crate::log_segment::LogSegment;
+use crate::path::AsUrl;
+use crate::schema::{DataType, Schema, StructField, StructType};
+use crate::snapshot::Snapshot;
+use crate::table_features::ColumnMappingMode;
+use crate::{DeltaResult, Engine, Error, Version};
+
+static CDF_FIELDS: LazyLock<[StructField; 3]> = LazyLock::new(|| {
+    [
+        StructField::new("_change_type", DataType::STRING, false),
+        StructField::new("_commit_version", DataType::LONG, false),
+        StructField::new("_commit_timestamp", DataType::TIMESTAMP, false),
+    ]
+});
+
+/// Represents a call to read the Change Data Feed between two versions of a table. The schema of
+/// `TableChanges` will be the schema of the table with three additional columns:
+/// - `_change_type`: String representing the type of change that for that commit. This may be one
+///   of `delete`, `insert`, `update_preimage`, or `update_postimage`.
+/// - `_commit_version`: Long representing the commit the change occurred in.
+/// - `_commit_timestamp`: Time at which the commit occurred. If In-commit timestamps is enabled,
+///   this is retrieved from the [`CommitInfo`] action. Otherwise, the timestamp is the same as the
+///   commit file's modification timestamp.
+///
+/// For more details, see the following sections of the protocol:
+/// - [Add CDC File](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#add-cdc-file)
+/// - [Change Data Files](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#change-data-files).
+///
+/// [`CommitInfo`]: crate::actions::CommitInfo
+#[derive(Debug)]
+pub struct TableChanges {
+    pub log_segment: LogSegment,
+    table_root: Url,
+    end_snapshot: Snapshot,
+    start_version: Version,
+    schema: Schema,
+}
+
+impl TableChanges {
+    /// Creates a new [`TableChanges`] instance for the given version range. This function checks
+    /// these two properties:
+    /// - The change data feed table feature must be enabled in both the start or end versions.
+    /// - The schema at the start and end versions are the same.
+    ///
+    /// Note that this does not check that change data feed is enabled for every commit in the
+    /// range. It also does not check that the schema remains the same for the entire range.
+    ///
+    /// # Parameters
+    /// - `table_root`: url pointing at the table root (where `_delta_log` folder is located)
+    /// - `engine`: Implementation of [`Engine`] apis.
+    /// - `start_version`: The start version of the change data feed
+    /// - `end_version`: The end version (inclusive) of the change data feed. If this is none, this
+    ///   defaults to the newest table version.
+    pub fn try_new(
+        table_root: Url,
+        engine: &dyn Engine,
+        start_version: Version,
+        end_version: Option<Version>,
+    ) -> DeltaResult<Self> {
+        // Both snapshots ensure that reading is supported at the start and end version using
+        // [`Protocol::ensure_read_supported`]. Note that we must still verify that reading is
+        // supported for every intermediary protocol actions
+        // [`Protocol`]: crate::actions::Protocol
+        let start_snapshot =
+            Snapshot::try_new(table_root.as_url().clone(), engine, Some(start_version))?;
+        let end_snapshot = Snapshot::try_new(table_root.as_url().clone(), engine, end_version)?;
+
+        // Verify CDF is enabled at the beginning and end of the interval. We must still check every
+        // intermediary metadata action in the range.
+        let is_cdf_enabled = |snapshot: &Snapshot| {
+            static ENABLE_CDF_FLAG: &str = "delta.enableChangeDataFeed";
+            snapshot
+                .metadata()
+                .configuration
+                .get(ENABLE_CDF_FLAG)
+                .is_some_and(|val| val == "true")
+        };
+        if !is_cdf_enabled(&start_snapshot) {
+            return Err(Error::change_data_feed_unsupported(start_version));
+        } else if !is_cdf_enabled(&end_snapshot) {
+            return Err(Error::change_data_feed_unsupported(end_snapshot.version()));
+        }
+
+        if start_snapshot.schema() != end_snapshot.schema() {
+            return Err(Error::generic(format!(
+                "Failed to build TableChanges: Start and end version schemas are different. Found start version schema {:?} and end version schema {:?}", start_snapshot.schema(), end_snapshot.schema(),
+            )));
+        }
+
+        let log_root = table_root.join("_delta_log/")?;
+        let log_segment = LogSegment::for_table_changes(
+            engine.get_file_system_client().as_ref(),
+            log_root,
+            start_version,
+            end_version,
+        )?;
+
+        let schema = StructType::new(
+            end_snapshot
+                .schema()
+                .fields()
+                .cloned()
+                .chain(CDF_FIELDS.clone()),
+        );
+
+        Ok(TableChanges {
+            table_root,
+            end_snapshot,
+            log_segment,
+            start_version,
+            schema,
+        })
+    }
+
+    /// The start version of the `TableChanges`.
+    pub fn start_version(&self) -> Version {
+        self.start_version
+    }
+    /// The end version (inclusive) of the [`TableChanges`]. If no `end_version` was specified in
+    /// [`TableChanges::try_new`], this returns the newest version as of the call to `try_new`.
+    pub fn end_version(&self) -> Version {
+        self.log_segment.end_version
+    }
+    /// The logical schema of the change data feed. For details on the shape of the schema, see
+    /// [`TableChanges`].
+    pub fn schema(&self) -> &Schema {
+        &self.schema
+    }
+    /// Path to the root of the table that is being read.
+    pub fn table_root(&self) -> &Url {
+        &self.table_root
+    }
+    /// The partition columns that will be read.
+    #[allow(unused)]
+    pub(crate) fn partition_columns(&self) -> &Vec<String> {
+        &self.end_snapshot.metadata().partition_columns
+    }
+    /// The column mapping mode at the end schema.
+    #[allow(unused)]
+    pub(crate) fn column_mapping_mode(&self) -> &ColumnMappingMode {
+        &self.end_snapshot.column_mapping_mode
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::engine::sync::SyncEngine;
+    use crate::schema::DataType;
+    use crate::schema::StructField;
+    use crate::table_changes::CDF_FIELDS;
+    use crate::Error;
+    use crate::Table;
+    use itertools::assert_equal;
+
+    #[test]
+    fn table_changes_checks_enable_cdf_flag() {
+        // Table with CDF enabled, then disabled at 2 and enabled at 3
+        let path = "./tests/data/table-with-cdf";
+        let engine = Box::new(SyncEngine::new());
+        let table = Table::try_from_uri(path).unwrap();
+
+        let valid_ranges = [(0, 1), (0, 0), (1, 1)];
+        for (start_version, end_version) in valid_ranges {
+            let table_changes = table
+                .table_changes(engine.as_ref(), start_version, end_version)
+                .unwrap();
+            assert_eq!(table_changes.start_version, start_version);
+            assert_eq!(table_changes.end_version(), end_version);
+        }
+
+        let invalid_ranges = [(0, 2), (1, 2), (2, 2), (2, 3)];
+        for (start_version, end_version) in invalid_ranges {
+            let res = table.table_changes(engine.as_ref(), start_version, end_version);
+            assert!(matches!(res, Err(Error::ChangeDataFeedUnsupported(_))))
+        }
+    }
+    #[test]
+    fn schema_evolution_fails() {
+        let path = "./tests/data/table-with-cdf";
+        let engine = Box::new(SyncEngine::new());
+        let table = Table::try_from_uri(path).unwrap();
+        let expected_msg = "Failed to build TableChanges: Start and end version schemas are different. Found start version schema StructType { type_name: \"struct\", fields: {\"part\": StructField { name: \"part\", data_type: Primitive(Integer), nullable: true, metadata: {} }, \"id\": StructField { name: \"id\", data_type: Primitive(Integer), nullable: true, metadata: {} }} } and end version schema StructType { type_name: \"struct\", fields: {\"part\": StructField { name: \"part\", data_type: Primitive(Integer), nullable: true, metadata: {} }, \"id\": StructField { name: \"id\", data_type: Primitive(Integer), nullable: false, metadata: {} }} }";
+
+        // A field in the schema goes from being nullable to non-nullable
+        let table_changes_res = table.table_changes(engine.as_ref(), 3, 4);
+        assert!(matches!(table_changes_res, Err(Error::Generic(msg)) if msg == expected_msg));
+    }
+
+    #[test]
+    fn table_changes_has_cdf_schema() {
+        let path = "./tests/data/table-with-cdf";
+        let engine = Box::new(SyncEngine::new());
+        let table = Table::try_from_uri(path).unwrap();
+        let expected_schema = [
+            StructField::new("part", DataType::INTEGER, true),
+            StructField::new("id", DataType::INTEGER, true),
+        ]
+        .into_iter()
+        .chain(CDF_FIELDS.clone());
+
+        let table_changes = table.table_changes(engine.as_ref(), 0, 0).unwrap();
+        assert_equal(expected_schema, table_changes.schema().fields().cloned());
+    }
+}

kernel/tests/data/table-with-cdf/_delta_log/00000000000000000000.json

@@ -0,0 +1,4 @@
+{"commitInfo":{"timestamp":1704392842074,"operation":"Manual Update","operationParameters":{},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.0 Delta-Lake/3.1.0-SNAPSHOT","txnId":"95ec924a-6859-4433-8008-6d6b4a0e3ba5"}}
+{"protocol":{"minReaderVersion":3,"minWriterVersion":7, "readerFeatures":[], "writerFeatures":[]}}
+{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"part\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.enableChangeDataFeed": "true"}}}
+{"add":{"path":"fake/path/1","partitionValues":{},"size":1,"modificationTime":1,"dataChange":true}}

kernel/tests/data/table-with-cdf/_delta_log/00000000000000000001.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1704392846030,"operation":"Manual Update","operationParameters":{},"readVersion":0,"isolationLevel":"Serializable","isBlindAppend":false,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.0 Delta-Lake/3.1.0-SNAPSHOT","txnId":"01d40235-c8b4-4f8e-8f19-8c97872217fd"}}
+{"cdc":{"path":"fake/path/2","partitionValues":{"partition_foo":"partition_bar"},"size":1,"tags":{"tag_foo":"tag_bar"},"dataChange":false}}
+{"remove":{"path":"fake/path/1","deletionTimestamp":100,"dataChange":true}}

kernel/tests/data/table-with-cdf/_delta_log/00000000000000000002.json

@@ -0,0 +1,4 @@
+{"commitInfo":{"timestamp":1704392846603,"operation":"Manual Update","operationParameters":{},"readVersion":1,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.0 Delta-Lake/3.1.0-SNAPSHOT","txnId":"6cef7579-ca93-4427-988e-9269e8db50c7"}}
+{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"part\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.enableChangeDataFeed": "false"}}}
+{"remove":{"path":"fake/path/1","deletionTimestamp":1704392846603,"dataChange":true,"extendedFileMetadata":true,"partitionValues":{},"size":635,"tags":{"INSERTION_TIME":"1677811178336000","MIN_INSERTION_TIME":"1677811178336000","MAX_INSERTION_TIME":"1677811178336000","OPTIMIZE_TARGET_SIZE":"268435456"}}}
+{"txn":{"appId":"fakeAppId","version":3,"lastUpdated":200}}

kernel/tests/data/table-with-cdf/_delta_log/00000000000000000003.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1704392846603,"operation":"Manual Update","operationParameters":{},"readVersion":1,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.0 Delta-Lake/3.1.0-SNAPSHOT","txnId":"6cef7579-ca93-4427-988e-9269e8db50c7"}}
+{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"part\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.enableChangeDataFeed": "true"}}}
+{"txn":{"appId":"fakeAppId","version":3,"lastUpdated":200}}

kernel/tests/data/table-with-cdf/_delta_log/00000000000000000004.json

@@ -0,0 +1,3 @@
+{"commitInfo":{"timestamp":1704392846603,"operation":"Manual Update","operationParameters":{},"readVersion":1,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{},"engineInfo":"Apache-Spark/3.5.0 Delta-Lake/3.1.0-SNAPSHOT","txnId":"6cef7579-ca93-4427-988e-9269e8db50c7"}}
+{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"part\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"id\",\"type\":\"integer\",\"nullable\":false,\"metadata\":{}}]}","partitionColumns":[],"configuration":{"delta.enableChangeDataFeed": "true"}}}
+{"txn":{"appId":"fakeAppId","version":3,"lastUpdated":200}}