feat: Allow writes to CDF tables for add-only, remove-only, and non-data-change transactions (#1490)

### What changes are proposed in this pull request?
* This PR removes the overly restrictive check that required appendOnly
mode to be enabled for writes to CDF-enabled tables. Now writers can
perform add-only or remove-only operations on CDF tables without
requiring appendOnly at the table level.
* A more targeted validation is introduced at commit time: mixed
add+remove operations with dataChange=true are blocked (since kernel
doesn't yet support writing CDC files), while add-only and remove-only
operations are allowed. The error message has been improved to explain
why the operation is blocked and suggest using separate transactions.
### How was this change tested?
* Added 4 comprehensive unit tests covering all CDF write scenarios:
* Add-only transactions (allowed)
* Remove-only transactions (allowed)
* Mixed add+remove with dataChange=false (allowed for
optimization/vacuum)
* Mixed add+remove with dataChange=true (blocked with helpful error
message)

---------

Co-authored-by: Zach Schuermann <[email protected]>

Author: OussamaSaoudi

kernel/src/table_configuration.rs

@@ -20,7 +20,6 @@ use crate::table_features::{
     ColumnMappingMode, EnablementCheck, FeatureInfo, FeatureType, TableFeature,
 };
 use crate::table_properties::TableProperties;
-use crate::utils::require;
 use crate::{DeltaResult, Error, Version};
 use delta_kernel_derive::internal_api;
 
@@ -183,27 +182,6 @@ 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()
@@ -686,17 +664,17 @@ mod test {
         use TableFeature::*;
         let cases = [
             (
-                // Should fail since AppendOnly is not supported
+                // Writing to CDF-enabled table is supported for writes
                 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"))
+                Ok(())
             ),
             (
-                // Should fail since AppendOnly is supported but not enabled
+                // Should succeed even if 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"))
+                Ok(())
             ),
                     (
                 // Should succeed since AppendOnly is enabled

kernel/src/transaction/mod.rs

@@ -26,7 +26,7 @@ use crate::scan::log_replay::{
 use crate::scan::scan_row_schema;
 use crate::schema::{ArrayType, MapType, SchemaRef, StructField, StructType};
 use crate::snapshot::SnapshotRef;
-use crate::utils::current_time_ms;
+use crate::utils::{current_time_ms, require};
 use crate::{
     DataType, DeltaResult, Engine, EngineData, Expression, ExpressionRef, IntoEngineData,
     RowVisitor, SchemaTransform, Version,
@@ -232,6 +232,31 @@ impl Transaction {
                 dup.app_id
             )));
         }
+
+        // If there are add and remove files with data change in the same transaction, we block it.
+        // This is because kernel does not yet have a way to discern DML operations. For DML
+        // operations that perform updates on rows, ChangeDataFeed requires that a `cdc` file be
+        // written to the delta log.
+        if !self.add_files_metadata.is_empty()
+            && !self.remove_files_metadata.is_empty()
+            && self.data_change
+        {
+            let cdf_enabled = self
+                .read_snapshot
+                .table_configuration()
+                .table_properties()
+                .enable_change_data_feed
+                .unwrap_or(false);
+            require!(
+                !cdf_enabled,
+                Error::generic(
+                    "Cannot add and remove data in the same transaction when Change Data Feed is enabled (delta.enableChangeDataFeed = true). \
+                     This would require writing CDC files for DML operations, which is not yet supported. \
+                     Consider using separate transactions: one to add files, another to remove files."
+                )
+            );
+        }
+
         // Step 1: Generate SetTransaction actions
         let set_transaction_actions = self
             .set_transactions

kernel/tests/write.rs

@@ -2208,3 +2208,225 @@ async fn test_remove_files_with_modified_selection_vector() -> Result<(), Box<dy
     }
     Ok(())
 }
+
+// Helper function to create a table with CDF enabled
+async fn create_cdf_table(
+    table_name: &str,
+    schema: SchemaRef,
+) -> Result<(Url, Arc<DefaultEngine<TokioBackgroundExecutor>>, TempDir), Box<dyn std::error::Error>>
+{
+    let tmp_dir = tempdir()?;
+    let tmp_test_dir_url = Url::from_directory_path(tmp_dir.path()).unwrap();
+
+    let (store, engine, table_location) = engine_store_setup(table_name, Some(&tmp_test_dir_url));
+
+    let table_url = create_table(
+        store.clone(),
+        table_location,
+        schema.clone(),
+        &[],
+        true, // use protocol 3.7
+        vec![],
+        vec!["changeDataFeed"],
+    )
+    .await?;
+
+    Ok((table_url, Arc::new(engine), tmp_dir))
+}
+
+// Helper function to write data to a table
+async fn write_data_to_table(
+    table_url: &Url,
+    engine: &Arc<DefaultEngine<TokioBackgroundExecutor>>,
+    schema: SchemaRef,
+    values: Vec<i32>,
+) -> Result<Version, Box<dyn std::error::Error>> {
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
+    let mut txn = snapshot
+        .transaction(Box::new(FileSystemCommitter::new()))?
+        .with_engine_info("test");
+
+    add_files_to_transaction(&mut txn, engine, schema, values).await?;
+
+    let result = txn.commit(engine.as_ref())?;
+    match result {
+        CommitResult::CommittedTransaction(committed) => Ok(committed.commit_version()),
+        _ => panic!("Transaction should be committed"),
+    }
+}
+
+// Helper function to add files to an existing transaction
+async fn add_files_to_transaction(
+    txn: &mut delta_kernel::transaction::Transaction,
+    engine: &Arc<DefaultEngine<TokioBackgroundExecutor>>,
+    schema: SchemaRef,
+    values: Vec<i32>,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let data = RecordBatch::try_new(
+        Arc::new(schema.as_ref().try_into_arrow()?),
+        vec![Arc::new(Int32Array::from(values))],
+    )?;
+
+    let write_context = Arc::new(txn.get_write_context());
+    let add_files_metadata = engine
+        .write_parquet(
+            &ArrowEngineData::new(data),
+            write_context.as_ref(),
+            HashMap::new(),
+        )
+        .await?;
+    txn.add_files(add_files_metadata);
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_cdf_write_all_adds_succeeds() -> Result<(), Box<dyn std::error::Error>> {
+    // This test verifies that add-only transactions work with CDF enabled
+    let _ = tracing_subscriber::fmt::try_init();
+
+    let schema = Arc::new(StructType::try_new(vec![StructField::nullable(
+        "number",
+        DataType::INTEGER,
+    )])?);
+
+    let (table_url, engine, _tmp_dir) =
+        create_cdf_table("test_cdf_all_adds", schema.clone()).await?;
+
+    // Add files - this should succeed
+    let version = write_data_to_table(&table_url, &engine, schema, vec![1, 2, 3]).await?;
+    assert_eq!(version, 1);
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_cdf_write_all_removes_succeeds() -> Result<(), Box<dyn std::error::Error>> {
+    // This test verifies that remove-only transactions work with CDF enabled
+    let _ = tracing_subscriber::fmt::try_init();
+
+    let schema = Arc::new(StructType::try_new(vec![StructField::nullable(
+        "number",
+        DataType::INTEGER,
+    )])?);
+
+    let (table_url, engine, _tmp_dir) =
+        create_cdf_table("test_cdf_all_removes", schema.clone()).await?;
+
+    // First, add some data
+    write_data_to_table(&table_url, &engine, schema, vec![1, 2, 3]).await?;
+
+    // Now remove the files
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
+    let mut txn = snapshot
+        .clone()
+        .transaction(Box::new(FileSystemCommitter::new()))?
+        .with_engine_info("cdf remove test")
+        .with_data_change(true);
+
+    let scan = snapshot.scan_builder().build()?;
+    let scan_metadata = scan.scan_metadata(engine.as_ref())?.next().unwrap()?;
+    let (data, selection_vector) = scan_metadata.scan_files.into_parts();
+    txn.remove_files(FilteredEngineData::try_new(data, selection_vector)?);
+
+    // This should succeed - remove-only transactions are allowed with CDF
+    let result = txn.commit(engine.as_ref())?;
+    match result {
+        CommitResult::CommittedTransaction(committed) => {
+            assert_eq!(committed.commit_version(), 2);
+        }
+        _ => panic!("Transaction should be committed"),
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_cdf_write_mixed_no_data_change_succeeds() -> Result<(), Box<dyn std::error::Error>> {
+    // This test verifies that mixed add+remove transactions work when dataChange=false.
+    // It's allowed because the transaction does not contain any logical data changes.
+    // This can happen when a table is being optimized/compacted.
+    let _ = tracing_subscriber::fmt::try_init();
+
+    let schema = Arc::new(StructType::try_new(vec![StructField::nullable(
+        "number",
+        DataType::INTEGER,
+    )])?);
+
+    let (table_url, engine, _tmp_dir) =
+        create_cdf_table("test_cdf_mixed_no_data_change", schema.clone()).await?;
+
+    // First, add some data
+    write_data_to_table(&table_url, &engine, schema.clone(), vec![1, 2, 3]).await?;
+
+    // Now create a transaction with both add AND remove files, but dataChange=false
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
+    let mut txn = snapshot
+        .clone()
+        .transaction(Box::new(FileSystemCommitter::new()))?
+        .with_engine_info("cdf mixed test")
+        .with_data_change(false); // dataChange=false is key here
+
+    // Add new files
+    add_files_to_transaction(&mut txn, &engine, schema, vec![4, 5, 6]).await?;
+
+    // Also remove existing files
+    let scan = snapshot.scan_builder().build()?;
+    let scan_metadata = scan.scan_metadata(engine.as_ref())?.next().unwrap()?;
+    let (data, selection_vector) = scan_metadata.scan_files.into_parts();
+    txn.remove_files(FilteredEngineData::try_new(data, selection_vector)?);
+
+    // This should succeed - mixed operations are allowed when dataChange=false
+    let result = txn.commit(engine.as_ref())?;
+    match result {
+        CommitResult::CommittedTransaction(committed) => {
+            assert_eq!(committed.commit_version(), 2);
+        }
+        _ => panic!("Transaction should be committed"),
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_cdf_write_mixed_with_data_change_fails() -> Result<(), Box<dyn std::error::Error>> {
+    // This test verifies that mixed add+remove transactions fail with helpful error when dataChange=true
+    let _ = tracing_subscriber::fmt::try_init();
+
+    let schema = Arc::new(StructType::try_new(vec![StructField::nullable(
+        "number",
+        DataType::INTEGER,
+    )])?);
+
+    let (table_url, engine, _tmp_dir) =
+        create_cdf_table("test_cdf_mixed_with_data_change", schema.clone()).await?;
+
+    // First, add some data
+    write_data_to_table(&table_url, &engine, schema.clone(), vec![1, 2, 3]).await?;
+
+    // Now create a transaction with both add AND remove files with dataChange=true
+    let snapshot = Snapshot::builder_for(table_url.clone()).build(engine.as_ref())?;
+    let mut txn = snapshot
+        .clone()
+        .transaction(Box::new(FileSystemCommitter::new()))?
+        .with_engine_info("cdf mixed fail test")
+        .with_data_change(true); // dataChange=true - this should fail
+
+    // Add new files
+    add_files_to_transaction(&mut txn, &engine, schema, vec![4, 5, 6]).await?;
+
+    // Also remove existing files
+    let scan = snapshot.scan_builder().build()?;
+    let scan_metadata = scan.scan_metadata(engine.as_ref())?.next().unwrap()?;
+    let (data, selection_vector) = scan_metadata.scan_files.into_parts();
+    txn.remove_files(FilteredEngineData::try_new(data, selection_vector)?);
+
+    // This should fail with our new error message
+    assert_result_error_with_message(
+        txn.commit(engine.as_ref()),
+        "Cannot add and remove data in the same transaction when Change Data Feed is enabled (delta.enableChangeDataFeed = true). \
+         This would require writing CDC files for DML operations, which is not yet supported. \
+         Consider using separate transactions: one to add files, another to remove files."
+    );
+
+    Ok(())
+}

test-utils/src/lib.rs

@@ -325,6 +325,9 @@ pub async fn create_table(
                 json!("1612345678"),
             );
         }
+        if writer_features.contains(&"changeDataFeed") {
+            config.insert("delta.enableChangeDataFeed".to_string(), json!("true"));
+        }
 
         config
     };