refactor!: Snapshot should not expose delta implementation details (delta-io#1339)

Fixes delta-io#648

## Why this is a breaking change:
This PR removes `Snapshot::metadata`, `Snapshot::protocol`, and
`Snapshot::column_mapping_mode`. These are now all accessed through
`snapshot.table_configuration`

Author: aleksandarskrbic

acceptance/src/meta.rs

@@ -83,8 +83,8 @@ impl TestCaseInfo {
         assert_eq!(snapshot.version(), case.version);
 
         // assert correct metadata is read
-        let metadata = snapshot.metadata();
-        let protocol = snapshot.protocol();
+        let metadata = snapshot.table_configuration().metadata();
+        let protocol = snapshot.table_configuration().protocol();
         let tvm = TableVersionMetaData {
             version: snapshot.version(),
             properties: metadata

ffi/src/lib.rs

@@ -675,7 +675,11 @@ pub unsafe extern "C" fn snapshot_table_root(
 #[no_mangle]
 pub unsafe extern "C" fn get_partition_column_count(snapshot: Handle<SharedSnapshot>) -> usize {
     let snapshot = unsafe { snapshot.as_ref() };
-    snapshot.metadata().partition_columns().len()
+    snapshot
+        .table_configuration()
+        .metadata()
+        .partition_columns()
+        .len()
 }
 
 /// Get an iterator of the list of partition columns for this snapshot.
@@ -687,8 +691,14 @@ pub unsafe extern "C" fn get_partition_columns(
     snapshot: Handle<SharedSnapshot>,
 ) -> Handle<StringSliceIterator> {
     let snapshot = unsafe { snapshot.as_ref() };
-    let iter: Box<StringIter> =
-        Box::new(snapshot.metadata().partition_columns().clone().into_iter());
+    let iter: Box<StringIter> = Box::new(
+        snapshot
+            .table_configuration()
+            .metadata()
+            .partition_columns()
+            .clone()
+            .into_iter(),
+    );
     iter.into()
 }
 

kernel/examples/inspect-table/src/main.rs

@@ -191,7 +191,7 @@ fn try_main() -> DeltaResult<()> {
             println!("Latest table version: {}", snapshot.version());
         }
         Commands::Metadata => {
-            println!("{:#?}", snapshot.metadata());
+            println!("{:#?}", snapshot.table_configuration().metadata());
         }
         Commands::Schema => {
             println!("{:#?}", snapshot.schema());

kernel/src/scan/mod.rs

@@ -116,10 +116,17 @@ impl ScanBuilder {
     pub fn build(self) -> DeltaResult<Scan> {
         // if no schema is provided, use snapshot's entire schema (e.g. SELECT *)
         let logical_schema = self.schema.unwrap_or_else(|| self.snapshot.schema());
+        let partition_columns = self
+            .snapshot
+            .table_configuration()
+            .metadata()
+            .partition_columns();
+        let column_mapping_mode = self.snapshot.table_configuration().column_mapping_mode();
+
         let state_info = StateInfo::try_new(
             logical_schema,
-            self.snapshot.metadata().partition_columns(),
-            self.snapshot.table_configuration().column_mapping_mode(),
+            partition_columns,
+            column_mapping_mode,
             self.predicate,
         )?;
 

kernel/src/snapshot.rs

@@ -6,14 +6,13 @@ use std::sync::Arc;
 use crate::action_reconciliation::calculate_transaction_expiration_timestamp;
 use crate::actions::domain_metadata::domain_metadata_configuration;
 use crate::actions::set_transaction::SetTransactionScanner;
-use crate::actions::{Metadata, Protocol, INTERNAL_DOMAIN_PREFIX};
+use crate::actions::INTERNAL_DOMAIN_PREFIX;
 use crate::checkpoint::CheckpointWriter;
 use crate::listed_log_files::ListedLogFiles;
 use crate::log_segment::LogSegment;
 use crate::scan::ScanBuilder;
 use crate::schema::SchemaRef;
 use crate::table_configuration::TableConfiguration;
-use crate::table_features::ColumnMappingMode;
 use crate::table_properties::TableProperties;
 use crate::transaction::Transaction;
 use crate::LogCompactionWriter;
@@ -50,7 +49,7 @@ impl std::fmt::Debug for Snapshot {
         f.debug_struct("Snapshot")
             .field("path", &self.log_segment.log_root.as_str())
             .field("version", &self.version())
-            .field("metadata", &self.metadata())
+            .field("metadata", &self.table_configuration().metadata())
             .finish()
     }
 }
@@ -311,19 +310,6 @@ impl Snapshot {
         self.table_configuration.schema()
     }
 
-    /// Table [`Metadata`] at this `Snapshot`s version.
-    #[internal_api]
-    pub(crate) fn metadata(&self) -> &Metadata {
-        self.table_configuration.metadata()
-    }
-
-    /// Table [`Protocol`] at this `Snapshot`s version.
-    #[allow(dead_code)]
-    #[internal_api]
-    pub(crate) fn protocol(&self) -> &Protocol {
-        self.table_configuration.protocol()
-    }
-
     /// Get the [`TableProperties`] for this [`Snapshot`].
     pub fn table_properties(&self) -> &TableProperties {
         self.table_configuration().table_properties()
@@ -335,13 +321,6 @@ impl Snapshot {
         &self.table_configuration
     }
 
-    /// Get the [column mapping
-    /// mode](https://github.com/delta-io/delta/blob/master/PROTOCOL.md#column-mapping) at this
-    /// `Snapshot`s version.
-    pub fn column_mapping_mode(&self) -> ColumnMappingMode {
-        self.table_configuration.column_mapping_mode()
-    }
-
     /// Create a [`ScanBuilder`] for an `SnapshotRef`.
     pub fn scan_builder(self: Arc<Self>) -> ScanBuilder {
         ScanBuilder::new(self)
@@ -408,6 +387,7 @@ mod tests {
     use crate::arrow::record_batch::RecordBatch;
     use crate::parquet::arrow::ArrowWriter;
 
+    use crate::actions::Protocol;
     use crate::engine::arrow_data::ArrowEngineData;
     use crate::engine::default::executor::tokio::TokioBackgroundExecutor;
     use crate::engine::default::filesystem::ObjectStoreStorageHandler;
@@ -432,7 +412,7 @@ mod tests {
 
         let expected =
             Protocol::try_new(3, 7, Some(["deletionVectors"]), Some(["deletionVectors"])).unwrap();
-        assert_eq!(snapshot.protocol(), &expected);
+        assert_eq!(snapshot.table_configuration().protocol(), &expected);
 
         let schema_string = r#"{"type":"struct","fields":[{"name":"value","type":"integer","nullable":true,"metadata":{}}]}"#;
         let expected: SchemaRef = serde_json::from_str(schema_string).unwrap();
@@ -450,7 +430,7 @@ mod tests {
 
         let expected =
             Protocol::try_new(3, 7, Some(["deletionVectors"]), Some(["deletionVectors"])).unwrap();
-        assert_eq!(snapshot.protocol(), &expected);
+        assert_eq!(snapshot.table_configuration().protocol(), &expected);
 
         let schema_string = r#"{"type":"struct","fields":[{"name":"value","type":"integer","nullable":true,"metadata":{}}]}"#;
         let expected: SchemaRef = serde_json::from_str(schema_string).unwrap();

kernel/src/table_changes/mod.rs

@@ -166,8 +166,14 @@ impl TableChanges {
         // TODO: link issue
         #[cfg(feature = "catalog-managed")]
         require!(
-            !start_snapshot.protocol().is_catalog_managed()
-                && !end_snapshot.protocol().is_catalog_managed(),
+            !start_snapshot
+                .table_configuration()
+                .protocol()
+                .is_catalog_managed()
+                && !end_snapshot
+                    .table_configuration()
+                    .protocol()
+                    .is_catalog_managed(),
             Error::unsupported("Change data feed is not supported for catalog-managed tables")
         );
 
@@ -236,7 +242,10 @@ impl TableChanges {
     }
     /// The partition columns that will be read.
     pub(crate) fn partition_columns(&self) -> &Vec<String> {
-        self.end_snapshot.metadata().partition_columns()
+        self.end_snapshot
+            .table_configuration()
+            .metadata()
+            .partition_columns()
     }
 
     /// Create a [`TableChangesScanBuilder`] for an `Arc<TableChanges>`.

kernel/src/table_changes/scan.rs

@@ -159,8 +159,12 @@ impl TableChangesScanBuilder {
                 } else {
                     // Add to read schema, store field so we can build a `Column` expression later
                     // if needed (i.e. if we have partition columns)
-                    let physical_field = logical_field
-                        .make_physical(self.table_changes.end_snapshot.column_mapping_mode());
+                    let physical_field = logical_field.make_physical(
+                        self.table_changes
+                            .end_snapshot
+                            .table_configuration()
+                            .column_mapping_mode(),
+                    );
                     debug!("\n\n{logical_field:#?}\nAfter mapping: {physical_field:#?}\n\n");
                     let physical_name = physical_field.name.clone();
                     read_fields.push(physical_field);

kernel/src/transaction/mod.rs

@@ -336,7 +336,11 @@ impl Transaction {
     fn generate_logical_to_physical(&self) -> Expression {
         // for now, we just pass through all the columns except partition columns.
         // note this is _incorrect_ if table config deems we need partition columns.
-        let partition_columns = self.read_snapshot.metadata().partition_columns();
+        let partition_columns = self
+            .read_snapshot
+            .table_configuration()
+            .metadata()
+            .partition_columns();
         let schema = self.read_snapshot.schema();
         let fields = schema
             .fields()