fix

Author: DrakeLin

kernel/src/table_changes/log_replay.rs

@@ -21,6 +21,7 @@ use crate::schema::{
 use crate::table_changes::scan_file::{cdf_scan_row_expression, cdf_scan_row_schema};
 use crate::table_configuration::TableConfiguration;
 use crate::table_features::Operation;
+use crate::table_features::TableFeature;
 use crate::utils::require;
 use crate::{DeltaResult, Engine, EngineData, Error, PredicateRef, RowVisitor};
 
@@ -186,7 +187,9 @@ impl LogReplayScanner {
             visitor.visit_rows_of(actions.as_ref())?;
 
             let metadata_opt = Metadata::try_new_from_data(actions.as_ref())?;
+            let has_metadata_update = metadata_opt.is_some();
             let protocol_opt = Protocol::try_new_from_data(actions.as_ref())?;
+            let has_protocol_update = protocol_opt.is_some();
 
             if let Some(ref metadata) = metadata_opt {
                 let schema = metadata.parse_schema()?;
@@ -200,15 +203,25 @@ impl LogReplayScanner {
             }
 
             // Update table configuration with any new Protocol or Metadata from this commit
-            if metadata_opt.is_some() || protocol_opt.is_some() {
+            if has_metadata_update || has_protocol_update {
                 *table_configuration = TableConfiguration::try_new_from(
                     table_configuration,
                     metadata_opt,
                     protocol_opt,
                     commit_file.version,
                 )?;
+            }
+
+            // If metadata is updated, check if Change Data Feed is supported
+            if has_metadata_update {
+                require!(
+                    table_configuration.is_feature_enabled(&TableFeature::ChangeDataFeed),
+                    Error::change_data_feed_unsupported(commit_file.version)
+                );
+            }
 
-                // If protocol or metadata is updated, check if Change Data Feed is supported
+            // If protocol is updated, check if Change Data Feed is supported
+            if has_protocol_update {
                 table_configuration
                     .ensure_operation_supported(Operation::Cdf)
                     .map_err(|_| Error::change_data_feed_unsupported(commit_file.version))?;

kernel/src/table_changes/log_replay/tests.rs

@@ -41,7 +41,8 @@ fn get_default_table_config(table_root: &url::Url) -> TableConfiguration {
         ]),
     )
     .unwrap();
-    let protocol = Protocol::try_new(1, 1, None::<Vec<String>>, None::<Vec<String>>).unwrap();
+    // CDF requires min_writer_version = 4
+    let protocol = Protocol::try_new(1, 4, None::<Vec<String>>, None::<Vec<String>>).unwrap();
     TableConfiguration::try_new(metadata, protocol, table_root.clone(), 0).unwrap()
 }
 
@@ -329,7 +330,7 @@ async fn unsupported_protocol_feature_midstream() {
     // First commit: Basic protocol with CDF enabled
     mock_table
         .commit([
-            protocol_action(1, 1, None, None),
+            protocol_action(2, 6, None, None),
             metadata_with_cdf(get_schema()),
         ])
         .await;