feat: Add logSegment.new_with_commit_appended API (delta-io#1602)

## What changes are proposed in this pull request?

This PR adds a `LogSegment::new_with_commit_appended` API. This consumes the
input LogSegment and returns a new LogSegment with that new commit
(published or staged) added.

Soon, our txn.commit() call will return, grab the FileMeta from the
Committer, and use it to extend the LogSegment. That updated LogSegment
will then be used to create the Post-Commit-Snapshot.

## How was this change tested?

New UTs.

Author: scottsand-db

kernel/src/log_segment.rs

@@ -327,6 +327,45 @@ impl LogSegment {
         LogSegment::try_new(listed_commits, log_root, Some(end_version), None)
     }
 
+    /// Creates a new LogSegment with the given commit file added to the end.
+    /// TODO: Take in multiple commits when Kernel-RS supports txn retries and conflict rebasing.
+    #[allow(unused)]
+    pub(crate) fn new_with_commit_appended(
+        &self,
+        tail_commit_file: ParsedLogPath,
+    ) -> DeltaResult<Self> {
+        require!(
+            tail_commit_file.is_commit(),
+            Error::internal_error(format!(
+                "Cannot extend and create new LogSegment. Tail log file is not a commit file. \
+                Path: {}, Type: {:?}.",
+                tail_commit_file.location.location, tail_commit_file.file_type
+            ))
+        );
+        require!(
+            tail_commit_file.version == self.end_version + 1,
+            Error::internal_error(format!(
+                "Cannot extend and create new LogSegment. Tail commit file version ({}) does not \
+                equal LogSegment end_version ({}) + 1.",
+                tail_commit_file.version, self.end_version
+            ))
+        );
+
+        let mut new_log_segment = self.clone();
+
+        new_log_segment.end_version = tail_commit_file.version;
+        new_log_segment
+            .ascending_commit_files
+            .push(tail_commit_file.clone());
+        new_log_segment.latest_commit_file = Some(tail_commit_file.clone());
+        new_log_segment.max_published_version = match tail_commit_file.file_type {
+            LogPathFileType::Commit => Some(tail_commit_file.version),
+            _ => self.max_published_version,
+        };
+
+        Ok(new_log_segment)
+    }
+
     /// Read a stream of actions from this log segment. This returns an iterator of
     /// [`ActionsBatch`]s which includes EngineData of actions + a boolean flag indicating whether
     /// the data was read from a commit file (true) or a checkpoint file (false).

kernel/src/log_segment/tests.rs

@@ -2582,6 +2582,10 @@ async fn test_get_file_actions_schema_v1_parquet_with_hint() -> DeltaResult<()>
     Ok(())
 }
 
