refactor!: New mod transforms for expression and schema transforms (delta-io#2077)

## What changes are proposed in this pull request?

Create a new top-level `transforms` module to host all things related to
schema and expression transforms (which become sub-modules). Move
`SchemaTransform` and `ExpressionTransform`, plus associated tests.

NOTE: The vast majority of "changes" in this PR are really just code
movement that git/github can't show cleanly (they show as big blocks of
deleted and added code). The rest is mod declaration and import churn.

### This PR affects the following public APIs

Module names changed/moved. No API signature or behavior changes.

## How was this change tested?

Pure code movement. Compilation suffices.

Author: scovich

CLAUDE.md

@@ -81,8 +81,9 @@ version. From it you build a `Scan` (reads) or `Transaction` (writes).
 assembles commit actions, enforces protocol compliance, delegates atomic commit to a
 `Committer`.
 
-**Engine trait:** four handlers (`StorageHandler`, `JsonHandler`, `ParquetHandler`,
-`EvaluationHandler`). `DefaultEngine` lives in `kernel/src/engine/default/`.
+**Engine trait:** five handlers (`StorageHandler`, `JsonHandler`, `ParquetHandler`,
+`EvaluationHandler`, optional `MetricsReporter`). `DefaultEngine` lives in
+`kernel/src/engine/default/`.
 
 **EngineData:** opaque columnar data interface. IMPORTANT: never access `EngineData` columns
 directly -- always use the visitor pattern (`visit_rows` with typed `GetData` accessors).
@@ -158,6 +159,8 @@ Keep this list updated when new protocol features are added to kernel.
 - **Column mapping:** Physical column names can differ from logical names. Always use
   the schema from `Snapshot::schema()` for user data columns. Metadata/system schema
   column names (defined by the protocol) are not subject to column mapping.
+- **Transforms:** Generic recursive schema and expression transform traits and helpers
+  are in `kernel/src/transforms/`.
 
 ## Code Style / Documentation
 

CLAUDE/architecture.md

@@ -127,6 +127,8 @@ all returned batches -- the engine may split a single file across multiple batch
 - `kernel/src/schema/` -- `StructType`/`StructField`/`DataType`, projections
 - `kernel/src/expressions/` -- expression AST (`Expression`, `Predicate`, `Scalar`),
   `column_expr!` macro
+- `kernel/src/transforms/` -- generic recursive transforms (`ExpressionTransform`,
+  `SchemaTransform`)
 - `kernel/src/checkpoint/` -- checkpoint writing (V1 and V2 single-file classic-named)
 - `kernel/src/table_configuration.rs` -- table metadata, properties, feature management
 - `kernel/src/table_features/` -- protocol feature definitions, `TableFeature` enum

kernel/src/engine/arrow_conversion.rs

@@ -357,9 +357,9 @@ mod tests {
     use crate::engine::arrow_data::unshredded_variant_arrow_type;
     use crate::parquet::arrow::PARQUET_FIELD_ID_META_KEY;
     use crate::schema::{
-        ArrayType, ColumnMetadataKey, DataType, MapType, MetadataValue, SchemaTransform,
-        StructField, StructType,
+        ArrayType, ColumnMetadataKey, DataType, MapType, MetadataValue, StructField, StructType,
     };
+    use crate::transforms::SchemaTransform;
     use crate::DeltaResult;
     use std::collections::HashMap;
 
@@ -410,7 +410,7 @@ mod tests {
         }
     }
 
