refactor: move ensure_read_support/ensure_write_support to operation entry points

kernel/src/scan/mod.rs

@@ -28,7 +28,7 @@ use crate::schema::{
     ArrayType, DataType, MapType, PrimitiveType, Schema, SchemaRef, SchemaTransform, StructField,
     ToSchema as _,
 };
-use crate::table_features::ColumnMappingMode;
+use crate::table_features::{ColumnMappingMode, Operation};
 use crate::{DeltaResult, Engine, EngineData, Error, FileMeta, SnapshotRef, Version};
 
 use self::log_replay::scan_action_iter;
@@ -121,6 +121,10 @@ impl ScanBuilder {
         // if no schema is provided, use snapshot's entire schema (e.g. SELECT *)
         let logical_schema = self.schema.unwrap_or_else(|| self.snapshot.schema());
 
+        self.snapshot
+            .table_configuration()
+            .ensure_read_supported(Operation::Scan)?;
+
         let state_info = StateInfo::try_new(
             logical_schema,
             self.snapshot.table_configuration(),

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, TableFeature};
+use crate::table_features::{ColumnMappingMode, Operation, TableFeature};
 use crate::table_properties::TableProperties;
 use crate::utils::require;
 use crate::{DeltaResult, Engine, Error, Version};
@@ -146,18 +146,22 @@ impl TableChanges {
             end_version,
         )?;
 
-        // Both snapshots ensure that reading is supported at the start and end version using
-        // `ensure_read_supported`. Note that we must still verify that reading is
-        // supported for every protocol action in the CDF range.
         let start_snapshot = Snapshot::builder_for(table_root.as_url().clone())
             .at_version(start_version)
             .build(engine)?;
+        start_snapshot
+            .table_configuration()
+            .ensure_read_supported(Operation::Cdf)?;
+
         let end_snapshot = match end_version {
             Some(version) => Snapshot::builder_from(start_snapshot.clone())
                 .at_version(version)
                 .build(engine)?,
             None => Snapshot::builder_from(start_snapshot.clone()).build(engine)?,
         };
+        end_snapshot
+            .table_configuration()
+            .ensure_read_supported(Operation::Cdf)?;
 
         // we block reading catalog-managed tables with CDF for now. note this is best-effort just
         // checking that start/end snapshots are not catalog-managed.

kernel/src/table_configuration.rs

@@ -108,9 +108,6 @@
             version,
         };
 
-        // Validate read support after construction so we have access to all fields
-        table_config.ensure_read_supported()?;
-
         Ok(table_config)
     }
 
@@ -291,7 +288,7 @@
 
     /// Returns `Ok` if the kernel supports reading from this table. This checks that the
     /// protocol's reader features are all supported.
-    fn ensure_read_supported(&self) -> DeltaResult<()> {
+    pub(crate) fn ensure_read_supported(&self, operation: Operation) -> DeltaResult<()> {
         // Version check
         match self.protocol.min_reader_version() {
             1..=3 => {}
@@ -319,7 +316,7 @@
                     )))
                 }
                 KernelSupport::Custom(check) => {
-                    check(&self.protocol, &self.table_properties, Operation::Scan)?;
+                    check(&self.protocol, &self.table_properties, operation)?;
                 }
             };
 
@@ -333,7 +330,7 @@
     /// Returns `true` if the kernel supports writing to this table. This checks that the
     /// protocol's writer features are all supported.
     #[internal_api]
-    pub(crate) fn ensure_write_supported(&self) -> DeltaResult<()> {
+    pub(crate) fn ensure_write_supported(&self, operation: Operation) -> DeltaResult<()> {
         // Version check: We currently only support writing to tables with minWriterVersion 1, 2, or 7.
         // Below is a mapping of unsupported writer versions and the features they enable:
         //
@@ -373,7 +370,7 @@
                     )))
                 }
                 KernelSupport::Custom(check) => {
-                    check(&self.protocol, &self.table_properties, Operation::Scan)?;
+                    check(&self.protocol, &self.table_properties, operation)?;
                 }
             };
 
@@ -914,7 +911,7 @@
         ];
 
         for (table_configuration, result) in cases {
             match (table_configuration.ensure_write_supported(), result) {
                 (Ok(()), Ok(())) => { /* Correct result */ }
                 (actual_result, Err(expected)) => {
                     assert_result_error_with_message(actual_result, &expected.to_string());

kernel/src/transaction/mod.rs

@@ -26,6 +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::table_features::Operation;
 use crate::utils::{current_time_ms, require};
 use crate::{
     DataType, DeltaResult, Engine, EngineData, Expression, ExpressionRef, IntoEngineData,
@@ -177,7 +178,7 @@ impl Transaction {
         // important! before a read/write to the table we must check it is supported
         read_snapshot
             .table_configuration()
-            .ensure_write_supported()?;
+            .ensure_write_supported(Operation::Scan)?;
 
         let commit_timestamp = current_time_ms()?;