feat: enable CDF for column-mapped tables (#1510)

## What changes are proposed in this pull request?
enable CDF (previously blocked due to some table configuration checks)
on column-mapped tables. After migrating to use transforms in CDF
implementation, all the pieces are there - just need to ungate + test
it.

additionally, adds ability to read CDF for the following table features:
`V2Checkpoint`, `VacuumProtocolCheck`, and `TimestampWithoutTimezones`


## How was this change tested?
new integration tests

Author: zachschuermann

kernel/src/actions/mod.rs

@@ -559,6 +559,33 @@ impl Protocol {
         self.has_table_feature(&TableFeature::CatalogManaged)
             || self.has_table_feature(&TableFeature::CatalogOwnedPreview)
     }
+
+    pub(crate) fn is_cdf_supported(&self) -> bool {
+        // TODO: we should probably expand this to ~all supported reader features instead of gating
+        // on a subset here. missing ones are:
+        // - variantType
+        // - typeWidening
+        // - catalogManaged
+        static CDF_SUPPORTED_READER_FEATURES: LazyLock<Vec<TableFeature>> = LazyLock::new(|| {
+            vec![
+                TableFeature::DeletionVectors,
+                TableFeature::ColumnMapping,
+                TableFeature::TimestampWithoutTimezone,
+                TableFeature::V2Checkpoint,
+                TableFeature::VacuumProtocolCheck,
+            ]
+        });
+        match self.reader_features() {
+            // if min_reader_version = 3 and all reader features are subset of supported => OK
+            Some(reader_features) if self.min_reader_version() == 3 => {
+                ensure_supported_features(reader_features, &CDF_SUPPORTED_READER_FEATURES).is_ok()
+            }
+            // if min_reader_version = 1 or 2 and there are no reader features => OK
+            None => (1..=2).contains(&self.min_reader_version()),
+            // any other protocol is not supported
+            _ => false,
+        }
+    }
 }
 
 // TODO: implement Scalar::From<HashMap<K, V>> so we can derive IntoEngineData using a macro (issue#1083)

kernel/src/table_changes/log_replay.rs

@@ -19,8 +19,8 @@ use crate::schema::{
     ArrayType, ColumnNamesAndTypes, DataType, MapType, SchemaRef, StructField, StructType,
     ToSchema as _,
 };
+use crate::table_changes::check_cdf_table_properties;
 use crate::table_changes::scan_file::{cdf_scan_row_expression, cdf_scan_row_schema};
-use crate::table_changes::{check_cdf_table_properties, ensure_cdf_read_supported};
 use crate::table_properties::TableProperties;
 use crate::utils::require;
 use crate::{DeltaResult, Engine, EngineData, Error, PredicateRef, RowVisitor};
@@ -180,8 +180,10 @@ impl LogReplayScanner {
             visitor.visit_rows_of(actions.as_ref())?;
 
             if let Some(protocol) = visitor.protocol {
-                ensure_cdf_read_supported(&protocol)
-                    .map_err(|_| Error::change_data_feed_unsupported(commit_file.version))?;
+                protocol
+                    .is_cdf_supported()
+                    .then_some(())
+                    .ok_or_else(|| Error::change_data_feed_unsupported(commit_file.version))?;
             }
             if let Some((schema, configuration)) = visitor.metadata_info {
                 let schema: StructType = serde_json::from_str(&schema)?;

kernel/src/table_changes/log_replay/tests.rs

@@ -138,8 +138,8 @@ async fn unsupported_reader_feature() {
             Protocol::try_new(
                 3,
                 7,
-                Some([TableFeature::DeletionVectors, TableFeature::ColumnMapping]),
-                Some([TableFeature::DeletionVectors, TableFeature::ColumnMapping]),
+                Some([TableFeature::DeletionVectors, TableFeature::TypeWidening]),
+                Some([TableFeature::DeletionVectors, TableFeature::TypeWidening]),
             )
             .unwrap(),
         )])
@@ -157,7 +157,7 @@ async fn unsupported_reader_feature() {
     assert!(matches!(res, Err(Error::ChangeDataFeedUnsupported(_))));
 }
 #[tokio::test]
