serde first draft

Author: OussamaSaoudi

kernel/src/log_replay.rs

@@ -26,7 +26,7 @@ use tracing::debug;
 
 /// The subset of file action fields that uniquely identifies it in the log, used for deduplication
 /// of adds and removes during log replay.
-#[derive(Debug, Hash, Eq, PartialEq, Clone)]
+#[derive(Debug, Hash, Eq, PartialEq, serde::Serialize, serde::Deserialize, Clone)]
 pub(crate) struct FileActionKey {
     pub(crate) path: String,
     pub(crate) dv_unique_id: Option<String>,

kernel/src/scan/log_replay.rs

@@ -12,12 +12,34 @@ use crate::expressions::{column_name, ColumnName, Expression, ExpressionRef, Pre
 use crate::kernel_predicates::{DefaultKernelPredicateEvaluator, KernelPredicateEvaluator as _};
 use crate::log_replay::{ActionsBatch, FileActionDeduplicator, FileActionKey, LogReplayProcessor};
 use crate::scan::Scalar;
-use crate::schema::ToSchema as _;
+use crate::schema::{SchemaRef, ToSchema as _};
 use crate::schema::{ColumnNamesAndTypes, DataType, MapType, StructField, StructType};
-use crate::transforms::{get_transform_expr, parse_partition_values};
+use crate::table_features::ColumnMappingMode;
+use crate::transforms::{get_transform_expr, parse_partition_values, TransformSpec};
 use crate::utils::require;
 use crate::{DeltaResult, Engine, Error, ExpressionEvaluator};
 
+/// Internal serializable state (schemas, transform spec, column mapping, etc.)
+/// This is opaque to the engine - just passed through as a blob.
+#[derive(serde::Serialize, serde::Deserialize, Clone)]
+struct InternalState {
+    logical_schema: StructType,
+    physical_schema: StructType,
+    transform_spec: Option<TransformSpec>,
+    column_mapping_mode: ColumnMappingMode,
+}
+
+/// Public-facing serialized processor state for distribution to executors.
+/// The engine passes everything needed as Arc references, plus an opaque internal state blob.
+#[derive(Clone)]
+pub struct SerializedScanState {
+    /// Optional predicate for data skipping (if provided)
+    pub predicate: Option<PredicateRef>,
+    /// Opaque internal state blob (JSON for now)
+    pub internal_state_blob: Vec<u8>,
+}
+
+
 /// [`ScanLogReplayProcessor`] performs log replay (processes actions) specifically for doing a table scan.
 ///
 /// During a table scan, the processor reads batches of log actions (in reverse chronological order)
@@ -44,11 +66,11 @@ pub(crate) struct ScanLogReplayProcessor {
     partition_filter: Option<PredicateRef>,
     data_skipping_filter: Option<DataSkippingFilter>,
     add_transform: Arc<dyn ExpressionEvaluator>,
-    state_info: Arc<StateInfo>,
+    pub(crate) state_info: Arc<StateInfo>,
     /// A set of (data file path, dv_unique_id) pairs that have been seen thus
     /// far in the log. This is used to filter out files with Remove actions as
     /// well as duplicate entries in the log.
-    seen_file_keys: HashSet<FileActionKey>,
+    pub(crate) seen_file_keys: HashSet<FileActionKey>,
 }
 
 impl ScanLogReplayProcessor {
@@ -84,6 +106,91 @@ impl ScanLogReplayProcessor {
             state_info,
         })
     }
+
+    /// Serialize the processor state for distribution to executors.
+    ///
+    /// Consumes the processor and returns:
+    /// - `SerializedScanState`: Public-facing state with predicate and internal blob
+    /// - `HashSet<FileActionKey>`: The deduplication set (moved for independent use on executors)
+    ///
+    /// Executors can use `from_serialized` to reconstruct the processor with this state.
+    pub(crate) fn serialize(self) -> DeltaResult<(SerializedScanState, HashSet<FileActionKey>)> {
+        // Serialize internal state to JSON blob (schemas, transform spec, and column mapping mode)
+        let internal_state = InternalState {
+            logical_schema: (*self.state_info.logical_schema).clone(),
+            physical_schema: (*self.state_info.physical_schema).clone(),
+            transform_spec: self.state_info.transform_spec.as_ref().map(|ts| (**ts).clone()),
+            column_mapping_mode: self.state_info.column_mapping_mode,
+        };
+        let internal_state_blob = serde_json::to_vec(&internal_state)
+            .map_err(|e| Error::generic(format!("Failed to serialize internal state: {}", e)))?;
+
+        // Extract predicate from PhysicalPredicate
+        let predicate = match &self.state_info.physical_predicate {
+            PhysicalPredicate::Some(pred, _schema) => Some(pred.clone()),
+            _ => None,
+        };
+
+        let state = SerializedScanState {
+            predicate,
+            internal_state_blob,
+        };
+
+        Ok((state, self.seen_file_keys))
+    }
+
+    /// Reconstruct a processor from serialized state.
+    ///
+    /// Creates a new processor with the provided state and seen_file_keys.
+    /// All other fields (partition_filter, data_skipping_filter, add_transform) are
+    /// reconstructed from the state and engine.
+    ///
+    /// # Parameters
+    /// - `engine`: Engine for creating evaluators and filters
+    /// - `state`: The serialized state from serialization
+    /// - `seen_file_keys`: The deduplication set from serialization
+    pub(crate) fn from_serialized(
+        engine: &dyn Engine,
+        state: SerializedScanState,
+        seen_file_keys: HashSet<FileActionKey>,
+    ) -> DeltaResult<Self> {
+        // Deserialize internal state
+        let internal_state: InternalState = serde_json::from_slice(&state.internal_state_blob)
+            .map_err(|e| Error::generic(format!("Failed to deserialize internal state: {}", e)))?;
+
+        // Convert schemas to Arc
+        let logical_schema = Arc::new(internal_state.logical_schema);
+        let physical_schema = Arc::new(internal_state.physical_schema);
+
+        // Reconstruct PhysicalPredicate from predicate and schema
+        let physical_predicate = match state.predicate {
+            Some(pred) => PhysicalPredicate::Some(pred, physical_schema.clone()),
+            None => PhysicalPredicate::None,
+        };
+
+        // Reconstruct StateInfo
+        let state_info = Arc::new(StateInfo {
+            logical_schema,
+            physical_schema,
+            physical_predicate,
+            transform_spec: internal_state.transform_spec.map(Arc::new),
+            column_mapping_mode: internal_state.column_mapping_mode,
+        });
+
+        // Create processor and set seen_file_keys
+        let mut processor = Self::new(engine, state_info)?;
+        processor.seen_file_keys = seen_file_keys;
+        Ok(processor)
+    }
+
+    /// Get the projected schema needed to read checkpoint/sidecar files.
+    ///
+    /// Returns the schema that should be used when reading leaf checkpoint files
+    /// or sidecars during the executor phase.
+    pub(crate) fn get_projected_schema(&self) -> DeltaResult<crate::schema::SchemaRef> {
+        use crate::actions::ADD_NAME;
+        get_log_add_schema().project(&[ADD_NAME])
+    }
 }
 
 /// A visitor that deduplicates a stream of add and remove actions into a stream of valid adds. Log

