feat(catalog-managed): UCCommitter (#1418)

## What changes are proposed in this pull request?
Adds a new `UCCommitter` for writing to tables in UC (in `uc-catalog`
crate for now, final resting place TBD).

Additionally, adds the following RW table features in supported writer
features (did we never enable `V2Checkpoint` write?)
- `VacuumProtocolCheck`
- `V2Checkpoint`
- `CatalogManaged` (behind `catalog-managed` feature flag)
- `CatalogOwnedPreview` (behind `catalog-managed` feature flag)

## How was this change tested?
added an ignored test to write to cc tables; run with `ENDPOINT=".."
TABLENAME=".." TOKEN=".." cargo t write_uc_table --nocapture --
--ignored`

Author: zachschuermann

kernel/src/actions/mod.rs

@@ -590,11 +590,6 @@ impl Protocol {
     /// Check if writing to a table with this protocol is supported. That is: does the kernel
     /// support the specified protocol writer version and all enabled writer features?
     pub(crate) fn ensure_write_supported(&self) -> DeltaResult<()> {
-        #[cfg(feature = "catalog-managed")]
-        require!(
-            !self.is_catalog_managed(),
-            Error::unsupported("Writes are not yet supported for catalog-managed tables")
-        );
         match &self.writer_features {
             Some(writer_features) if self.min_writer_version == 7 => {
                 // if we're on version 7, make sure we support all the specified features
@@ -1610,7 +1605,7 @@ mod tests {
         .unwrap();
         assert_result_error_with_message(
             protocol.ensure_write_supported(),
-            r#"Unsupported: Found unsupported TableFeatures: "identityColumns". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#,
+            r#"Unsupported: Found unsupported TableFeatures: "identityColumns". Supported TableFeatures: "changeDataFeed", "appendOnly", "catalogManaged", "catalogOwned-preview", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck", "variantType", "variantType-preview", "variantShredding-preview""#,
         );
 
         // Unknown writer features are allowed during creation for forward compatibility,
@@ -1624,7 +1619,7 @@ mod tests {
         .unwrap();
         assert_result_error_with_message(
             protocol.ensure_write_supported(),
-            r#"Unsupported: Found unsupported TableFeatures: "unsupported writer". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#,
+            r#"Unsupported: Found unsupported TableFeatures: "unsupported writer". Supported TableFeatures: "changeDataFeed", "appendOnly", "catalogManaged", "catalogOwned-preview", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck", "variantType", "variantType-preview", "variantShredding-preview""#,
         );
     }
 
@@ -1680,24 +1675,25 @@ mod tests {
         assert_eq!(parse_features::<TableFeature>(features), expected);
     }
 
+    #[cfg(feature = "catalog-managed")]
     #[test]
-    fn test_no_catalog_managed_writes() {
+    fn test_catalog_managed_writes() {
         let protocol = Protocol::try_new(
             3,
             7,
             Some([TableFeature::CatalogManaged]),
             Some([TableFeature::CatalogManaged]),
         )
         .unwrap();
-        assert!(protocol.ensure_write_supported().is_err());
+        assert!(protocol.ensure_write_supported().is_ok());
         let protocol = Protocol::try_new(
             3,
             7,
             Some([TableFeature::CatalogOwnedPreview]),
             Some([TableFeature::CatalogOwnedPreview]),
         )
         .unwrap();
-        assert!(protocol.ensure_write_supported().is_err());
+        assert!(protocol.ensure_write_supported().is_ok());
     }
 
     #[test]

kernel/src/committer.rs

@@ -45,12 +45,17 @@ use url::Url;
 pub struct CommitMetadata {
     pub(crate) log_root: LogRoot,
     pub(crate) version: Version,
+    pub(crate) in_commit_timestamp: i64,
     // in the future this will include Protocol, Metadata, CommitInfo, Domain Metadata, etc.
 }
 
 impl CommitMetadata {
-    pub(crate) fn new(log_root: LogRoot, version: Version) -> Self {
-        Self { log_root, version }
+    pub(crate) fn new(log_root: LogRoot, version: Version, in_commit_timestamp: i64) -> Self {
+        Self {
+            log_root,
+            version,
+            in_commit_timestamp,
+        }
     }
 
     /// The commit path is the absolute path (e.g. s3://bucket/table/_delta_log/{version}.json) to
@@ -73,6 +78,16 @@ impl CommitMetadata {
     pub fn version(&self) -> Version {
         self.version
     }
+
+    /// The in-commit timestamp for the commit. Note that this may differ from the actual commit
+    /// file modification time.
+    pub fn in_commit_timestamp(&self) -> i64 {
+        self.in_commit_timestamp
+    }
+
+    pub fn table_root(&self) -> &Url {
+        self.log_root.table_root()
+    }
 }
 
 /// `CommitResponse` is the result of committing a transaction via a catalog. The committer uses
@@ -171,11 +186,14 @@ mod tests {
         let table_root = Url::parse("s3://my-bucket/path/to/table/").unwrap();
         let log_root = LogRoot::new(table_root).unwrap();
         let version = 42;
+        let ts = 1234;
 
-        let commit_metadata = CommitMetadata::new(log_root, version);
+        let commit_metadata = CommitMetadata::new(log_root, version, ts);
 
         // version
         assert_eq!(commit_metadata.version(), 42);
+        // in_commit_timestamp
+        assert_eq!(commit_metadata.in_commit_timestamp(), 1234);
 
         // published commit path
         let published_path = commit_metadata.published_commit_path().unwrap();
@@ -189,8 +207,9 @@ mod tests {
         let staged_path_str = staged_path.as_str();
 
         assert!(
-            staged_path_str
-                .starts_with("s3://my-bucket/path/to/table/_delta_log/00000000000000000042."),
+            staged_path_str.starts_with(
+                "s3://my-bucket/path/to/table/_delta_log/_staged_commits/00000000000000000042."
+            ),
             "Staged path should start with the correct prefix, got: {}",
             staged_path_str
         );
@@ -200,15 +219,17 @@ mod tests {
             staged_path_str
         );
         let uuid_str = staged_path_str
-            .strip_prefix("s3://my-bucket/path/to/table/_delta_log/00000000000000000042.")
+            .strip_prefix(
+                "s3://my-bucket/path/to/table/_delta_log/_staged_commits/00000000000000000042.",
+            )
             .and_then(|s| s.strip_suffix(".json"))
             .expect("Staged path should have expected format");
         uuid::Uuid::parse_str(uuid_str).expect("Staged path should contain a valid UUID");
     }
 
     #[cfg(feature = "catalog-managed")]
     #[tokio::test]
-    async fn catalog_managed_tables_block_transactions() {
+    async fn disallow_filesystem_committer_for_catalog_managed_tables() {
         let storage = Arc::new(InMemory::new());
         let table_root = Url::parse("memory:///").unwrap();
         let engine = DefaultEngine::new(storage.clone());
@@ -225,18 +246,16 @@ mod tests {
         let snapshot = crate::snapshot::SnapshotBuilder::new_for(table_root)
             .build(&engine)
             .unwrap();
-        // Try to create a transaction with FileSystemCommitter
+        // Try to commit a transaction with FileSystemCommitter
         let committer = Box::new(FileSystemCommitter::new());
-        let err = snapshot.transaction(committer).unwrap_err();
+        let err = snapshot
+            .transaction(committer)
+            .unwrap()
+            .commit(&engine)
+            .unwrap_err();
         assert!(matches!(
             err,
-            crate::Error::Unsupported(e) if e.contains("Writes are not yet supported for catalog-managed tables")
+            crate::Error::Generic(e) if e.contains("The FileSystemCommitter cannot be used to commit to catalog-managed tables. Please provide a committer for your catalog via Transaction::with_committer().")
         ));
-        // after allowing writes, we will check that this disallows default committer for
-        // catalog-managed tables.
-        // assert!(matches!(
-        //     err,
-        //     crate::Error::Generic(e) if e.contains("Cannot use the default committer for a catalog-managed table")
-        // ));
     }
 }

kernel/src/log_path.rs

@@ -44,23 +44,27 @@ impl LogPath {
         last_modified: i64,
         size: FileSize,
     ) -> DeltaResult<LogPath> {
+        let commit_path = Self::staged_commit_url(table_root, filename)?;
+        let file_meta = FileMeta {
+            location: commit_path,
+            last_modified,
+            size,
+        };
+        LogPath::try_new(file_meta)
+    }
+
+    /// Create the URL for a staged commit file given the table root and filename. The table_root
+    /// must point to the root of the table and end with a '/'.
+    pub fn staged_commit_url(table_root: Url, filename: &str) -> DeltaResult<Url> {
         // TODO: we should introduce TablePath/LogPath types which enforce checks like ending '/'
         if !table_root.path().ends_with('/') {
             return Err(Error::invalid_table_location(table_root));
         }
-
-        let commit_path = table_root
+        table_root
             .join("_delta_log/")
             .and_then(|url| url.join("_staged_commits/"))
             .and_then(|url| url.join(filename))
-            .map_err(|_| Error::invalid_table_location(table_root))?;
-
-        let file_meta = FileMeta {
-            location: commit_path,
-            last_modified,
-            size,
-        };
-        LogPath::try_new(file_meta)
+            .map_err(|_| Error::invalid_table_location(table_root))
     }
 }
 

kernel/src/path.rs

@@ -23,6 +23,8 @@ const UUID_PART_LEN: usize = 36;
 /// The subdirectory name within the table root where the delta log resides
 const DELTA_LOG_DIR: &str = "_delta_log";
 const DELTA_LOG_DIR_WITH_SLASH: &str = "_delta_log/";
+/// The subdirectory name within the delta log where staged commits reside
+const STAGED_COMMITS_DIR: &str = "_staged_commits/";
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 #[internal_api]
@@ -387,22 +389,40 @@ impl ParsedLogPath<Url> {
 /// A wrapper around parsed log path to provide more structure/safety when handling
 /// table/log/commit paths.
 #[derive(Debug, Clone)]
-pub(crate) struct LogRoot(Url);
+pub(crate) struct LogRoot {
+    table_root: Url,
+    log_root: Url,
+}
 
 impl LogRoot {
     /// Create a new LogRoot from the table root URL (e.g. s3://bucket/table ->
     /// s3://bucket/table/_delta_log/)
     ///
     /// TODO: could take a `table_root: TableRoot`
-    pub(crate) fn new(table_root: Url) -> DeltaResult<Self> {
-        // FIXME: need to check for trailing slash
-        Ok(Self(table_root.join(DELTA_LOG_DIR_WITH_SLASH)?))
+    pub(crate) fn new(mut table_root: Url) -> DeltaResult<Self> {
+        if !table_root.path().ends_with('/') {
+            let new_path = format!("{}/", table_root.path());
+            table_root.set_path(&new_path);
+        }
+        let log_root = table_root.join(DELTA_LOG_DIR_WITH_SLASH)?;
+        Ok(Self {
+            table_root,
+            log_root,
+        })
+    }
+
+    pub(crate) fn table_root(&self) -> &Url {
+        &self.table_root
+    }
+
+    pub(crate) fn log_root(&self) -> &Url {
+        &self.log_root
     }
 
     /// Create a new commit path (absolute path) for the given version.
     pub(crate) fn new_commit_path(&self, version: Version) -> DeltaResult<ParsedLogPath<Url>> {
         let filename = format!("{version:020}.json");
-        let path = self.0.join(&filename)?;
+        let path = self.log_root().join(&filename)?;
         ParsedLogPath::try_from(path)?.ok_or_else(|| {
             Error::internal_error(format!("Attempted to create an invalid path: {filename}"))
         })
@@ -416,7 +436,7 @@ impl LogRoot {
     ) -> DeltaResult<ParsedLogPath<Url>> {
         let uuid = uuid::Uuid::new_v4();
         let filename = format!("{version:020}.{uuid}.json");
-        let path = self.0.join(&filename)?;
+        let path = self.log_root().join(STAGED_COMMITS_DIR)?.join(&filename)?;
         ParsedLogPath::try_from(path)?.ok_or_else(|| {
             Error::internal_error(format!("Attempted to create an invalid path: {filename}"))
         })

kernel/src/table_configuration.rs

@@ -702,15 +702,15 @@ mod test {
                     &["delta.enableChangeDataFeed", "delta.appendOnly"],
                     &[ChangeDataFeed, ColumnMapping, AppendOnly],
                 ),
-                Err(Error::unsupported(r#"Found unsupported TableFeatures: "columnMapping". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#)),
+                Err(Error::unsupported(r#"Found unsupported TableFeatures: "columnMapping". Supported TableFeatures: "changeDataFeed", "appendOnly", "catalogManaged", "catalogOwned-preview", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck", "variantType", "variantType-preview", "variantShredding-preview""#)),
             ),
             (
                 // The table does not require writing CDC files, so it is safe to write to it.
                 create_mock_table_config(
                     &["delta.appendOnly"],
                     &[ChangeDataFeed, ColumnMapping, AppendOnly],
                 ),
-                Err(Error::unsupported(r#"Found unsupported TableFeatures: "columnMapping". Supported TableFeatures: "changeDataFeed", "appendOnly", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "variantType", "variantType-preview", "variantShredding-preview""#)),
+                Err(Error::unsupported(r#"Found unsupported TableFeatures: "columnMapping". Supported TableFeatures: "changeDataFeed", "appendOnly", "catalogManaged", "catalogOwned-preview", "deletionVectors", "domainMetadata", "inCommitTimestamp", "invariants", "rowTracking", "timestampNtz", "v2Checkpoint", "vacuumProtocolCheck", "variantType", "variantType-preview", "variantShredding-preview""#)),
             ),
             (
                 // Should succeed since change data feed is not enabled

kernel/src/table_features/mod.rs

@@ -450,6 +450,9 @@ static CATALOG_MANAGED_INFO: FeatureInfo = FeatureInfo {
     read_support: KernelSupport::Supported,
     #[cfg(not(feature = "catalog-managed"))]
     read_support: KernelSupport::NotSupported,
+    #[cfg(feature = "catalog-managed")]
+    write_support: KernelSupport::Supported,
+    #[cfg(not(feature = "catalog-managed"))]
     write_support: KernelSupport::NotSupported,
     enablement_check: EnablementCheck::AlwaysIfSupported,
 };
@@ -465,6 +468,9 @@ static CATALOG_OWNED_PREVIEW_INFO: FeatureInfo = FeatureInfo {
     read_support: KernelSupport::Supported,
     #[cfg(not(feature = "catalog-managed"))]
     read_support: KernelSupport::NotSupported,
+    #[cfg(feature = "catalog-managed")]
+    write_support: KernelSupport::Supported,
+    #[cfg(not(feature = "catalog-managed"))]
     write_support: KernelSupport::NotSupported,
     enablement_check: EnablementCheck::AlwaysIfSupported,
 };
@@ -554,7 +560,7 @@ static VACUUM_PROTOCOL_CHECK_INFO: FeatureInfo = FeatureInfo {
     feature_type: FeatureType::ReaderWriter,
     feature_requirements: &[],
     read_support: KernelSupport::Supported,
-    write_support: KernelSupport::NotSupported,
+    write_support: KernelSupport::Supported,
     enablement_check: EnablementCheck::AlwaysIfSupported,
 };
 
@@ -717,12 +723,18 @@ pub(crate) static SUPPORTED_WRITER_FEATURES: LazyLock<Vec<TableFeature>> = LazyL
     vec![
         TableFeature::ChangeDataFeed,
         TableFeature::AppendOnly,
+        #[cfg(feature = "catalog-managed")]
+        TableFeature::CatalogManaged,
+        #[cfg(feature = "catalog-managed")]
+        TableFeature::CatalogOwnedPreview,
         TableFeature::DeletionVectors,
         TableFeature::DomainMetadata,
         TableFeature::InCommitTimestamp,
         TableFeature::Invariants,
         TableFeature::RowTracking,
         TableFeature::TimestampWithoutTimezone,
+        TableFeature::V2Checkpoint,
+        TableFeature::VacuumProtocolCheck,
         TableFeature::VariantType,
         TableFeature::VariantTypePreview,
         TableFeature::VariantShreddingPreview,

kernel/src/transaction/mod.rs

@@ -316,7 +316,7 @@ impl Transaction {
             return Err(Error::generic("The FileSystemCommitter cannot be used to commit to catalog-managed tables. Please provide a committer for your catalog via Transaction::with_committer()."));
         }
         let log_root = LogRoot::new(self.read_snapshot.table_root().clone())?;
-        let commit_metadata = CommitMetadata::new(log_root, commit_version);
+        let commit_metadata = CommitMetadata::new(log_root, commit_version, self.commit_timestamp);
         match self
             .committer
             .commit(engine, Box::new(filtered_actions), commit_metadata)

uc-catalog/Cargo.toml

@@ -23,6 +23,7 @@ serde_json = "1.0"
 tokio = { version = "1", features = ["full"] }
 tracing = "0.1"
 url = "2"
+uuid = { version = "1", features = ["v4"] }
 
 [dev-dependencies]
 delta_kernel = { path = "../kernel", features = ["arrow-56", "default-engine-rustls", "catalog-managed"] }