feat: enable writes to CDF enabled tables only if append only is supported (#1449)

## What changes are proposed in this pull request?
This PR enables writes to tables that have CDF enabled only if append
only is also enabled.

Change Data Feed writes require cdc files to be written to the delta
table for DML operations. However, delta-kernel-rs does not support
writing cdc files. However, Change Data Feed appends do not require
writing cdc files. Thus we enable CDF for tables where append only is
enabled. Such tables are safe to write to because there will never be a
DML operation that requires a cdc write.

## How was this change tested?
Verify that table configuration fails on incorrect/invalid
configurations.

Author: OussamaSaoudi

kernel/src/actions/mod.rs

@@ -693,7 +693,7 @@ where
         .filter(|feature| !supported_features.contains(*feature));
 
     Err(Error::Unsupported(format!(
-        "Unknown {}s: \"{}\". Supported {}s: \"{}\"",
+        "Found unsupported {}s: \"{}\". Supported {}s: \"{}\"",
         features_type,
         unsupported.join("\", \""),
         features_type,
@@ -1595,7 +1595,7 @@ mod tests {
         .unwrap();
         assert_result_error_with_message(
             protocol.ensure_write_supported(),
-            r#"Unsupported: Unknown TableFeatures: "identityColumns". Supported TableFeatures: "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#,
+            r#"Unsupported: Found unsupported TableFeatures: "identityColumns". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#,
         );
 
         // Unknown writer features are allowed during creation for forward compatibility,
@@ -1609,7 +1609,7 @@ mod tests {
         .unwrap();
         assert_result_error_with_message(
             protocol.ensure_write_supported(),
-            r#"Unsupported: Unknown TableFeatures: "unsupported writer". Supported TableFeatures: "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#,
+            r#"Unsupported: Found unsupported TableFeatures: "unsupported writer". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#,
         );
     }
 
@@ -1643,7 +1643,7 @@ mod tests {
         let error = ensure_supported_features(&table_features, &supported_features).unwrap_err();
         match error {
             Error::Unsupported(e) if e ==
-                "Unknown TableFeatures: \"idk\". Supported TableFeatures: \"columnMapping\", \"deletionVectors\""
+                "Found unsupported TableFeatures: \"idk\". Supported TableFeatures: \"columnMapping\", \"deletionVectors\""
             => {},
             _ => panic!("Expected unsupported error, got: {error}"),
         }

kernel/src/table_configuration.rs

@@ -20,6 +20,7 @@ use crate::table_features::{
     ColumnMappingMode, TableFeature,
 };
 use crate::table_properties::TableProperties;
+use crate::utils::require;
 use crate::{DeltaResult, Error, Version};
 use delta_kernel_derive::internal_api;
 
@@ -182,6 +183,27 @@ impl TableConfiguration {
     pub(crate) fn ensure_write_supported(&self) -> DeltaResult<()> {
         self.protocol.ensure_write_supported()?;
 
+        // We allow Change Data Feed to be enabled only if AppendOnly is enabled.
+        // This is because kernel does not yet support writing `.cdc` files for DML operations.
+        if self
+            .table_properties()
+            .enable_change_data_feed
+            .unwrap_or(false)
+        {
+            require!(
+                self.is_append_only_enabled(),
+                Error::unsupported("Writing to table with Change Data Feed is only supported if append only mode is enabled")
+            );
+            require!(
+                self.is_cdf_read_supported(),
+                Error::unsupported(
+                    "Change data feed is enabled on this table, but found invalid table
+                    table_configuration. Ensure that column mapping is disabled and ensure correct
+                    protocol reader/writer features"
+                )
+            );
+        }
+
         // for now we don't allow invariants so although we support writer version 2 and the
         // ColumnInvariant TableFeature we _must_ check here that they are not actually in use
         if self.is_invariants_supported()
@@ -423,13 +445,65 @@ mod test {
 
     use crate::actions::{Metadata, Protocol};
     use crate::schema::{DataType, StructField, StructType};
-    use crate::table_features::TableFeature;
+    use crate::table_features::{FeatureType, TableFeature};
     use crate::table_properties::TableProperties;
     use crate::utils::test_utils::assert_result_error_with_message;
     use crate::Error;
 
     use super::{InCommitTimestampEnablement, TableConfiguration};
 
+    fn create_mock_table_config(
+        props_to_enable: &[&str],
+        features: &[TableFeature],
+    ) -> TableConfiguration {
+        create_mock_table_config_with_version(props_to_enable, features.into(), 3, 7)
+    }
+
+    fn create_mock_table_config_with_version(
+        props_to_enable: &[&str],
+        features_opt: Option<&[TableFeature]>,
+        min_reader_version: i32,
+        min_writer_version: i32,
+    ) -> TableConfiguration {
+        let schema = StructType::new_unchecked([StructField::nullable("value", DataType::INTEGER)]);
+        let metadata = Metadata::try_new(
+            None,
+            None,
+            schema,
+            vec![],
+            0,
+            HashMap::from_iter(
+                props_to_enable
+                    .iter()
+                    .map(|key| (key.to_string(), "true".to_string())),
+            ),
+        )
+        .unwrap();
+        let reader_features = features_opt.map(|features| {
+            features
+                .iter()
+                .filter(|feature| matches!(feature.feature_type(), FeatureType::ReaderWriter))
+        });
+        let writer_features = features_opt.map(|features| {
+            features.iter().filter(|feature| {
+                matches!(
+                    feature.feature_type(),
+                    FeatureType::Writer | FeatureType::ReaderWriter
+                )
+            })
+        });
+
+        let protocol = Protocol::try_new(
+            min_reader_version,
+            min_writer_version,
+            reader_features,
+            writer_features,
+        )
+        .unwrap();
+        let table_root = Url::try_from("file:///").unwrap();
+        TableConfiguration::try_new(metadata, protocol, table_root, 0).unwrap()
+    }
+
     #[test]
     fn dv_supported_not_enabled() {
         let schema = StructType::new_unchecked([StructField::nullable("value", DataType::INTEGER)]);
@@ -454,6 +528,7 @@ mod test {
         assert!(table_config.is_deletion_vector_supported());
         assert!(!table_config.is_deletion_vector_enabled());
     }
+
     #[test]
     fn dv_enabled() {
         let schema = StructType::new_unchecked([StructField::nullable("value", DataType::INTEGER)]);
@@ -484,6 +559,79 @@ mod test {
         assert!(table_config.is_deletion_vector_supported());
         assert!(table_config.is_deletion_vector_enabled());
     }
+
+    #[test]
+    fn write_with_cdf() {
+        use TableFeature::*;
+        let cases = [
+            (
+                // Should fail since AppendOnly is not supported
+                create_mock_table_config(&["delta.enableChangeDataFeed"], &[ChangeDataFeed]),
+                Err(Error::unsupported("Writing to table with Change Data Feed is only supported if append only mode is enabled"))
+            ),
+            (
+                // Should fail since AppendOnly is supported but not enabled
+                create_mock_table_config(
+                    &["delta.enableChangeDataFeed"],
+                    &[ChangeDataFeed, AppendOnly],
+                ),
+                Err(Error::unsupported("Writing to table with Change Data Feed is only supported if append only mode is enabled"))
+            ),
+                    (
+                // Should succeed since AppendOnly is enabled
+                create_mock_table_config(
+                    &["delta.delta.enableChangeDataFeed", "delta.appendOnly"],
+                    &[ChangeDataFeed, AppendOnly],
+                ),
+                Ok(()),
+            ),
+
+            (
+                // Fails since writes are not supported on min_writer_version=4. Once version 4 is
+                // supported, ensure that this still fails since ChangeDataFeed is enabled while
+                // append only is not enabled.
+                create_mock_table_config_with_version(&["delta.enableChangeDataFeed"],None, 1, 4),
+                Err(Error::unsupported("Currently delta-kernel-rs can only write to tables with protocol.minWriterVersion = 1, 2, or 7"))
+
+            ),
+            // NOTE: The following cases should be updated if column mapping for writes is
+            // supported before cdc is.
+            (
+                // Should fail since change data feed and column mapping features cannot both be
+                // present.
+                create_mock_table_config(
+                    &["delta.enableChangeDataFeed", "delta.appendOnly"],
+                    &[ChangeDataFeed, ColumnMapping, AppendOnly],
+                ),
+                Err(Error::unsupported(r#"Found unsupported TableFeatures: "columnMapping". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#)),
+            ),
+            (
+                // The table does not require writing CDC files, so it is safe to write to it.
+                create_mock_table_config(
+                    &["delta.appendOnly"],
+                    &[ChangeDataFeed, ColumnMapping, AppendOnly],
+                ),
+                Err(Error::unsupported(r#"Found unsupported TableFeatures: "columnMapping". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#)),
+            ),
+            (
+                // Should succeed since change data feed is not enabled
+                create_mock_table_config(&["delta.appendOnly"], &[AppendOnly]),
+                Ok(()),
+            ),
+        ];
+
+        for (table_configuration, result) in cases {
+            match (table_configuration.ensure_write_supported(), result) {
+                (Ok(()), Ok(())) => { /* Correct result */ }
+                (actual_result, Err(expected)) => {
+                    assert_result_error_with_message(actual_result, &expected.to_string());
+                }
+                (Err(actual_result), Ok(())) => {
+                    panic!("Expected Ok but got error: {actual_result}");
+                }
+            }
+        }
+    }
     #[test]
     fn ict_enabled_from_table_creation() {
         let schema = StructType::new_unchecked([StructField::nullable("value", DataType::INTEGER)]);

kernel/src/table_features/mod.rs

@@ -207,6 +207,7 @@ pub(crate) static SUPPORTED_READER_FEATURES: LazyLock<Vec<TableFeature>> = LazyL
 ///   tables with row tracking yet.
 pub(crate) static SUPPORTED_WRITER_FEATURES: LazyLock<Vec<TableFeature>> = LazyLock::new(|| {
     vec![
+        TableFeature::ChangeDataFeed,
         TableFeature::AppendOnly,
         TableFeature::DeletionVectors,
         TableFeature::DomainMetadata,

kernel/src/utils.rs

@@ -273,7 +273,7 @@ pub(crate) mod test_utils {
         message: &str,
     ) {
         match res {
-            Ok(_) => panic!("Expected error, but got Ok result"),
+            Ok(_) => panic!("Expected error with message {message}, but got Ok result"),
             Err(error) => {
                 let error_str = error.to_string();
                 assert!(