kernel/src/scan/mod.rs

@@ -146,6 +146,27 @@ pub(crate) enum PhysicalPredicate {
     None,
 }
 
+/// Serializable representation of PhysicalPredicate (skipping the predicate expression itself)
+#[derive(serde::Serialize, serde::Deserialize, Clone)]
+enum SerializablePhysicalPredicate {
+    /// Has a predicate and schema (but we skip serializing the predicate itself)
+    Some { schema: crate::schema::StructType },
+    StaticSkipAll,
+    None,
+}
+
+impl From<&PhysicalPredicate> for SerializablePhysicalPredicate {
+    fn from(pred: &PhysicalPredicate) -> Self {
+        match pred {
+            PhysicalPredicate::Some(_pred, schema) => SerializablePhysicalPredicate::Some {
+                schema: (**schema).clone(),
+            },
+            PhysicalPredicate::StaticSkipAll => SerializablePhysicalPredicate::StaticSkipAll,
+            PhysicalPredicate::None => SerializablePhysicalPredicate::None,
+        }
+    }
+}
+
 impl PhysicalPredicate {
     /// If we have a predicate, verify the columns it references and apply column mapping. First, get
     /// the set of references; use that to filter the schema to only the columns of interest (and
@@ -404,6 +425,68 @@ impl Scan {
         self.scan_metadata_inner(engine, self.replay_for_scan_metadata(engine)?)
     }
 
+    /// Get a distributed driver phase for processing scan metadata.
+    ///
+    /// This method returns a [`DriverPhase`] that can be used for distributed execution.
+    /// The driver processes commits and single-part checkpoint manifests, then returns
+    /// the processor state and files that need to be distributed to executors.
+    ///
+    /// # Distributed Execution Pattern
+    ///
+    /// 1. **Driver phase**: Process commits and manifests on the coordinator
+    /// 2. **Serialize state**: Extract processor state and deduplicator
+    /// 3. **Distribute**: Send state + files to executors
+    /// 4. **Executor phase**: Each executor processes its assigned files
+    ///
+    /// # Example
+    ///
+    /// ```ignore
+    /// // On driver/coordinator
+    /// let mut driver = scan.scan_metadata_distributed(engine.clone())?;
+    ///
+    /// // Process driver-side batches
+    /// for batch in driver.by_ref() {
+    ///     let scan_metadata = batch?;
+    ///     // Handle scan metadata from commits/manifest
+    /// }
+    ///
+    /// // Check if executor phase is needed
+    /// match driver.finish()? {
+    ///     DriverPhaseResult::Complete(_processor) => {
+    ///         // All done! No executor phase needed
+    ///     }
+    ///     DriverPhaseResult::NeedsExecutorPhase { processor, files } => {
+    ///         // Serialize for distribution
+    ///         let (state, deduplicator) = processor.serialize()?;
+    ///        
+    ///         // Partition files and distribute to executors
+    ///         for (executor, file_partition) in partition(files) {
+    ///             executor.send(state.clone(), deduplicator.clone(), file_partition);
+    ///         }
+    ///     }
+    /// }
+    /// ```
+    ///
+    /// [`DriverPhase`]: crate::distributed::DriverPhase
+    pub(crate) fn scan_metadata_distributed(
+        &self,
+        engine: Arc<dyn Engine>,
+    ) -> DeltaResult<crate::distributed::DriverPhase<log_replay::ScanLogReplayProcessor>> {
+        use crate::distributed::DriverPhase;
+        
+        // Create the processor
+        let processor = log_replay::ScanLogReplayProcessor::new(
+            engine.as_ref(),
+            self.state_info.clone(),
+        )?;
+        
+        // Get log segment
+        let log_segment = Arc::new(self.snapshot.log_segment().clone());
+        
+        // Create and return driver phase
+        DriverPhase::try_new(processor, log_segment, engine)
+    }
+
     /// Get an updated iterator of [`ScanMetadata`]s based on an existing iterator of [`EngineData`]s.
     ///
     /// The existing iterator is assumed to contain data from a previous call to `scan_metadata`.

kernel/src/transforms.rs

@@ -23,16 +23,8 @@ pub(crate) type TransformSpec = Vec<FieldTransformSpec>;
 ///
 /// These transformations are "sparse" - they only specify what changes, while unchanged fields
 /// pass through implicitly in their original order.
-#[derive(Debug)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub(crate) enum FieldTransformSpec {
-    /// Insert the given expression after the named input column (None = prepend instead)
-    // NOTE: It's quite likely we will sometimes need to reorder columns for one reason or another,
-    // which would usually be expressed as a drop+insert pair of transforms.
-    #[allow(unused)]
-    StaticInsert {
-        insert_after: Option<String>,
-        expr: ExpressionRef,
-    },
     /// Drops the named input column
     // NOTE: Row tracking needs to drop metadata columns that were used to compute rowids, since
     // they should not appear in the query's output.
@@ -102,7 +94,6 @@ pub(crate) fn parse_partition_values(
                 ))
             }
             FieldTransformSpec::DynamicColumn { .. }