-async fn column_mapping_should_fail() {
+async fn column_mapping_should_succeed() {
     let engine = Arc::new(SyncEngine::new());
     let mut mock_table = LocalMockTable::new();
     mock_table
@@ -190,7 +190,8 @@ async fn column_mapping_should_fail() {
             .unwrap()
             .try_collect();
 
-    assert!(matches!(res, Err(Error::ChangeDataFeedUnsupported(_))));
+    // Column mapping with CDF should now succeed
+    assert!(res.is_ok(), "CDF should now support column mapping");
 }
 
 // Note: This should be removed once type widening support is added for CDF

kernel/src/table_changes/mod.rs

@@ -36,12 +36,10 @@ use std::sync::{Arc, LazyLock};
 use scan::TableChangesScanBuilder;
 use url::Url;
 
-use crate::actions::{ensure_supported_features, Protocol};
 use crate::log_segment::LogSegment;
 use crate::path::AsUrl;
 use crate::schema::{DataType, Schema, StructField, StructType};
 use crate::snapshot::{Snapshot, SnapshotRef};
-use crate::table_features::{ColumnMappingMode, TableFeature};
 use crate::table_properties::TableProperties;
 use crate::utils::require;
 use crate::{DeltaResult, Engine, Error, Version};
@@ -253,40 +251,15 @@ impl TableChanges {
 
 /// Ensures that change data feed is enabled in `table_properties`. See the documentation
 /// of [`TableChanges`] for more details.
+// TODO: move to TableProperties and normalize with the check in TableConfiguration
 fn check_cdf_table_properties(table_properties: &TableProperties) -> DeltaResult<()> {
     require!(
         table_properties.enable_change_data_feed.unwrap_or(false),
         Error::unsupported("Change data feed is not enabled")
     );
-    require!(
-        matches!(
-            table_properties.column_mapping_mode,
-            None | Some(ColumnMappingMode::None)
-        ),
-        Error::unsupported("Change data feed not supported when column mapping is enabled")
-    );
     Ok(())
 }
 
-/// Ensures that Change Data Feed is supported for a table with this [`Protocol`] .
-/// See the documentation of [`TableChanges`] for more details.
-fn ensure_cdf_read_supported(protocol: &Protocol) -> DeltaResult<()> {
-    static CDF_SUPPORTED_READER_FEATURES: LazyLock<Vec<TableFeature>> =
-        LazyLock::new(|| vec![TableFeature::DeletionVectors]);
-    match &protocol.reader_features() {
-        // if min_reader_version = 3 and all reader features are subset of supported => OK
-        Some(reader_features) if protocol.min_reader_version() == 3 => {
-            ensure_supported_features(reader_features, &CDF_SUPPORTED_READER_FEATURES)
-        }
-        // if min_reader_version = 1 and there are no reader features => OK
-        None if protocol.min_reader_version() == 1 => Ok(()),
-        // any other protocol is not supported
-        _ => Err(Error::unsupported(
-            "Change data feed not supported on this protocol",
-        )),
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

kernel/src/table_configuration.rs

@@ -8,11 +8,11 @@
 //! [`TableProperties`].
 //!
 //! [`Schema`]: crate::schema::Schema
-use std::sync::{Arc, LazyLock};
+use std::sync::Arc;
 
 use url::Url;
 
-use crate::actions::{ensure_supported_features, Metadata, Protocol};
+use crate::actions::{Metadata, Protocol};
 use crate::schema::variant_utils::validate_variant_type_feature_support;
 use crate::schema::{InvariantChecker, SchemaRef};
 use crate::table_features::{
@@ -400,27 +400,12 @@ impl TableConfiguration {
     /// [`TableChanges`]: crate::table_changes::TableChanges
     #[internal_api]
     pub(crate) fn is_cdf_read_supported(&self) -> bool {
-        static CDF_SUPPORTED_READER_FEATURES: LazyLock<Vec<TableFeature>> =
-            LazyLock::new(|| vec![TableFeature::DeletionVectors]);
-        let protocol_supported = match self.protocol.reader_features() {
-            // if min_reader_version = 3 and all reader features are subset of supported => OK
-            Some(reader_features) if self.protocol.min_reader_version() == 3 => {
-                ensure_supported_features(reader_features, &CDF_SUPPORTED_READER_FEATURES).is_ok()
-            }
-            // if min_reader_version = 1 and there are no reader features => OK
-            None => self.protocol.min_reader_version() == 1,
-            // any other protocol is not supported
-            _ => false,
-        };
+        let protocol_supported = self.protocol.is_cdf_supported();
         let cdf_enabled = self
             .table_properties
             .enable_change_data_feed
             .unwrap_or(false);
-        let column_mapping_disabled = matches!(
-            self.table_properties.column_mapping_mode,
-            None | Some(ColumnMappingMode::None)
-        );
-        protocol_supported && cdf_enabled && column_mapping_disabled
+        protocol_supported && cdf_enabled
     }
 
     /// Returns `true` if deletion vectors is supported on this table. To support deletion vectors,

kernel/tests/cdf.rs

@@ -545,3 +545,63 @@ fn conditional_delete_two_rows() -> DeltaResult<()> {
     assert_batches_sorted_eq!(expected, &batches);
     Ok(())
 }
+
+#[test]
+fn cdf_with_column_mapping_name_mode() -> Result<(), Box<dyn error::Error>> {
+    // NOTE: these tables only have CDF enabled in version 1+, so we start reading from 1. This is
+    // due to pyspark limitation while writing: we were unable to create a table with column
+    // mapping + CDF enabled in commit 0, so we created with column mapping and enabled CDF in
+    // commit 1.
+    let batches = read_cdf_for_table("cdf-column-mapping-name-mode", 1, None, None)?;
+    let mut expected = vec![
+        "+----+-------+-------+------------------+-----------------+",
+        "| id | name  | value | _change_type     | _commit_version |",
+        "+----+-------+-------+------------------+-----------------+",
+        "| 1  | Alice | 100.0 | delete           | 4               |",
+        "| 2  | Bob   | 200.0 | update_preimage  | 2               |",
+        "| 2  | Bob   | 250.0 | update_postimage | 2               |",
+        "| 4  | David | 400.0 | insert           | 3               |",
+        "+----+-------+-------+------------------+-----------------+",
+    ];
+    sort_lines!(expected);
+    assert_batches_sorted_eq!(expected, &batches);
+
+    // same as above but instead of protocol 2,5 this is 3,7 with columnMapping+DV features
+    let batches = read_cdf_for_table("cdf-column-mapping-name-mode-3-7", 1, None, None)?;
+    let mut expected = vec![
+        "+----+-------+-------+------------------+-----------------+",
+        "| id | name  | value | _change_type     | _commit_version |",
+        "+----+-------+-------+------------------+-----------------+",
+        "| 1  | Alice | 100.0 | delete           | 4               |",
+        "| 2  | Bob   | 200.0 | update_preimage  | 2               |",
+        "| 2  | Bob   | 250.0 | update_postimage | 2               |",
+        "| 4  | David | 400.0 | insert           | 3               |",
+        "+----+-------+-------+------------------+-----------------+",
+    ];
+    sort_lines!(expected);
+    assert_batches_sorted_eq!(expected, &batches);
+
+    Ok(())
+}
+
+#[test]
+fn cdf_with_column_mapping_id_mode() -> Result<(), Box<dyn error::Error>> {
+    // NOTE: these tables only have CDF enabled in version 1+, so we start reading from 1. This is
+    // due to pyspark limitation while writing: we were unable to create a table with column
+    // mapping + CDF enabled in commit 0, so we created with column mapping and enabled CDF in
+    // commit 1.
+    let batches = read_cdf_for_table("cdf-column-mapping-id-mode", 1, None, None)?;
+    let mut expected = vec![
+        "+----+-------+-------+------------------+-----------------+",
+        "| id | name  | value | _change_type     | _commit_version |",
+        "+----+-------+-------+------------------+-----------------+",
+        "| 2  | Frank | 250.0 | update_preimage  | 2               |",
+        "| 2  | Frank | 275.0 | update_postimage | 2               |",
+        "| 3  | Grace | 350.0 | delete           | 4               |",
+        "| 4  | Henry | 450.0 | insert           | 3               |",
+        "+----+-------+-------+------------------+-----------------+",
+    ];
+    sort_lines!(expected);
+    assert_batches_sorted_eq!(expected, &batches);
+    Ok(())
+}