+// ============================================================================
+// max_published_version tests
+// ============================================================================
+
 #[tokio::test]
 async fn test_max_published_version_only_published_commits() {
     let log_segment = create_segment_for(LogSegmentConfig {
@@ -2681,6 +2685,11 @@ async fn test_max_published_version_checkpoint_only() {
     .await;
     assert_eq!(log_segment.max_published_version, None);
 }
+
+// ============================================================================
+// schema_has_compatible_stats_parsed tests
+// ============================================================================
+
 // Helper to create a checkpoint schema with stats_parsed for testing
 fn create_checkpoint_schema_with_stats_parsed(min_values_fields: Vec<StructField>) -> StructType {
     let stats_parsed = StructType::new_unchecked([
@@ -2883,3 +2892,107 @@ fn test_schema_has_compatible_stats_parsed_min_values_not_struct() {
         &stats_schema
     ));
 }
+
+// ============================================================================
+// new_with_commit tests
+// ============================================================================
+
+/// Asserts that `new` is `orig` extended with exactly one commit via `LogSegment::new_with_commit`.
+fn assert_log_segment_extended(orig: LogSegment, new: LogSegment) {
+    // Check: What should have changed
+    assert_eq!(orig.end_version + 1, new.end_version);
+    assert_eq!(
+        orig.ascending_commit_files.len() + 1,
+        new.ascending_commit_files.len()
+    );
+    assert_eq!(
+        orig.latest_commit_file.as_ref().unwrap().version + 1,
+        new.latest_commit_file.as_ref().unwrap().version
+    );
+
+    // Check: What should be the same
+    fn normalize(log_segment: LogSegment) -> LogSegment {
+        LogSegment {
+            end_version: 0,
+            max_published_version: None,
+            ascending_commit_files: vec![],
+            latest_commit_file: None,
+            ..log_segment
+        }
+    }
+
+    assert_eq!(normalize(orig), normalize(new));
+}
+
+#[tokio::test]
+async fn test_new_with_commit_published_commit() {
+    let log_segment = create_segment_for(LogSegmentConfig {
+        published_commit_versions: &[0, 1, 2, 3, 4],
+        ..Default::default()
+    })
+    .await;
+    let table_root = Url::parse("memory:///").unwrap();
+    let new_commit = ParsedLogPath::create_parsed_published_commit(&table_root, 5);
+
+    let new_log_segment = log_segment
+        .clone()
+        .new_with_commit_appended(new_commit)
+        .unwrap();
+
+    assert_eq!(new_log_segment.max_published_version, Some(5));
+    assert_log_segment_extended(log_segment, new_log_segment);
+}
+
+#[tokio::test]
+async fn test_new_with_commit_staged_commit() {
+    let log_segment = create_segment_for(LogSegmentConfig {
+        published_commit_versions: &[0, 1, 2, 3, 4],
+        ..Default::default()
+    })
+    .await;
+    let table_root = Url::parse("memory:///").unwrap();
+    let new_commit = ParsedLogPath::create_parsed_staged_commit(&table_root, 5);
+
+    let new_log_segment = log_segment
+        .clone()
+        .new_with_commit_appended(new_commit)
+        .unwrap();
+
+    assert_eq!(new_log_segment.max_published_version, Some(4));
+    assert_log_segment_extended(log_segment, new_log_segment);
+}
+
+#[tokio::test]
+async fn test_new_with_commit_not_commit_type() {
+    let log_segment = create_segment_for(LogSegmentConfig {
+        published_commit_versions: &[0, 1, 2, 3, 4],
+        ..Default::default()
+    })
+    .await;
+    let checkpoint = create_log_path("file:///_delta_log/00000000000000000005.checkpoint.parquet");
+
+    let result = log_segment.new_with_commit_appended(checkpoint);
+
+    assert_result_error_with_message(
+        result,
+        "Cannot extend and create new LogSegment. Tail log file is not a commit file.",
+    );
+}
+
+#[tokio::test]
+async fn test_new_with_commit_not_end_version_plus_one() {
+    let log_segment = create_segment_for(LogSegmentConfig {
+        published_commit_versions: &[0, 1, 2, 3, 4],
+        ..Default::default()
+    })
+    .await;
+    let table_root = Url::parse("memory:///").unwrap();
+
+    let wrong_version_commit = ParsedLogPath::create_parsed_published_commit(&table_root, 10);
+    let result = log_segment.new_with_commit_appended(wrong_version_commit);
+
+    assert_result_error_with_message(
+        result,
+        "Cannot extend and create new LogSegment. Tail commit file version (10) does not equal LogSegment end_version (4) + 1."
+    );
+}

kernel/src/path.rs

@@ -460,7 +460,7 @@ impl LogRoot {
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
     use std::path::PathBuf;
     use std::sync::Arc;
 
@@ -471,6 +471,39 @@ mod tests {
     use object_store::memory::InMemory;
     use test_utils::add_commit;
 
+    impl ParsedLogPath<FileMeta> {
+        pub(crate) fn create_parsed_published_commit(table_root: &Url, version: Version) -> Self {
+            let filename = format!("{version:020}.json");
+            let location = table_root
+                .join(DELTA_LOG_DIR_WITH_SLASH)
+                .unwrap()
+                .join(&filename)
+                .unwrap();
+            let parsed = ParsedLogPath::try_from(FileMeta::new(location, 0, 0))
+                .unwrap()
+                .unwrap();
+            assert!(parsed.file_type == LogPathFileType::Commit);
+            parsed
+        }
+
+        pub(crate) fn create_parsed_staged_commit(table_root: &Url, version: Version) -> Self {
+            let uuid = Uuid::new_v4();
+            let filename = format!("{version:020}.{uuid}.json");
+            let location = table_root
+                .join(DELTA_LOG_DIR_WITH_SLASH)
+                .unwrap()
+                .join(STAGED_COMMITS_DIR)
+                .unwrap()
+                .join(&filename)
+                .unwrap();
+            let parsed = ParsedLogPath::try_from(FileMeta::new(location, 0, 0))
+                .unwrap()
+                .unwrap();
+            assert!(parsed.file_type == LogPathFileType::StagedCommit);
+            parsed
+        }
+    }
+
     fn table_root_dir_url() -> Url {
         let path = PathBuf::from("./tests/data/table-with-dv-small/");
         let path = std::fs::canonicalize(path).unwrap();