-            | FieldTransformSpec::StaticInsert { .. }
             | FieldTransformSpec::GenerateRowId { .. }
             | FieldTransformSpec::StaticDrop { .. } => None,
         })
@@ -125,9 +116,6 @@ pub(crate) fn get_transform_expr(
     for field_transform in transform_spec {
         use FieldTransformSpec::*;
         transform = match field_transform {
-            StaticInsert { insert_after, expr } => {
-                transform.with_inserted_field(insert_after.clone(), expr.clone())
-            }
             StaticDrop { field_name } => transform.with_dropped_field(field_name.clone()),
             GenerateRowId {
                 field_name,
@@ -363,13 +351,8 @@ mod tests {
     }
 
     #[test]
-    fn test_get_transform_expr_static_transforms() {
-        let expr = Arc::new(Expression::literal(42));
+    fn test_get_transform_expr_static_drop() {
         let transform_spec = vec![
-            FieldTransformSpec::StaticInsert {
-                insert_after: Some("col1".to_string()),
-                expr: expr.clone(),
-            },
             FieldTransformSpec::StaticDrop {
                 field_name: "col2".to_string(),
             },
@@ -393,16 +376,6 @@ mod tests {
             panic!("Expected Transform expression");
         };
 
-        // Verify StaticInsert: should insert after col1
-        assert!(transform.field_transforms.contains_key("col1"));
-        assert!(!transform.field_transforms["col1"].is_replace);
-        assert_eq!(transform.field_transforms["col1"].exprs.len(), 1);
-        let Expression::Literal(scalar) = transform.field_transforms["col1"].exprs[0].as_ref()
-        else {
-            panic!("Expected literal expression for insert");
-        };
-        assert_eq!(scalar, &Scalar::Integer(42));
-
         // Verify StaticDrop: should drop col2 (empty expressions and is_replace = true)
         assert!(transform.field_transforms.contains_key("col2"));
         assert!(transform.field_transforms["col2"].is_replace);