rename

Author: OussamaSaoudi

kernel/src/log_reader/checkpoint_leaf.rs

@@ -0,0 +1,148 @@
+//! Sidecar phase for log replay - processes sidecar/leaf parquet files.
+
+use std::sync::Arc;
+
+use crate::log_replay::ActionsBatch;
+use crate::schema::SchemaRef;
+use crate::{DeltaResult, Engine, FileMeta};
+
+/// Phase that processes sidecar or leaf parquet files.
+///
+/// This phase is distributable - you can partition `files` and create multiple
+/// instances on different executors.
+#[allow(unused)]
+pub(crate) struct CheckpointLeafReader {
+    actions: Box<dyn Iterator<Item = DeltaResult<ActionsBatch>> + Send>,
+}
+
+impl CheckpointLeafReader {
+    /// Create a new sidecar phase from file list.
+    ///
+    /// # Distributability
+    ///
+    /// This phase is designed to be distributable. To distribute:
+    /// 1. Partition `files` across N executors/threads
+    /// 2. Create N `CheckpointLeafReader` instances, one per executor with its file partition
+    ///
+    /// # Parameters
+    /// - `files`: Sidecar/leaf files to process
+    /// - `engine`: Engine for reading files
+    /// - `schema`: Schema to use when reading sidecar files (projected based on processor requirements)
+    #[allow(unused)]
+    pub(crate) fn try_new(
+        engine: Arc<dyn Engine>,
+        files: Vec<FileMeta>,
+        schema: SchemaRef,
+    ) -> DeltaResult<Self> {
+        let actions = engine
+            .parquet_handler()
+            .read_parquet_files(&files, schema, None)?
+            .map(|batch| batch.map(|b| ActionsBatch::new(b, false)));
+
+        Ok(Self {
+            actions: Box::new(actions),
+        })
+    }
+}
+
+impl Iterator for CheckpointLeafReader {
+    type Item = DeltaResult<ActionsBatch>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.actions.next()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::actions::{get_commit_schema, ADD_NAME};
+    use crate::arrow::array::{Array, StringArray, StructArray};
+    use crate::engine::arrow_data::EngineDataArrowExt as _;
+    use crate::log_reader::checkpoint_manifest::CheckpointManifestReader;
+    use crate::utils::test_utils::load_test_table_with_data;
+    use itertools::Itertools;
+    use std::sync::Arc;
+
+    #[test]
+    fn test_sidecar_phase_processes_files() -> DeltaResult<()> {
+        let (engine, snapshot, _tempdir) =
+            load_test_table_with_data("tests/data", "v2-checkpoints-json-with-sidecars")?;
+        let log_segment = snapshot.log_segment();
+
+        // First we need to run through manifest phase to get the sidecar files
+        if log_segment.checkpoint_parts.is_empty() {
+            println!("Test table has no checkpoint parts, skipping");
+            return Ok(());
+        }
+
+        // Get the first checkpoint part
+        let checkpoint_file = &log_segment.checkpoint_parts[0];
+
+        let mut manifest_phase = CheckpointManifestReader::try_new(
+            checkpoint_file,
+            log_segment.log_root.clone(),
+            engine.clone(),
+        )?;
+
+        // Drain manifest phase
+        for batch in manifest_phase.by_ref() {
+            let _batch = batch?;
+        }
+
+        let sidecars = manifest_phase.extract_sidecars()?;
+
+        println!("Testing with {} sidecar files", sidecars.len());
+
+        let schema = get_commit_schema().project(&[ADD_NAME])?;
+
+        let mut sidecar_phase = CheckpointLeafReader::try_new(engine.clone(), sidecars, schema)?;
+
+        let mut sidecar_file_paths = Vec::new();
+
+        for result in sidecar_phase {
+            let batch = result?;
+            let ActionsBatch {
+                actions,
+                is_log_batch,
+            } = batch;
+            assert!(!is_log_batch, "Sidecars should not be log batches");
+
+            let record_batch = actions.try_into_record_batch()?;
+            let add = record_batch.column_by_name("add").unwrap();
+            let add_struct = add.as_any().downcast_ref::<StructArray>().unwrap();
+            let path = add_struct
+                .column_by_name("path")
+                .unwrap()
+                .as_any()
+                .downcast_ref::<StringArray>()
+                .unwrap();
+
+            let batch_paths = path.iter().flatten().map(ToString::to_string).collect_vec();
+            sidecar_file_paths.extend(batch_paths);
+        }
+
+        sidecar_file_paths.sort();
+
+        assert_eq!(
+            sidecar_file_paths.len(),
+            101,
+            "CheckpointLeafReader should find exactly 101 files from sidecars"
+        );
+
+        // Verify first few files match expected (sampling to keep test readable)
+        let expected_first_files = vec![
+            "test%25file%25prefix-part-00000-01086c52-1b86-48d0-8889-517fe626849d-c000.snappy.parquet",
+            "test%25file%25prefix-part-00000-0fd71c0e-fd08-4685-87d6-aae77532d3ea-c000.snappy.parquet",
+            "test%25file%25prefix-part-00000-2710dd7f-9fa5-429d-b3fb-c005ba16e062-c000.snappy.parquet",
+        ];
+
+        assert_eq!(
+            &sidecar_file_paths[..3],
+            &expected_first_files[..],
+            "CheckpointLeafReader should process files in expected order"
+        );
+
+        Ok(())
+    }
+}

kernel/src/log_reader/leaf.rs

@@ -1,3 +1,3 @@
+pub(crate) mod checkpoint_leaf;
 pub(crate) mod checkpoint_manifest;
 pub(crate) mod commit;
-pub(crate) mod leaf;