refactor: unify Reader/WriterFeature into a single TableFeature

kernel/src/actions/crc.rs

@@ -150,7 +150,7 @@ mod tests {
     use crate::engine::sync::SyncEngine;
     use crate::schema::derive_macro_utils::ToDataType as _;
     use crate::schema::{ArrayType, DataType, StructField, StructType};
-    use crate::table_features::{ReaderFeature, WriterFeature};
+    use crate::table_features::TableFeature;
     use crate::utils::test_utils::string_array_to_engine_data;
     use crate::Engine;
 
@@ -252,8 +252,8 @@ mod tests {
         let expected_protocol = Protocol {
             min_reader_version: 3,
             min_writer_version: 7,
-            reader_features: Some(vec![ReaderFeature::ColumnMapping]),
-            writer_features: Some(vec![WriterFeature::ColumnMapping]),
+            reader_features: Some(vec![TableFeature::ColumnMapping]),
+            writer_features: Some(vec![TableFeature::ColumnMapping]),
         };
         let expected_metadata = Metadata {
             id: "testId".to_string(),

kernel/src/actions/visitors.rs

@@ -685,7 +685,7 @@ mod tests {
 
     use crate::engine::sync::SyncEngine;
     use crate::expressions::{column_expr_ref, Expression};
-    use crate::table_features::{ReaderFeature, WriterFeature};
+    use crate::table_features::TableFeature;
     use crate::utils::test_utils::{action_batch, parse_json_batch};
     use crate::Engine;
 
@@ -696,8 +696,8 @@ mod tests {
         let expected = Protocol {
             min_reader_version: 3,
             min_writer_version: 7,
-            reader_features: Some(vec![ReaderFeature::DeletionVectors]),
-            writer_features: Some(vec![WriterFeature::DeletionVectors]),
+            reader_features: Some(vec![TableFeature::DeletionVectors]),
+            writer_features: Some(vec![TableFeature::DeletionVectors]),
         };
         assert_eq!(parsed, expected);
         Ok(())

kernel/src/engine/arrow_data.rs

@@ -352,7 +352,7 @@ mod tests {
     use crate::engine::sync::SyncEngine;
     use crate::expressions::ArrayData;
     use crate::schema::{ArrayType, DataType, StructField, StructType};
-    use crate::table_features::{ReaderFeature, WriterFeature};
+    use crate::table_features::TableFeature;
     use crate::utils::test_utils::{assert_result_error_with_message, string_array_to_engine_data};
     use crate::{DeltaResult, Engine as _, EngineData as _};
 
@@ -394,11 +394,11 @@ mod tests {
         assert_eq!(protocol.min_writer_version(), 7);
         assert_eq!(
             protocol.reader_features(),
-            Some([ReaderFeature::unknown("rw1")].as_slice())
+            Some([TableFeature::unknown("rw1")].as_slice())
         );
         assert_eq!(
             protocol.writer_features(),
-            Some([WriterFeature::unknown("rw1"), WriterFeature::unknown("w2")].as_slice())
+            Some([TableFeature::unknown("rw1"), TableFeature::unknown("w2")].as_slice())
         );
         Ok(())
     }

kernel/src/schema/variant_utils.rs

@@ -2,7 +2,7 @@
 
 use crate::actions::Protocol;
 use crate::schema::{Schema, SchemaTransform, StructType};
-use crate::table_features::{ReaderFeature, WriterFeature};
+use crate::table_features::TableFeature;
 use crate::utils::require;
 use crate::{DeltaResult, Error};
 use std::borrow::Cow;
@@ -24,10 +24,8 @@ pub(crate) fn validate_variant_type_feature_support(
 ) -> DeltaResult<()> {
     // Both the reader and writer need to have either the VariantType or the VariantTypePreview
     // features.
-    if (!protocol.has_reader_feature(&ReaderFeature::VariantType)
-        && !protocol.has_reader_feature(&ReaderFeature::VariantTypePreview))
-        || (!protocol.has_writer_feature(&WriterFeature::VariantType)
-            && !protocol.has_writer_feature(&WriterFeature::VariantTypePreview))
+    if !protocol.has_writer_feature(&TableFeature::VariantType)
+        && !protocol.has_writer_feature(&TableFeature::VariantTypePreview)
     {
         let mut uses_variant = UsesVariant::default();
         let _ = uses_variant.transform_struct(schema);
@@ -46,7 +44,7 @@ mod tests {
     use super::*;
     use crate::actions::Protocol;
     use crate::schema::{DataType, StructField, StructType};
-    use crate::table_features::{ReaderFeature, WriterFeature};
+    use crate::table_features::TableFeature;
     use crate::utils::test_utils::assert_result_error_with_message;
 
     #[test]
@@ -74,10 +72,10 @@ mod tests {
     #[test]
     fn test_variant_feature_validation() {
         let features = [
-            (ReaderFeature::VariantType, WriterFeature::VariantType),
+            (TableFeature::VariantType, TableFeature::VariantType),
             (
-                ReaderFeature::VariantTypePreview,
-                WriterFeature::VariantTypePreview,
+                TableFeature::VariantTypePreview,
+                TableFeature::VariantTypePreview,
             ),
         ];
         let schema_with_variant = StructType::new_unchecked([
@@ -120,15 +118,17 @@ mod tests {
                 )
                 .unwrap();
 
-                // Protocol without variantType writer feature
+                // Since variant features are ReaderWriter feature, protocol that
+                // lists a variant feature in only one of reader/writer feature = ERR
                 let protocol_without_writer_feature =
-                    Protocol::try_new(3, 7, Some([variant_reader]), Some::<Vec<String>>(vec![]))
-                        .unwrap();
+                    Protocol::try_new(3, 7, Some([variant_reader]), Some::<Vec<String>>(vec![]));
+                assert_result_error_with_message(protocol_without_writer_feature,
+                    "Reader features must contain only ReaderWriter features that are also listed in writer features");
 
-                // Protocol without variantType reader feature
                 let protocol_without_reader_feature =
-                    Protocol::try_new(3, 7, Some::<Vec<String>>(vec![]), Some([variant_writer]))
-                        .unwrap();
+                    Protocol::try_new(3, 7, Some::<Vec<String>>(vec![]), Some([variant_writer]));
+                assert_result_error_with_message(protocol_without_reader_feature,
+                    "Writer features must be Writer-only or also listed in reader features");
 
                 // Schema with VARIANT + Protocol with features = OK
                 validate_variant_type_feature_support(
@@ -163,20 +163,6 @@ mod tests {
                     &protocol_without_features,
                 );
                 assert_result_error_with_message(result, "Unsupported: Table contains VARIANT columns but does not have the required 'variantType' feature in reader and writer features");
-
-                // Schema with VARIANT + Protocol without writer feature = ERROR
-                let result = validate_variant_type_feature_support(
-                    &schema_with_variant,
-                    &protocol_without_writer_feature,
-                );
-                assert_result_error_with_message(result, "Unsupported: Table contains VARIANT columns but does not have the required 'variantType' feature in reader and writer features");
-
-                // Schema with VARIANT + Protocol without reader feature = ERROR
-                let result = validate_variant_type_feature_support(
-                    &schema_with_variant,
-                    &protocol_without_reader_feature,
-                );
-                assert_result_error_with_message(result, "Unsupported: Table contains VARIANT columns but does not have the required 'variantType' feature in reader and writer features");
             });
     }
 }

kernel/src/table_changes/log_replay/tests.rs

@@ -10,7 +10,7 @@ use crate::scan::state::DvInfo;
 use crate::scan::PhysicalPredicate;
 use crate::schema::{DataType, StructField, StructType};
 use crate::table_changes::log_replay::LogReplayScanner;
-use crate::table_features::ReaderFeature;
+use crate::table_features::TableFeature;
 use crate::utils::test_utils::{assert_result_error_with_message, Action, LocalMockTable};
 use crate::Predicate;
 use crate::{DeltaResult, Engine, Error, Version};
@@ -79,8 +79,8 @@ async fn metadata_protocol() {
                 Protocol::try_new(
                     3,
                     7,
-                    Some([ReaderFeature::DeletionVectors]),
-                    Some([ReaderFeature::ColumnMapping]),
+                    Some([TableFeature::DeletionVectors]),
+                    Some([TableFeature::DeletionVectors]),
                 )
                 .unwrap(),
             ),
@@ -138,8 +138,8 @@ async fn unsupported_reader_feature() {
             Protocol::try_new(
                 3,
                 7,
-                Some([ReaderFeature::DeletionVectors, ReaderFeature::ColumnMapping]),
-                Some([""; 0]),
+                Some([TableFeature::DeletionVectors, TableFeature::ColumnMapping]),
+                Some([TableFeature::DeletionVectors]),
             )
             .unwrap(),
         )])

kernel/src/table_changes/mod.rs

@@ -41,7 +41,7 @@ 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, ReaderFeature};
+use crate::table_features::{ColumnMappingMode, TableFeature};
 use crate::table_properties::TableProperties;
 use crate::utils::require;
 use crate::{DeltaResult, Engine, Error, Version};
@@ -270,8 +270,8 @@ fn check_cdf_table_properties(table_properties: &TableProperties) -> DeltaResult
 /// 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<ReaderFeature>> =
-        LazyLock::new(|| vec![ReaderFeature::DeletionVectors]);
+    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 => {