refactor: harmonize checkpoint and log compaction iterators (#1436)

## What changes are proposed in this pull request?

Log replay for checkpoint writes and log compaction writes uses
physically separate but logically identical iterator implementations.
Eliminate the duplication by creating one iterator type that both can
use.

## Breaking changes

The two replaced iterators were pub

## How was this change tested?

Existing unit tests.

---------

Co-authored-by: Nick Lanham <[email protected]>

Author: scovich

kernel/src/action_reconciliation/log_replay.rs

@@ -64,6 +64,12 @@ pub(crate) struct ActionReconciliationProcessor {
 ///
 /// It contains the filtered batch of actions to be included, along with statistics about the
 /// number of actions filtered for inclusion.
+///
+/// # Warning
+///
+/// This iterator must be fully consumed to ensure proper collection of statistics. Additionally,
+/// all yielded data must be written to the specified path before e.g. calling
+/// [`CheckpointWriter::finalize`]. Failing to do so may result in data loss or corruption.
 pub(crate) struct ActionReconciliationBatch {
     /// The filtered batch of actions.
     pub(crate) filtered_data: FilteredEngineData,
@@ -79,6 +85,81 @@ impl HasSelectionVector for ActionReconciliationBatch {
     }
 }
 
+/// Iterator over action reconciliation data.
+///
+/// This iterator yields a stream of [`FilteredEngineData`] items while, tracking action
+/// counts. Used by both checkpoint and log compaction workflows.
+pub struct ActionReconciliationIterator {
+    inner: Box<dyn Iterator<Item = DeltaResult<ActionReconciliationBatch>> + Send>,
+    actions_count: i64,
+    add_actions_count: i64,
+    is_exhausted: bool,
+}
+
+impl ActionReconciliationIterator {
+    /// Create a new iterator with counters initialized to 0
+    pub(crate) fn new(
+        inner: Box<dyn Iterator<Item = DeltaResult<ActionReconciliationBatch>> + Send>,
+    ) -> Self {
+        Self {
+            inner,
+            actions_count: 0,
+            add_actions_count: 0,
+            is_exhausted: false,
+        }
+    }
+
+    /// True if this iterator has been exhausted (ie all batches have been processed)
+    pub(crate) fn is_exhausted(&self) -> bool {
+        self.is_exhausted
+    }
+
+    /// Get the total number of actions processed so far
+    pub(crate) fn actions_count(&self) -> i64 {
+        self.actions_count
+    }
+
+    /// Get the total number of add actions processed so far
+    pub(crate) fn add_actions_count(&self) -> i64 {
+        self.add_actions_count
+    }
+
+    /// Helper to transform a batch: update metrics and extract filtered data
+    fn transform_batch(
+        &mut self,
+        batch: Option<DeltaResult<ActionReconciliationBatch>>,
+    ) -> Option<DeltaResult<FilteredEngineData>> {
+        let Some(batch) = batch else {
+            self.is_exhausted = true;
+            return None;
+        };
+        Some(batch.map(|batch| {
+            self.actions_count += batch.actions_count;
+            self.add_actions_count += batch.add_actions_count;
+            batch.filtered_data
+        }))
+    }
+}
+
+impl std::fmt::Debug for ActionReconciliationIterator {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ActionReconciliationIterator")
+            .field("actions_count", &self.actions_count)
+            .field("add_actions_count", &self.add_actions_count)
+            .field("is_exhausted", &self.is_exhausted)
+            .finish()
+    }
+}
+
+impl Iterator for ActionReconciliationIterator {
+    type Item = DeltaResult<FilteredEngineData>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        let batch = self.inner.next();
+        self.transform_batch(batch)
+    }
+}
+
 impl LogReplayProcessor for ActionReconciliationProcessor {
     type Output = ActionReconciliationBatch;
 

kernel/src/action_reconciliation/mod.rs

@@ -23,6 +23,8 @@ use crate::{DeltaResult, Error};
 
 pub(crate) mod log_replay;
 
+pub use log_replay::ActionReconciliationIterator;
+
 const SECONDS_PER_MINUTE: u64 = 60;
 const MINUTES_PER_HOUR: u64 = 60;
 const HOURS_PER_DAY: u64 = 24;

kernel/src/checkpoint/mod.rs

@@ -16,7 +16,7 @@
 //! ## Architecture
 //!
 //! - [`CheckpointWriter`] - Core component that manages the checkpoint creation workflow
-//! - [`CheckpointDataIterator`] - Iterator over the checkpoint data to be written
+//! - [`ActionReconciliationIterator`] - Iterator over the checkpoint data to be written
 //!
 //! ## Usage
 //!
@@ -31,7 +31,7 @@
 //!
 //! ```no_run
 //! # use std::sync::Arc;
-//! # use delta_kernel::checkpoint::CheckpointDataIterator;
+//! # use delta_kernel::ActionReconciliationIterator;
 //! # use delta_kernel::checkpoint::CheckpointWriter;
 //! # use delta_kernel::Engine;
 //! # use delta_kernel::Snapshot;
@@ -40,7 +40,7 @@
 //! # use delta_kernel::Error;
 //! # use delta_kernel::FileMeta;
 //! # use url::Url;
-//! fn write_checkpoint_file(path: Url, data: &CheckpointDataIterator) -> DeltaResult<FileMeta> {
+//! fn write_checkpoint_file(path: Url, data: &ActionReconciliationIterator) -> DeltaResult<FileMeta> {
 //!     todo!() /* engine-specific logic to write data to object storage*/
 //! }
 //!
@@ -89,7 +89,7 @@ use std::sync::{Arc, LazyLock};
 use crate::action_reconciliation::log_replay::{
     ActionReconciliationBatch, ActionReconciliationProcessor,
 };
-use crate::action_reconciliation::RetentionCalculator;
+use crate::action_reconciliation::{ActionReconciliationIterator, RetentionCalculator};
 use crate::actions::{
     Add, Metadata, Protocol, Remove, SetTransaction, Sidecar, ADD_NAME, CHECKPOINT_METADATA_NAME,
     METADATA_NAME, PROTOCOL_NAME, REMOVE_NAME, SET_TRANSACTION_NAME, SIDECAR_NAME,
@@ -145,43 +145,6 @@ static CHECKPOINT_METADATA_ACTION_SCHEMA: LazyLock<SchemaRef> = LazyLock::new(||
     )]))
 });
 
