fix

Author: DrakeLin

kernel/src/actions/mod.rs

@@ -2,8 +2,7 @@
 //! specification](https://github.com/delta-io/delta/blob/master/PROTOCOL.md)
 
 use std::collections::HashMap;
-use std::fmt::{Debug, Display};
-use std::hash::Hash;
+use std::fmt::Debug;
 use std::str::FromStr;
 use std::sync::{Arc, LazyLock};
 
@@ -24,7 +23,6 @@ use url::Url;
 use visitors::{MetadataVisitor, ProtocolVisitor};
 
 use delta_kernel_derive::{internal_api, IntoEngineData, ToSchema};
-use itertools::Itertools;
 use serde::{Deserialize, Serialize};
 
 const KERNEL_VERSION: &str = env!("CARGO_PKG_VERSION");

kernel/src/table_changes/log_replay.rs

@@ -18,9 +18,9 @@ use crate::scan::state::DvInfo;
 use crate::schema::{
     ColumnNamesAndTypes, DataType, 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_configuration::TableConfiguration;
+use crate::table_features::Operation;
 use crate::utils::require;
 use crate::{DeltaResult, Engine, EngineData, Error, PredicateRef, RowVisitor};
 
@@ -197,9 +197,6 @@ impl LogReplayScanner {
                     table_schema.as_ref() == &schema,
                     Error::change_data_feed_incompatible_schema(table_schema, &schema)
                 );
-                let table_properties = metadata.parse_table_properties();
-                check_cdf_table_properties(&table_properties)
-                    .map_err(|_| Error::change_data_feed_unsupported(commit_file.version))?;
             }
 
             // Update table configuration with any new Protocol or Metadata from this commit
@@ -214,8 +211,7 @@ impl LogReplayScanner {
                 // If protocol or metadata is updated, check if Change Data Feed is supported
                 table_configuration
                     .ensure_operation_supported(Operation::Cdf)
-                    .then_some(())
-                    .ok_or_else(|| Error::change_data_feed_unsupported(commit_file.version))?;
+                    .map_err(|_| Error::change_data_feed_unsupported(commit_file.version))?;
             }
         }
         // We resolve the remove deletion vector map after visiting the entire commit.

kernel/src/table_changes/log_replay/tests.rs

@@ -230,8 +230,14 @@ async fn unsupported_reader_feature() {
             Protocol::try_new(
                 3,
                 7,
-                Some([TableFeature::DeletionVectors, TableFeature::TypeWidening]),
-                Some([TableFeature::DeletionVectors, TableFeature::TypeWidening]),
+                Some([
+                    TableFeature::DeletionVectors,
+                    TableFeature::unknown("unsupportedReaderFeature"),
+                ]),
+                Some([
+                    TableFeature::DeletionVectors,
+                    TableFeature::unknown("unsupportedReaderFeature"),
+                ]),
             )
             .unwrap(),
         )])
@@ -328,20 +334,19 @@ async fn unsupported_protocol_feature_midstream() {
         ])
         .await;
 
-    // Second commit: Protocol update with unsupported feature (TypeWidening)
+    // Second commit: Protocol update with unsupported feature
     mock_table
         .commit([protocol_action(
             3,
             7,
-            Some(vec![TableFeature::TypeWidening]),
-            Some(vec![TableFeature::TypeWidening]),
+            Some(vec![TableFeature::unknown("unsupportedFeature")]),
+            Some(vec![TableFeature::unknown("unsupportedFeature")]),
         )])
         .await;
 
     assert_midstream_failure(engine, &mock_table);
 }
 
-// Note: This should be removed once type widening support is added for CDF
 #[tokio::test]
 async fn incompatible_schemas_fail() {
     async fn assert_incompatible_schema(commit_schema: StructType, cdf_schema: StructType) {

kernel/src/table_changes/mod.rs

@@ -43,8 +43,6 @@ use crate::snapshot::{Snapshot, SnapshotRef};
 use crate::table_configuration::TableConfiguration;
 use crate::table_features::Operation;
 use crate::table_features::TableFeature;
-use crate::table_properties::TableProperties;
-use crate::utils::require;
 use crate::{DeltaResult, Engine, Error, Version};
 
 mod log_replay;
@@ -169,9 +167,6 @@ impl TableChanges {
         // [`check_cdf_table_properties`] to fail early. This also ensures that column mapping is
         // disabled.
         //
-        // We also check the [`Protocol`] using [`ensure_cdf_read_supported`] to verify that
-        // we support CDF with those features enabled.
-        //
         // Note: We must still check each metadata and protocol action in the CDF range.
         let check_table_config = |snapshot: &Snapshot| {
             if snapshot
@@ -244,17 +239,6 @@ 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")
-    );
-    Ok(())
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;