From b6558d0b6829369158842f480a7f46a79c8f94fe Mon Sep 17 00:00:00 2001
From: Qianqian <130200611+Sevenannn@users.noreply.github.com>
Date: Mon, 16 Dec 2024 16:50:08 -0800
Subject: [PATCH] Create InvalidCheckpoint Error to bucket errors relevant to
 checkpoint (#593)

<!--
Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, please read our contributor guidelines:
https://github.com/delta-incubator/delta-kernel-rs/blob/main/CONTRIBUTING.md
2. Run `cargo t --all-features --all-targets` to get started testing,
and run `cargo fmt`.
  3. Ensure you have added or run the appropriate tests for your PR.
4. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]
Your PR title ...'.
  5. Be sure to keep the PR description updated to reflect all changes.
-->

## What changes are proposed in this pull request?
Create a new variation of error: InvalidCheckpoint error to bucket
errors relevant to invalid checkpoint files. This will allow for a more
flexible handling of checkpoint related errors when using delta kernel
crate.

## How was this change tested?
Via `cargo t --all-features --all-targets`

---------

Co-authored-by: Zach Schuermann <zachary.zvs@gmail.com>
---
 ffi/src/error.rs          | 2 ++
 kernel/src/error.rs       | 8 ++++++++
 kernel/src/log_segment.rs | 6 +++---
 3 files changed, 13 insertions(+), 3 deletions(-)

diff --git a/ffi/src/error.rs b/ffi/src/error.rs
index 629cac619..a615d0330 100644
--- a/ffi/src/error.rs
+++ b/ffi/src/error.rs
@@ -51,6 +51,7 @@ pub enum KernelError {
     ParseIntervalError,
     ChangeDataFeedUnsupported,
     ChangeDataFeedIncompatibleSchema,
+    InvalidCheckpoint,
 }
 
 impl From<Error> for KernelError {
@@ -108,6 +109,7 @@ impl From<Error> for KernelError {
             Error::ChangeDataFeedIncompatibleSchema(_, _) => {
                 KernelError::ChangeDataFeedIncompatibleSchema
             }
+            Error::InvalidCheckpoint(_) => KernelError::InvalidCheckpoint,
         }
     }
 }
diff --git a/kernel/src/error.rs b/kernel/src/error.rs
index 7fb6e5d60..d3e5e53c9 100644
--- a/kernel/src/error.rs
+++ b/kernel/src/error.rs
@@ -191,6 +191,10 @@ pub enum Error {
 
     #[error("Change data feed encountered incompatible schema. Expected {0}, got {1}")]
     ChangeDataFeedIncompatibleSchema(String, String),
+
+    /// Invalid checkpoint files
+    #[error("Invalid Checkpoint: {0}")]
+    InvalidCheckpoint(String),
 }
 
 // Convenience constructors for Error types that take a String argument
@@ -264,6 +268,10 @@ impl Error {
         Self::ChangeDataFeedIncompatibleSchema(format!("{expected:?}"), format!("{actual:?}"))
     }
 
+    pub fn invalid_checkpoint(msg: impl ToString) -> Self {
+        Self::InvalidCheckpoint(msg.to_string())
+    }
+
     // Capture a backtrace when the error is constructed.
     #[must_use]
     pub fn with_backtrace(self) -> Self {
diff --git a/kernel/src/log_segment.rs b/kernel/src/log_segment.rs
index c7567a087..24d78a986 100644
--- a/kernel/src/log_segment.rs
+++ b/kernel/src/log_segment.rs
@@ -68,7 +68,7 @@ impl LogSegment {
         {
             require!(
                 checkpoint_file.version + 1 == commit_file.version,
-                Error::generic(format!(
+                Error::InvalidCheckpoint(format!(
                     "Gap between checkpoint version {} and next commit {}",
                     checkpoint_file.version, commit_file.version,
                 ))
@@ -358,7 +358,7 @@ fn list_log_files_with_checkpoint(
 
     let Some(latest_checkpoint) = checkpoint_parts.last() else {
         // TODO: We could potentially recover here
-        return Err(Error::generic(
+        return Err(Error::invalid_checkpoint(
             "Had a _last_checkpoint hint but didn't find any checkpoints",
         ));
     };
@@ -369,7 +369,7 @@ fn list_log_files_with_checkpoint(
             latest_checkpoint.version
         );
     } else if checkpoint_parts.len() != checkpoint_metadata.parts.unwrap_or(1) {
-        return Err(Error::Generic(format!(
+        return Err(Error::InvalidCheckpoint(format!(
             "_last_checkpoint indicated that checkpoint should have {} parts, but it has {}",
             checkpoint_metadata.parts.unwrap_or(1),
             checkpoint_parts.len()