refactor: unify Reader/WriterFeature into a single TableFeature (delta-io#1345)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-incubator/delta-kernel-rs/blob/main/CONTRIBUTING.md
2. Run `cargo t --all-features --all-targets` to get started testing,
and run `cargo fmt`.
  3. Ensure you have added or run the appropriate tests for your PR.
4. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  5. Be sure to keep the PR description updated to reflect all changes.
-->

<!--
PR title formatting:
This project uses conventional commits:
https://www.conventionalcommits.org/

Each PR corresponds to a commit on the `main` branch, with the title of
the PR (typically) being
used for the commit message on main. In order to ensure proper
formatting in the CHANGELOG please
ensure your PR title adheres to the conventional commit specification.

Examples:
- new feature PR: "feat: new API for snapshot.update()"
- bugfix PR: "fix: correctly apply DV in read-table example"
-->

## What changes are proposed in this pull request?
closes delta-io#1340 

As suggested in above issue, this pr 
- adds validation check for given (reader_features, writer_features) on
`Protocol::try_new`
- refactors `ReaderFeature` `WriterFeature` into single `TableFeature` 

so that we can check only writer feature like
```rust
// If protocol is created via Protocol::try_new(), then it is valid
if !protocol.has_write_feature(&TableFeature::VariantType)
```
Note that there are three feature types, and the `Unknown` variant is
for handling `TableFeature::Unknown` which is either `Writer` or
`ReaderWriter` based on its actual usage.
```rust
pub(crate) enum FeatureType {
    Writer,
    ReaderWriter,
    Unknown,
}
```

<!--
Please clarify what changes you are proposing and why the changes are
needed.
The purpose of this section is to outline the changes, why they are
needed, and how this PR fixes the issue.
If the reason for the change is already explained clearly in an issue,
then it does not need to be restated here.
1. If you propose a new API or feature, clarify the use case for a new
API or feature.
  2. If you fix a bug, you can clarify why it is a bug.
-->

<!--
Uncomment this section if there are any changes affecting public APIs:
### This PR affects the following public APIs

If there are breaking changes, please ensure the `breaking-changes`
label gets added by CI, and describe why the changes are needed.

Note that _new_ public APIs are not considered breaking.
-->


## How was this change tested?
<!--
Please make sure to add test cases that check the changes thoroughly
including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please
clarify how you tested, ideally via a reproducible test documented in
the PR description.
-->
Yes, added new test

---------

Co-authored-by: Drake Lin <drakelin18@gmail.com>

Author: ding-young

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/mod.rs

@@ -686,7 +686,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;
 
@@ -697,8 +697,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/actions/visitors.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/engine/arrow_data.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_table_feature(&TableFeature::VariantType)
+        && !protocol.has_table_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/schema/variant_utils.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, TableFeature::ColumnMapping]),
             )
             .unwrap(),
         )])

kernel/src/table_changes/log_replay/tests.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};
@@ -272,8 +272,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 => {