fix: VARIANT follow-ups for SchemaTransform etc (delta-io#1106)

## What changes are proposed in this pull request?
Follow-up from VARIANT work in delta-io#1015. chiefly: (1) modify
`transform_variant` to take/return structtypes, and default to recursing
through children, and (2) track a couple follow ups in TODOs.

the actual bug that must be fixed is that when a VARIANT went in to
`transform_variant`, it came back out as a regular STRUCT if
transformed, since DataType::from on a struct is just a struct (note
that if not transformed, we return the original data_type which is still
variant)

original comment kept below for posterity

### Original exploration in earlier commits
Originally prompted as follow-up from delta-io#1015, this serves as an
exploration of introducing a new `VariantType` struct (similar to
`MapType`/`ArrayType`/etc.) which just wraps `StructType`.

The original question pertaining to `SchemaTransform` was basically: can
we implement `transform_variant` to take+return a `StructType` instead
of `DataType`? This loses the information that says the thing is a
variant and not a normal struct. This prompted the exploration of how we
actually want to represent Variant in our schemas. In general we need a
way of distinguishing variants from plain structs

I think we have two options: we either rely on `DataType` which
communicates `DataType::Variant(struct_type)` OR we introduce a new
`StructType` much like MapType/ArrayType/etc.

This PR serves as exploration of the former - though i'm realizing if we
_do_ take `DataType` as the thing that 'tags' a struct as variant then
it's probably also okay? And would just have the schema transform still
retain taking/returning DataType? (or perhaps could consume a StructType
but just always return a DataType)

I remember @nicklan had even more/better thoughts around this - please
comment and we can discuss :)

## How was this change tested?
Existing UT

Author: zachschuermann

kernel/src/schema/mod.rs

@@ -874,9 +874,10 @@ pub trait SchemaTransform<'a> {
         self.transform(etype)
     }
 
-    /// Called for each variant values encountered. By default does nothing
-    fn transform_variant(&mut self, etype: &'a DataType) -> Option<Cow<'a, DataType>> {
-        Some(Cow::Borrowed(etype))
+    /// Called for each variant value encountered. By default, recurses into the fields of the
+    /// variant struct type.
+    fn transform_variant(&mut self, stype: &'a StructType) -> Option<Cow<'a, StructType>> {
+        self.recurse_into_struct(stype)
     }
 
     /// General entry point for a recursive traversal over any data type. Also invoked internally to
@@ -896,9 +897,9 @@ pub trait SchemaTransform<'a> {
             Map(mtype) => self
                 .transform_map(mtype)?
                 .map_owned_or_else(data_type, DataType::from),
-            Variant(_) => self
-                .transform_variant(data_type)?
-                .map_owned_or_else(data_type, DataType::from),
+            Variant(stype) => self
+                .transform_variant(stype)?
+                .map_owned_or_else(data_type, |s| DataType::Variant(Box::new(s))),
         };
         Some(result)
     }

kernel/src/schema/variant_utils.rs

@@ -1,7 +1,7 @@
 //! Utility functions for the variant type and variant-related table features.
 
 use crate::actions::Protocol;
-use crate::schema::{DataType, Schema, SchemaTransform};
+use crate::schema::{Schema, SchemaTransform, StructType};
 use crate::table_features::{ReaderFeature, WriterFeature};
 use crate::utils::require;
 use crate::{DeltaResult, Error};
@@ -12,7 +12,7 @@ use std::borrow::Cow;
 pub(crate) struct UsesVariant(pub(crate) bool);
 
 impl<'a> SchemaTransform<'a> for UsesVariant {
-    fn transform_variant(&mut self, _: &'a DataType) -> Option<Cow<'a, DataType>> {
+    fn transform_variant(&mut self, _: &'a StructType) -> Option<Cow<'a, StructType>> {
         self.0 = true;
         None
     }

kernel/src/table_features/column_mapping.rs

@@ -1,7 +1,9 @@
 //! Code to handle column mapping, including modes and schema transforms
 use super::ReaderFeature;
 use crate::actions::Protocol;
-use crate::schema::{ColumnName, DataType, MetadataValue, Schema, SchemaTransform, StructField};
+use crate::schema::{
+    ColumnName, DataType, MetadataValue, Schema, SchemaTransform, StructField, StructType,
+};
 use crate::table_properties::TableProperties;
 use crate::{DeltaResult, Error};
 
@@ -155,6 +157,12 @@ impl<'a> SchemaTransform<'a> for ValidateColumnMappings<'a> {
         }
         None
     }
+    fn transform_variant(&mut self, _: &'a StructType) -> Option<Cow<'a, StructType>> {
+        // don't recurse into variant's fields, as they are not expected to have column mapping
+        // annotations
+        // TODO: this changes with icebergcompat right? see issue#1125 for icebergcompat.
+        None
+    }
 }
 
 #[cfg(test)]

test-utils/src/lib.rs

@@ -248,6 +248,10 @@ pub async fn create_table(
         }
         if enable_column_mapping {
             reader_features.push("columnMapping");
+            // TODO: (#1124) we don't actually support column mapping writes yet, but have some
+            // tests that do column mapping on writes. for now omit the writer feature to let tests
+            // run, but after actual support this should be enabled.
+            // writer_features.push("columnMapping");
         }
         (reader_features, writer_features)
     };
@@ -357,7 +361,6 @@ pub fn to_arrow(data: Box<dyn EngineData>) -> DeltaResult<RecordBatch> {
         .into())
 }
 
-// TODO (zach): this is listed as unused for acceptance crate
 pub fn read_scan(scan: &Scan, engine: Arc<dyn Engine>) -> DeltaResult<Vec<RecordBatch>> {
     let scan_results = scan.execute(engine)?;
     scan_results
@@ -375,7 +378,6 @@ pub fn read_scan(scan: &Scan, engine: Arc<dyn Engine>) -> DeltaResult<Vec<Record
         .try_collect()
 }
 
-// TODO (zach): this is listed as unused for acceptance crate
 pub fn test_read(
     expected: &ArrowEngineData,
     url: &Url,