-/// An iterator over the checkpoint data to be written to the file.
-///
-/// This iterator yields filtered checkpoint data batches ([`FilteredEngineData`]) and
-/// tracks action statistics required for finalizing the checkpoint.
-///
-/// # Warning
-/// The [`CheckpointDataIterator`] must be fully consumed to ensure proper collection of statistics for
-/// the checkpoint. Additionally, all yielded data must be written to the specified path before calling
-/// [`CheckpointWriter::finalize`]. Failing to do so may result in data loss or corruption.
-pub struct CheckpointDataIterator {
-    /// The nested iterator that yields checkpoint batches with action counts
-    checkpoint_batch_iterator:
-        Box<dyn Iterator<Item = DeltaResult<ActionReconciliationBatch>> + Send>,
-    /// Running total of actions included in the checkpoint
-    actions_count: i64,
-    /// Running total of add actions included in the checkpoint
-    add_actions_count: i64,
-}
-
-impl Iterator for CheckpointDataIterator {
-    type Item = DeltaResult<FilteredEngineData>;
-
-    /// Advances the iterator and returns the next value.
-    ///
-    /// This implementation transforms the `ActionReconciliationBatch` items from the nested iterator into
-    /// [`FilteredEngineData`] items for the engine to write, while accumulating action counts from
-    /// each batch. The [`CheckpointDataIterator`] is passed back to the kernel on call to
-    /// [`CheckpointWriter::finalize`] for counts to be read and written to the `_last_checkpoint` file
-    fn next(&mut self) -> Option<Self::Item> {
-        Some(self.checkpoint_batch_iterator.next()?.map(|batch| {
-            self.actions_count += batch.actions_count;
-            self.add_actions_count += batch.add_actions_count;
-            batch.filtered_data
-        }))
-    }
-}
-
 /// Orchestrates the process of creating a checkpoint for a table.
 ///
 /// The [`CheckpointWriter`] is the entry point for generating checkpoint data for a Delta table.
