feat: add classic and uuid parquet checkpoint path generation (#782)

sebastiantia · web-flow · commit c84b079a79f4 · 2025-04-07T13:54:29.000-07:00
## What changes are proposed in this pull request?  This PR introduces the helper methods: - `new_uuid_parquet_checkpoint` which creates a new `ParsedCheckpointPath<Url>` for a uuid-named parquet checkpoint file at the specified version. The UUID-naming scheme looks like: `n.checkpoint.u.parquet`, where u is a UUID and n is the snapshot version that this checkpoint represents. - `new_classic_parquet_checkpoint` which creates a new `ParsedCheckpointPath<Url>` for a classic-named parquet checkpoint file at the specified version. The classic-naming scheme looks like: `n.checkpoint.parquet`, where n is the snapshot version that this checkpoint represents. - **Updates the `uuid` dependency to always include `v4` and `fast-rng` features:** - This ensures that `uuid::new_v4()` is always available. - The `fast-rng` feature improves performance when generating UUIDs. For more information on the two checkpoint naming-schemes: https://github.com/delta-io/delta/blob/master/PROTOCOL.md#uuid-named-checkpoint https://github.com/delta-io/delta/blob/master/PROTOCOL.md#classic-checkpoint This PR is part of the on-going effort to implement single-file checkpoint write support. For reference, [[link to write API proposal]](#779)  ## How was this change tested?  - `test_new_uuid_parquet_checkpoint` - verifies UUID-named Parquet checkpoint creation with proper attributes. - `test_new_classic_parquet_checkpoint` - verifies classic-named Parquet checkpoint creation with proper attributes.
diff --git a/kernel/Cargo.toml b/kernel/Cargo.toml
@@ -49,7 +49,7 @@ thiserror = "1"
 # only for structured logging
 tracing = { version = "0.1", features = ["log"] }
 url = "2"
-uuid = "1.10.0"
+uuid = { version = "1.10.0", features = ["v4", "fast-rng"] }
 z85 = "3.0.5"
 
 # bring in our derive macros
@@ -118,8 +118,6 @@ default-engine-base = [
   "need_arrow",
   "object_store",
   "tokio",
-  "uuid/v4",
-  "uuid/fast-rng",
 ]
 
 # the default-engine use the reqwest crate with default features which uses native-tls. if you want
diff --git a/kernel/src/path.rs b/kernel/src/path.rs
@@ -2,6 +2,7 @@
 
 use std::str::FromStr;
 use url::Url;
+use uuid::Uuid;
 
 use crate::{DeltaResult, Error, FileMeta, Version};
 
@@ -180,22 +181,59 @@ impl<Location: AsUrl> ParsedLogPath<Location> {
 }
 
 impl ParsedLogPath<Url> {
-    /// Create a new ParsedCommitPath<Url> for a new json commit file at the specified version
-    pub(crate) fn new_commit(
-        table_root: &Url,
-        version: Version,
-    ) -> DeltaResult<ParsedLogPath<Url>> {
+    const DELTA_LOG_DIR: &'static str = "_delta_log/";
+
+    /// Helper method to create a path with the given filename generator
+    fn create_path(table_root: &Url, filename: String) -> DeltaResult<Self> {
+        let location = table_root.join(Self::DELTA_LOG_DIR)?.join(&filename)?;
+        Self::try_from(location)?.ok_or_else(|| {
+            Error::internal_error(format!("Attempted to create an invalid path: {}", filename))
+        })
+    }
+
+    /// Create a new ParsedCommitPath<Url> for a new json commit file
+    pub(crate) fn new_commit(table_root: &Url, version: Version) -> DeltaResult<Self> {
         let filename = format!("{:020}.json", version);
-        let location = table_root.join("_delta_log/")?.join(&filename)?;
-        let path = Self::try_from(location)?
-            .ok_or_else(|| Error::internal_error("attempted to create invalid commit path"))?;
+        let path = Self::create_path(table_root, filename)?;
         if !path.is_commit() {
             return Err(Error::internal_error(
                 "ParsedLogPath::new_commit created a non-commit path",
             ));
         }
         Ok(path)
     }
+
+    /// Create a new ParsedCheckpointPath<Url> for a classic parquet checkpoint file
+    #[allow(dead_code)] // TODO: Remove this once we have a use case for it
+    pub(crate) fn new_classic_parquet_checkpoint(
+        table_root: &Url,
+        version: Version,
+    ) -> DeltaResult<Self> {
+        let filename = format!("{:020}.checkpoint.parquet", version);
+        let path = Self::create_path(table_root, filename)?;
+        if !path.is_checkpoint() {
+            return Err(Error::internal_error(
+                "ParsedLogPath::new_classic_parquet_checkpoint created a non-checkpoint path",
+            ));
+        }
+        Ok(path)
+    }
+
+    /// Create a new ParsedCheckpointPath<Url> for a UUID-based parquet checkpoint file
+    #[allow(dead_code)] // TODO: Remove this once we have a use case for it
+    pub(crate) fn new_uuid_parquet_checkpoint(
+        table_root: &Url,
+        version: Version,
+    ) -> DeltaResult<Self> {
+        let filename = format!("{:020}.checkpoint.{}.parquet", version, Uuid::new_v4());
+        let path = Self::create_path(table_root, filename)?;
+        if !path.is_checkpoint() {
+            return Err(Error::internal_error(
+                "ParsedLogPath::new_uuid_parquet_checkpoint created a non-checkpoint path",
+            ));
+        }
+        Ok(path)
+    }
 }
 
 #[cfg(test)]
@@ -566,4 +604,42 @@ mod tests {
         assert!(matches!(log_path.file_type, LogPathFileType::Commit));
         assert_eq!(log_path.filename, "00000000000000000010.json");
     }
+
+    #[test]
+    fn test_new_uuid_parquet_checkpoint() {
+        let table_log_dir = table_log_dir_url();
+        let log_path = ParsedLogPath::new_uuid_parquet_checkpoint(&table_log_dir, 10).unwrap();
+
+        assert_eq!(log_path.version, 10);
+        assert!(log_path.is_checkpoint());
+        assert_eq!(log_path.extension, "parquet");
+        if let LogPathFileType::UuidCheckpoint(uuid) = &log_path.file_type {
+            assert_eq!(uuid.len(), UUID_PART_LEN);
+        } else {
+            panic!("Expected UuidCheckpoint file type");
+        }
+
+        let filename = log_path.filename.to_string();
+        let filename_parts: Vec<&str> = filename.split('.').collect();
+        assert_eq!(filename_parts.len(), 4);
+        assert_eq!(filename_parts[0], "00000000000000000010");
+        assert_eq!(filename_parts[1], "checkpoint");
+        assert_eq!(filename_parts[2].len(), UUID_PART_LEN);
+        assert_eq!(filename_parts[3], "parquet");
+    }
+
+    #[test]
+    fn test_new_classic_parquet_checkpoint() {
+        let table_log_dir = table_log_dir_url();
+        let log_path = ParsedLogPath::new_classic_parquet_checkpoint(&table_log_dir, 10).unwrap();
+
+        assert_eq!(log_path.version, 10);
+        assert!(log_path.is_checkpoint());
+        assert_eq!(log_path.extension, "parquet");
+        assert!(matches!(
+            log_path.file_type,
+            LogPathFileType::SinglePartCheckpoint
+        ));
+        assert_eq!(log_path.filename, "00000000000000000010.checkpoint.parquet");
+    }
 }
