propogate error and add test

Author: sebastiantia

kernel/src/log_segment.rs

@@ -9,7 +9,7 @@ use crate::utils::require;
 use crate::{
     DeltaResult, Engine, EngineData, Error, Expression, ExpressionRef, FileSystemClient, Version,
 };
-use itertools::Itertools;
+use itertools::{process_results, Itertools};
 use std::convert::identity;
 use std::sync::{Arc, LazyLock};
 use tracing::warn;
@@ -310,36 +310,25 @@ fn list_log_files_with_version(
     // on config at some point
     let mut commit_files = Vec::with_capacity(10);
     let mut checkpoint_parts = vec![];
-    let mut max_checkpoint_version = start_version;
 
     let log_files = list_log_files(fs_client, log_root, start_version, end_version)?;
 
-    for (version, files) in &log_files
-        .filter_map(|res| match res {
-            Ok(path) => Some(path),
-            Err(e) => {
-                warn!("Error processing path: {:?}", e);
-                None
+    process_results(log_files, |iter| {
+        let log_files = iter.chunk_by(move |x| x.version);
+        for (version, files) in &log_files {
+            let mut new_checkpoint_parts = vec![];
+            for file in files {
+                if file.is_commit() {
+                    commit_files.push(file);
+                } else if file.is_checkpoint() {
+                    new_checkpoint_parts.push(file);
+                }
             }
-        })
-        .chunk_by(|path| path.version)
-    {
-        let mut new_checkpoint_parts = vec![];
-
-        for file in files {
-            if file.is_commit() {
-                commit_files.push(file);
-            } else if file.is_checkpoint() {
-                new_checkpoint_parts.push(file);
+            if validate_checkpoint_parts(version, &new_checkpoint_parts) {
+                checkpoint_parts = new_checkpoint_parts;
             }
         }
-        if validate_checkpoint_parts(version, &new_checkpoint_parts)
-            && (max_checkpoint_version.is_none() || Some(version) >= max_checkpoint_version)
-        {
-            max_checkpoint_version = Some(version);
-            checkpoint_parts = new_checkpoint_parts;
-        }
-    }
+    })?;
 
     Ok((commit_files, checkpoint_parts))
 }
@@ -385,10 +374,6 @@ fn list_log_files_with_checkpoint(
 /// Validates that all the checkpoint parts belong to the same checkpoint version and that all parts
 /// are present. Returns `true` if we have a complete checkpoint, `false` otherwise.
 fn validate_checkpoint_parts(version: u64, checkpoint_parts: &[ParsedLogPath]) -> bool {
-    if checkpoint_parts.is_empty() {
-        return false;
-    }
-
     match checkpoint_parts.last().map(|file| &file.file_type) {
         Some(LogPathFileType::MultiPartCheckpoint { num_parts, .. }) => {
             if *num_parts as usize != checkpoint_parts.len() {
@@ -411,8 +396,29 @@ fn validate_checkpoint_parts(version: u64, checkpoint_parts: &[ParsedLogPath]) -
                 return false;
             }
         }
-        // TODO: Include UuidCheckpoint once we actually support v2 checkpoints
-        _ => {}
+        Some(LogPathFileType::UuidCheckpoint(_)) => {
+            warn!(
+                "Found a UUID checkpoint at version {} when it is not supported",
+                version
+            );
+            return false;
+        }
+        Some(LogPathFileType::Commit) | Some(LogPathFileType::CompactedCommit { .. }) => {
+            warn!(
+                "Found a commit file at version {} when expecting a checkpoint",
+                version
+            );
+            return false;
+        }
+        Some(LogPathFileType::Unknown) => {
+            warn!(
+                "Found an unknown file type at version {} when expecting a checkpoint",
+                version
+            );
+            return false;
+        }
+        // No checkpoint parts
+        None => return false,
     }
 
     true

kernel/src/log_segment/tests.rs

@@ -107,6 +107,35 @@ fn build_log_with_paths_and_checkpoint(
     (Box::new(client), log_root)
 }
 
+#[test]
+fn build_snapshot_with_unsupported_uuid_checkpoint() {
+    let (client, log_root) = build_log_with_paths_and_checkpoint(
+        &[
+            delta_path_for_version(0, "json"),
+            delta_path_for_version(1, "checkpoint.parquet"),
+            delta_path_for_version(2, "json"),
+            delta_path_for_version(3, "checkpoint.parquet"),
+            delta_path_for_version(4, "json"),
+            delta_path_for_version(5, "json"),
+            delta_path_for_version(5, "checkpoint.3a0d65cd-4056-49b8-937b-95f9e3ee90e5.parquet"),
+            delta_path_for_version(6, "json"),
+            delta_path_for_version(7, "json"),
+        ],
+        None,
+    );
+
+    let log_segment = LogSegment::for_snapshot(client.as_ref(), log_root, None, None).unwrap();
+    let commit_files = log_segment.ascending_commit_files;
+    let checkpoint_parts = log_segment.checkpoint_parts;
+
+    assert_eq!(checkpoint_parts.len(), 1);
+    assert_eq!(checkpoint_parts[0].version, 3);
+
+    let versions = commit_files.into_iter().map(|x| x.version).collect_vec();
+    let expected_versions = vec![4, 5, 6, 7];
+    assert_eq!(versions, expected_versions);
+}
+
 #[test]
 fn build_snapshot_with_out_of_date_last_checkpoint() {
     let checkpoint_metadata = CheckpointMetadata {
@@ -331,6 +360,7 @@ fn build_snapshot_with_out_of_date_last_checkpoint_and_incomplete_recent_checkpo
     let checkpoint_parts = log_segment.checkpoint_parts;
 
     assert_eq!(checkpoint_parts.len(), 1);
+    assert_eq!(checkpoint_parts[0].version, 3);
 
     let versions = commit_files.into_iter().map(|x| x.version).collect_vec();
     let expected_versions = vec![4, 5, 6, 7];