@@ -253,7 +216,7 @@ impl CheckpointWriter {
     /// # Parameters
     /// - `engine`: Implementation of [`Engine`] APIs.
     ///
-    /// # Returns: [`CheckpointDataIterator`] containing the checkpoint data
+    /// # Returns: [`ActionReconciliationIterator`] containing the checkpoint data
     // This method is the core of the checkpoint generation process. It:
     // 1. Determines whether to write a V1 or V2 checkpoint based on the table's
     //    `v2Checkpoints` feature support
@@ -262,7 +225,10 @@ impl CheckpointWriter {
     // 4. Chains the checkpoint metadata action if writing a V2 spec checkpoint
     //    (i.e., if `v2Checkpoints` feature is supported by table)
     // 5. Generates the appropriate checkpoint path
-    pub fn checkpoint_data(&self, engine: &dyn Engine) -> DeltaResult<CheckpointDataIterator> {
+    pub fn checkpoint_data(
+        &self,
+        engine: &dyn Engine,
+    ) -> DeltaResult<ActionReconciliationIterator> {
         let is_v2_checkpoints_supported = self
             .snapshot
             .table_configuration()
@@ -284,12 +250,10 @@ impl CheckpointWriter {
         let checkpoint_metadata =
             is_v2_checkpoints_supported.then(|| self.create_checkpoint_metadata_batch(engine));
 
-        // Wrap the iterator in a CheckpointDataIterator to track action counts
-        Ok(CheckpointDataIterator {
-            checkpoint_batch_iterator: Box::new(checkpoint_data.chain(checkpoint_metadata)),
-            actions_count: 0,
-            add_actions_count: 0,
-        })
+        // Wrap the iterator to track action counts
+        Ok(ActionReconciliationIterator::new(Box::new(
+            checkpoint_data.chain(checkpoint_metadata),
+        )))
     }
 
     /// Finalizes checkpoint creation by saving metadata about the checkpoint.
@@ -313,10 +277,10 @@ impl CheckpointWriter {
         self,
         engine: &dyn Engine,
         metadata: &FileMeta,
-        mut checkpoint_data: CheckpointDataIterator,
+        checkpoint_data: ActionReconciliationIterator,
     ) -> DeltaResult<()> {
         // Ensure the checkpoint data iterator is fully exhausted
-        if checkpoint_data.checkpoint_batch_iterator.next().is_some() {
+        if !checkpoint_data.is_exhausted() {
             return Err(Error::checkpoint_write(
                 "The checkpoint data iterator must be fully consumed and written to storage before calling finalize"
             ));
@@ -332,8 +296,8 @@ impl CheckpointWriter {
         let data = create_last_checkpoint_data(
             engine,
             self.version,
-            checkpoint_data.actions_count,
-            checkpoint_data.add_actions_count,
+            checkpoint_data.actions_count(),
+            checkpoint_data.add_actions_count(),
             size_in_bytes,
         );
 

kernel/src/lib.rs

@@ -148,12 +148,12 @@ pub mod history_manager;
 #[cfg(not(feature = "internal-api"))]
 pub(crate) mod history_manager;
 
-pub use crate::engine_data::FilteredEngineData;
+pub use action_reconciliation::ActionReconciliationIterator;
 pub use delta_kernel_derive;
-pub use engine_data::{EngineData, RowVisitor};
+pub use engine_data::{EngineData, FilteredEngineData, RowVisitor};
 pub use error::{DeltaResult, Error};
 pub use expressions::{Expression, ExpressionRef, Predicate, PredicateRef};
-pub use log_compaction::{should_compact, LogCompactionDataIterator, LogCompactionWriter};
+pub use log_compaction::{should_compact, LogCompactionWriter};
 pub use snapshot::Snapshot;
 pub use snapshot::SnapshotRef;
 

kernel/src/log_compaction/mod.rs

@@ -27,12 +27,12 @@
 //!
 //! ```no_run
 //! # use std::sync::Arc;
-//! # use delta_kernel::{LogCompactionDataIterator, LogCompactionWriter};
+//! # use delta_kernel::{ActionReconciliationIterator, LogCompactionWriter};
 //! # use delta_kernel::{Engine, Snapshot, DeltaResult, Error, FileMeta};
 //! # use url::Url;
 //!
 //! // Engine-specific function to write compaction data
-//! fn write_compaction_file(path: &Url, data: LogCompactionDataIterator) -> DeltaResult<FileMeta> {
+//! fn write_compaction_file(path: &Url, data: ActionReconciliationIterator) -> DeltaResult<FileMeta> {
 //!     // In a real implementation, this would write the data to cloud storage
 //!     todo!("Write data batches to storage at path: {}", path)
 //! }
@@ -84,7 +84,7 @@ use crate::schema::{SchemaRef, StructField, StructType, ToSchema as _};
 
 mod writer;
 
-pub use writer::{should_compact, LogCompactionDataIterator, LogCompactionWriter};
+pub use writer::{should_compact, LogCompactionWriter};
 
 #[cfg(test)]
 mod tests;

kernel/src/log_compaction/tests.rs

@@ -107,12 +107,12 @@ fn test_compaction_data() {
     let iterator = result.unwrap();
 
     // Test iterator methods
-    assert_eq!(iterator.total_actions(), 0);
-    assert_eq!(iterator.total_add_actions(), 0);
+    assert_eq!(iterator.actions_count(), 0);
+    assert_eq!(iterator.add_actions_count(), 0);
 
     // Test debug implementation
     let debug_str = format!("{:?}", iterator);
-    assert!(debug_str.contains("LogCompactionDataIterator"));
+    assert!(debug_str.contains("ActionReconciliationIterator"));
     assert!(debug_str.contains("actions_count"));
     assert!(debug_str.contains("add_actions_count"));
 }
@@ -152,8 +152,8 @@ fn test_compaction_data_with_actual_iterator() {
     let mut iterator = writer.compaction_data(&engine).unwrap();
 
     let mut batch_count = 0;
-    let initial_actions = iterator.total_actions();
-    let initial_add_actions = iterator.total_add_actions();
+    let initial_actions = iterator.actions_count();
+    let initial_add_actions = iterator.add_actions_count();
 
     // Both should start at 0
     assert_eq!(initial_actions, 0);
@@ -164,8 +164,8 @@ fn test_compaction_data_with_actual_iterator() {
         assert!(batch_result.is_ok());
 
         // After processing some batches, the counts should be >= the initial counts
-        assert!(iterator.total_actions() >= initial_actions);
-        assert!(iterator.total_add_actions() >= initial_add_actions);
+        assert!(iterator.actions_count() >= initial_actions);
+        assert!(iterator.add_actions_count() >= initial_add_actions);
     }
 
     assert!(batch_count > 0, "Expected to process at least one batch");
@@ -223,8 +223,8 @@ fn test_version_filtering() {
         );
 
         let iterator = result.unwrap();
-        assert!(iterator.total_actions() >= 0);
-        assert!(iterator.total_add_actions() >= 0);
+        assert!(iterator.actions_count() >= 0);
+        assert!(iterator.add_actions_count() >= 0);
     }
 }