-    impl<'a> crate::schema::SchemaTransform<'a> for FieldIdCollector {
+    impl<'a> SchemaTransform<'a> for FieldIdCollector {
         fn transform_struct_field(
             &mut self,
             field: &'a StructField,

kernel/src/expressions/literal_expression_transform.rs

@@ -7,9 +7,8 @@ use std::ops::Deref as _;
 use tracing::debug;
 
 use crate::expressions::{Expression, Scalar};
-use crate::schema::{
-    ArrayType, DataType, MapType, PrimitiveType, SchemaTransform, StructField, StructType,
-};
+use crate::schema::{ArrayType, DataType, MapType, PrimitiveType, StructField, StructType};
+use crate::transforms::SchemaTransform;
 use crate::DeltaResult;
 
 /// [`SchemaTransform`] that will transform a [`Schema`] and an ordered list of leaf values

kernel/src/expressions/mod.rs

@@ -13,19 +13,18 @@ pub use self::column_names::{
     ColumnName,
 };
 pub use self::scalars::{ArrayData, DecimalData, MapData, Scalar, StructData};
-use self::transforms::ExpressionTransform;
 use crate::kernel_predicates::{
     DirectDataSkippingPredicateEvaluator, DirectPredicateEvaluator,
     IndirectDataSkippingPredicateEvaluator,
 };
 use crate::schema::SchemaRef;
+use crate::transforms::ExpressionTransform;
 use crate::{DataType, DeltaResult, DynPartialEq};
 
 mod column_names;
 pub(crate) mod literal_expression_transform;
 pub(crate) use literal_expression_transform::literal_expression_transform;
 mod scalars;
-pub mod transforms;
 
 pub type ExpressionRef = std::sync::Arc<Expression>;
 pub type PredicateRef = std::sync::Arc<Predicate>;
@@ -286,7 +285,10 @@ where
 }
 
 impl OpaquePredicate {
-    fn new(op: OpaquePredicateOpRef, exprs: impl IntoIterator<Item = Expression>) -> Self {
+    pub(crate) fn new(
+        op: OpaquePredicateOpRef,
+        exprs: impl IntoIterator<Item = Expression>,
+    ) -> Self {
         let exprs = exprs.into_iter().collect();
         Self { op, exprs }
     }
@@ -305,7 +307,10 @@ pub struct OpaqueExpression {
 }
 
 impl OpaqueExpression {
-    fn new(op: OpaqueExpressionOpRef, exprs: impl IntoIterator<Item = Expression>) -> Self {
+    pub(crate) fn new(
+        op: OpaqueExpressionOpRef,
+        exprs: impl IntoIterator<Item = Expression>,
+    ) -> Self {
         let exprs = exprs.into_iter().collect();
         Self { op, exprs }
     }
@@ -545,21 +550,21 @@ impl JunctionPredicateOp {
 }
 
 impl UnaryExpression {
-    fn new(op: UnaryExpressionOp, expr: impl Into<Expression>) -> Self {
+    pub(crate) fn new(op: UnaryExpressionOp, expr: impl Into<Expression>) -> Self {
         let expr = Box::new(expr.into());
         Self { op, expr }
     }
 }
 
 impl UnaryPredicate {
-    fn new(op: UnaryPredicateOp, expr: impl Into<Expression>) -> Self {
+    pub(crate) fn new(op: UnaryPredicateOp, expr: impl Into<Expression>) -> Self {
         let expr = Box::new(expr.into());
         Self { op, expr }
     }
 }
 
 impl BinaryExpression {
-    fn new(
+    pub(crate) fn new(
         op: BinaryExpressionOp,
         left: impl Into<Expression>,
         right: impl Into<Expression>,
@@ -571,7 +576,7 @@ impl BinaryExpression {
 }
 
 impl BinaryPredicate {
-    fn new(
+    pub(crate) fn new(
         op: BinaryPredicateOp,
         left: impl Into<Expression>,
         right: impl Into<Expression>,
@@ -583,7 +588,7 @@ impl BinaryPredicate {
 }
 
 impl VariadicExpression {
-    fn new(
+    pub(crate) fn new(
         op: VariadicExpressionOp,
         exprs: impl IntoIterator<Item = impl Into<Expression>>,
     ) -> Self {
@@ -593,7 +598,7 @@ impl VariadicExpression {
 }
 
 impl ParseJsonExpression {
-    fn new(json_expr: impl Into<Expression>, output_schema: SchemaRef) -> Self {
+    pub(crate) fn new(json_expr: impl Into<Expression>, output_schema: SchemaRef) -> Self {
         Self {
             json_expr: Box::new(json_expr.into()),
             output_schema,
@@ -619,15 +624,15 @@ pub struct MapToStructExpression {
 }
 
 impl MapToStructExpression {
-    fn new(map_expr: impl Into<Expression>) -> Self {
+    pub(crate) fn new(map_expr: impl Into<Expression>) -> Self {
         Self {
             map_expr: Box::new(map_expr.into()),
         }
     }
 }
 
 impl JunctionPredicate {
-    fn new(op: JunctionPredicateOp, preds: Vec<Predicate>) -> Self {
+    pub(crate) fn new(op: JunctionPredicateOp, preds: Vec<Predicate>) -> Self {
         Self { op, preds }
     }
 }

kernel/src/lib.rs

@@ -108,6 +108,7 @@ pub mod table_configuration;
 pub mod table_features;
 pub mod table_properties;
 pub mod transaction;
+pub mod transforms;
 
 pub use log_path::LogPath;
 

kernel/src/scan/data_skipping/stats_schema/mod.rs

@@ -6,9 +6,10 @@ use std::borrow::Cow;
 use std::sync::Arc;
 
 use crate::schema::{
-    ArrayType, ColumnName, DataType, MapType, PrimitiveType, Schema, SchemaRef, SchemaTransform,
-    StructField, StructType,
+    ArrayType, ColumnName, DataType, MapType, PrimitiveType, Schema, SchemaRef, StructField,
+    StructType,
 };
+use crate::transforms::SchemaTransform;
 use crate::{DeltaResult, Error};
 
 use column_filter::StatsColumnFilter;

kernel/src/scan/mod.rs

@@ -16,7 +16,6 @@ use crate::actions::deletion_vector::{
 };
 use crate::actions::{get_commit_schema, Add, ADD_NAME, REMOVE_NAME};
 use crate::engine_data::FilteredEngineData;
-use crate::expressions::transforms::ExpressionTransform;
 use crate::expressions::{ColumnName, ExpressionRef, Predicate, PredicateRef, Scalar};
 use crate::kernel_predicates::{DefaultKernelPredicateEvaluator, EmptyColumnResolver};
 use crate::log_replay::{ActionsBatch, HasSelectionVector};
@@ -27,10 +26,11 @@ use crate::scan::log_replay::ScanLogReplayProcessor;
 use crate::scan::log_replay::{BASE_ROW_ID_NAME, CLUSTERING_PROVIDER_NAME};
 use crate::scan::state_info::StateInfo;
 use crate::schema::{
-    ArrayType, DataType, MapType, PrimitiveType, Schema, SchemaRef, SchemaTransform, StructField,
-    StructType, ToSchema as _,
+    ArrayType, DataType, MapType, PrimitiveType, Schema, SchemaRef, StructField, StructType,
+    ToSchema as _,
 };
 use crate::table_features::{ColumnMappingMode, Operation};
+use crate::transforms::{ExpressionTransform, SchemaTransform};
 use crate::{DeltaResult, Engine, EngineData, Error, FileMeta, SnapshotRef, Version};
 
 use self::log_replay::scan_action_iter;