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 <[email protected]>

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,
         }
     }
 }

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 {

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()