refactor!: make snapshot.schema() return a SchemaRef

ffi/src/lib.rs

@@ -626,7 +626,7 @@ pub unsafe extern "C" fn version(snapshot: Handle<SharedSnapshot>) -> u64 {
 #[no_mangle]
 pub unsafe extern "C" fn logical_schema(snapshot: Handle<SharedSnapshot>) -> Handle<SharedSchema> {
     let snapshot = unsafe { snapshot.as_ref() };
-    Arc::new(snapshot.schema().clone()).into()
+    snapshot.schema().into()
 }
 
 /// Free a schema

kernel/src/scan/mod.rs

@@ -96,9 +96,7 @@ impl ScanBuilder {
     /// perform actual data reads.
     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().clone().into());
+        let logical_schema = self.schema.unwrap_or_else(|| self.snapshot.schema());
         let state_info = get_state_info(
             logical_schema.as_ref(),
             &self.snapshot.metadata().partition_columns,

kernel/src/snapshot.rs

@@ -9,7 +9,7 @@ use url::Url;
 use crate::actions::{Metadata, Protocol};
 use crate::log_segment::LogSegment;
 use crate::scan::ScanBuilder;
-use crate::schema::Schema;
+use crate::schema::{Schema, SchemaRef};
 use crate::table_configuration::TableConfiguration;
 use crate::table_features::ColumnMappingMode;
 use crate::table_properties::TableProperties;
@@ -98,8 +98,7 @@ impl Snapshot {
     }
 
     /// Table [`Schema`] at this `Snapshot`s version.
-    // TODO should this return SchemaRef?
-    pub fn schema(&self) -> &Schema {
+    pub fn schema(&self) -> SchemaRef {
         self.table_configuration.schema()
     }
 
@@ -203,7 +202,6 @@ mod tests {
     use crate::engine::default::filesystem::ObjectStoreFileSystemClient;
     use crate::engine::sync::SyncEngine;
     use crate::path::ParsedLogPath;
-    use crate::schema::StructType;
 
     #[test]
     fn test_snapshot_read_metadata() {
@@ -219,8 +217,8 @@ mod tests {
         assert_eq!(snapshot.protocol(), &expected);
 
         let schema_string = r#"{"type":"struct","fields":[{"name":"value","type":"integer","nullable":true,"metadata":{}}]}"#;
-        let expected: StructType = serde_json::from_str(schema_string).unwrap();
-        assert_eq!(snapshot.schema(), &expected);
+        let expected: SchemaRef = serde_json::from_str(schema_string).unwrap();
+        assert_eq!(snapshot.schema(), expected);
     }
 
     #[test]
@@ -237,8 +235,8 @@ mod tests {
         assert_eq!(snapshot.protocol(), &expected);
 
         let schema_string = r#"{"type":"struct","fields":[{"name":"value","type":"integer","nullable":true,"metadata":{}}]}"#;
-        let expected: StructType = serde_json::from_str(schema_string).unwrap();
-        assert_eq!(snapshot.schema(), &expected);
+        let expected: SchemaRef = serde_json::from_str(schema_string).unwrap();
+        assert_eq!(snapshot.schema(), expected);
     }
 
     #[test]

kernel/src/table_changes/scan.rs

@@ -197,7 +197,7 @@ impl TableChangesScan {
             PhysicalPredicate::Some(predicate, schema) => Some((predicate, schema)),
             PhysicalPredicate::None => None,
         };
-        let schema = self.table_changes.end_snapshot.schema().clone().into();
+        let schema = self.table_changes.end_snapshot.schema();
         let it = table_changes_action_iter(engine, commits, schema, physical_predicate)?;
         Ok(Some(it).into_iter().flatten())
     }

kernel/src/table_configuration.rs

@@ -14,7 +14,7 @@ use std::sync::{Arc, LazyLock};
 use url::Url;
 
 use crate::actions::{ensure_supported_features, Metadata, Protocol};
-use crate::schema::{InvariantChecker, Schema, SchemaRef};
+use crate::schema::{InvariantChecker, SchemaRef};
 use crate::table_features::{
     column_mapping_mode, validate_schema_column_mapping, ColumnMappingMode, ReaderFeatures,
     WriterFeatures,
@@ -101,10 +101,10 @@ impl TableConfiguration {
         &self.protocol
     }
 
-    /// The [`Schema`] of for this table at this version.
+    /// The logical schema ([`SchemaRef`]) of this table at this version.
     #[cfg_attr(feature = "developer-visibility", visibility::make(pub))]
-    pub(crate) fn schema(&self) -> &Schema {
-        self.schema.as_ref()
+    pub(crate) fn schema(&self) -> SchemaRef {
+        self.schema.clone()
     }
 
     /// The [`TableProperties`] of this table at this version.
@@ -139,7 +139,9 @@ impl TableConfiguration {
 
         // for now we don't allow invariants so although we support writer version 2 and the
         // ColumnInvariant TableFeature we _must_ check here that they are not actually in use
-        if self.is_invariants_supported() && InvariantChecker::has_invariants(self.schema()) {
+        if self.is_invariants_supported()
+            && InvariantChecker::has_invariants(self.schema().as_ref())
+        {
             return Err(Error::unsupported(
                 "Column invariants are not yet supported",
             ));

kernel/src/transaction.rs

@@ -151,8 +151,9 @@ impl Transaction {
         // 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 fields = self.read_snapshot.schema().fields();
-        let fields = fields
+        let schema = self.read_snapshot.schema();
+        let fields = schema
+            .fields()
             .filter(|f| !partition_columns.contains(f.name()))
             .map(|f| Expression::column([f.name()]));
         Expression::struct_from(fields)
@@ -167,11 +168,7 @@ impl Transaction {
         let target_dir = self.read_snapshot.table_root();
         let snapshot_schema = self.read_snapshot.schema();
         let logical_to_physical = self.generate_logical_to_physical();
-        WriteContext::new(
-            target_dir.clone(),
-            Arc::new(snapshot_schema.clone()),
-            logical_to_physical,
-        )
+        WriteContext::new(target_dir.clone(), snapshot_schema, logical_to_physical)
     }
 
     /// Add write metadata about files to include in the transaction. This API can be called