feat(rust): post commit hook (v2), create checkpoint hook

crates/core/src/operations/transaction/mod.rs

@@ -57,24 +57,33 @@
 //!                      │                                    
 //!                      ▼                                    
 //!       ┌───────────────────────────────┐                   
+//!       │          Post Commit          │                   
+//!       │                               │                   
+//!       │ Commit that was materialized  │                   
+//!       │ to storage with post commit   │                   
+//!       │      hooks to be executed     │                   
+//!       └──────────────┬────────────────┘                 
+//!                      │                                    
+//!                      ▼    
+//!       ┌───────────────────────────────┐                   
 //!       │        Finalized Commit       │                   
 //!       │                               │                   
 //!       │ Commit that was materialized  │                   
 //!       │         to storage            │                   
 //!       │                               │                   
-//!       └───────────────────────────────┘                   
+//!       └───────────────────────────────┘           
 //!</pre>
 
-use std::collections::HashMap;
-
 use chrono::Utc;
 use conflict_checker::ConflictChecker;
 use futures::future::BoxFuture;
 use object_store::path::Path;
 use object_store::{Error as ObjectStoreError, ObjectStore};
 use serde_json::Value;
+use std::collections::HashMap;
 
 use self::conflict_checker::{CommitConflictError, TransactionInfo, WinningCommitSummary};
+use crate::checkpoints::create_checkpoint_for;
 use crate::errors::DeltaTableError;
 use crate::kernel::{
     Action, CommitInfo, EagerSnapshot, Metadata, Protocol, ReaderFeatures, WriterFeatures,
@@ -293,19 +302,27 @@ impl CommitData {
     }
 }
 
+#[derive(Clone, Debug, Copy)]
+/// Properties for post commit hook.
+pub struct PostCommitHookProperties {
+    create_checkpoint: bool,
+}
+
 #[derive(Clone, Debug)]
 /// End user facing interface to be used by operations on the table.
 /// Enable controling commit behaviour and modifying metadata that is written during a commit.
 pub struct CommitProperties {
     pub(crate) app_metadata: HashMap<String, Value>,
     max_retries: usize,
+    create_checkpoint: bool,
 }
 
 impl Default for CommitProperties {
     fn default() -> Self {
         Self {
             app_metadata: Default::default(),
             max_retries: DEFAULT_RETRIES,
+            create_checkpoint: true,
         }
     }
 }
@@ -319,13 +336,23 @@ impl CommitProperties {
         self.app_metadata = HashMap::from_iter(metadata);
         self
     }
+
+    /// Specify if it should create a checkpoint when the commit interval condition is met
+    pub fn with_create_checkpoint(mut self, create_checkpoint: bool) -> Self {
+        self.create_checkpoint = create_checkpoint;
+        self
+    }
 }
 
 impl From<CommitProperties> for CommitBuilder {
     fn from(value: CommitProperties) -> Self {
         CommitBuilder {
             max_retries: value.max_retries,
             app_metadata: value.app_metadata,
+            post_commit_hook: PostCommitHookProperties {
+                create_checkpoint: value.create_checkpoint,
+            }
+            .into(),
             ..Default::default()
         }
     }
@@ -336,6 +363,7 @@ pub struct CommitBuilder {
     actions: Vec<Action>,
     app_metadata: HashMap<String, Value>,
     max_retries: usize,
+    post_commit_hook: Option<PostCommitHookProperties>,
 }
 
 impl Default for CommitBuilder {
@@ -344,6 +372,7 @@ impl Default for CommitBuilder {
             actions: Vec::new(),
             app_metadata: HashMap::new(),
             max_retries: DEFAULT_RETRIES,
+            post_commit_hook: None,
         }
     }
 }
@@ -367,6 +396,12 @@ impl<'a> CommitBuilder {
         self
     }
 
+    /// Specify all the post commit hook properties
+    pub fn with_post_commit_hook(mut self, post_commit_hook: PostCommitHookProperties) -> Self {
+        self.post_commit_hook = post_commit_hook.into();
+        self
+    }
+
     /// Prepare a Commit operation using the configured builder
     pub fn build(
         self,
@@ -380,6 +415,7 @@ impl<'a> CommitBuilder {
             table_data,
             max_retries: self.max_retries,
             data,
+            post_commit_hook: self.post_commit_hook,
         })
     }
 }
@@ -390,6 +426,7 @@ pub struct PreCommit<'a> {
     table_data: Option<&'a dyn TableReference>,
     data: CommitData,
     max_retries: usize,
+    post_commit_hook: Option<PostCommitHookProperties>,
 }
 
 impl<'a> std::future::IntoFuture for PreCommit<'a> {
@@ -399,7 +436,7 @@ impl<'a> std::future::IntoFuture for PreCommit<'a> {
     fn into_future(self) -> Self::IntoFuture {
         let this = self;
 
-        Box::pin(async move { this.into_prepared_commit_future().await?.await })
+        Box::pin(async move { this.into_prepared_commit_future().await?.await?.await })
     }
 }
 
@@ -429,6 +466,7 @@ impl<'a> PreCommit<'a> {
                 table_data: this.table_data,
                 max_retries: this.max_retries,
                 data: this.data,
+                post_commit: this.post_commit_hook,
             })
         })
     }
@@ -441,6 +479,7 @@ pub struct PreparedCommit<'a> {
     data: CommitData,
     table_data: Option<&'a dyn TableReference>,
     max_retries: usize,
+    post_commit: Option<PostCommitHookProperties>,
 }
 
 impl<'a> PreparedCommit<'a> {
@@ -451,7 +490,7 @@ impl<'a> PreparedCommit<'a> {
 }
 
 impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
-    type Output = DeltaResult<FinalizedCommit>;
+    type Output = DeltaResult<PostCommit<'a>>;
     type IntoFuture = BoxFuture<'a, Self::Output>;
 
     fn into_future(self) -> Self::IntoFuture {
@@ -462,9 +501,12 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
 
             if this.table_data.is_none() {
                 this.log_store.write_commit_entry(0, tmp_commit).await?;
-                return Ok(FinalizedCommit {
+                return Ok(PostCommit {
                     version: 0,
                     data: this.data,
+                    create_checkpoint: false,
+                    log_store: this.log_store,
+                    table_data: this.table_data,
                 });
             }
 
@@ -483,10 +525,16 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
                 let version = read_snapshot.version() + attempt_number as i64;
                 match this.log_store.write_commit_entry(version, tmp_commit).await {
                     Ok(()) => {
-                        return Ok(FinalizedCommit {
+                        return Ok(PostCommit {
                             version,
                             data: this.data,
-                        })
+                            create_checkpoint: this
+                                .post_commit
+                                .map(|v| v.create_checkpoint)
+                                .unwrap_or_default(),
+                            log_store: this.log_store,
+                            table_data: this.table_data,
+                        });
                     }
                     Err(TransactionError::VersionAlreadyExists(version)) => {
                         let summary = WinningCommitSummary::try_new(
@@ -534,6 +582,54 @@ impl<'a> std::future::IntoFuture for PreparedCommit<'a> {
     }
 }
 
+/// Represents items for the post commit hook
+pub struct PostCommit<'a> {
+    /// The winning version number of the commit
+    pub version: i64,
+    /// The data that was comitted to the log store
+    pub data: CommitData,
+    create_checkpoint: bool,
+    log_store: LogStoreRef,
+    table_data: Option<&'a dyn TableReference>,
+}
+
+impl<'a> PostCommit<'a> {
+    /// Runs the post commit activities
+    async fn run_post_commit_hook(
+        &self,
+        version: i64,
+        commit_data: &CommitData,
+    ) -> DeltaResult<()> {
+        if self.create_checkpoint {
+            self.create_checkpoint(&self.table_data, &self.log_store, version, commit_data)
+                .await?
+        }
+        Ok(())
+    }
+    async fn create_checkpoint(
+        &self,
+        table: &Option<&'a dyn TableReference>,
+        log_store: &LogStoreRef,
+        version: i64,
+        commit_data: &CommitData,
+    ) -> DeltaResult<()> {
+        if let Some(table) = table {
+            let checkpoint_interval = table.config().checkpoint_interval() as i64;
+            if ((version + 1) % checkpoint_interval) == 0 {
+                // We have to advance the snapshot otherwise we can't create a checkpoint
+                let mut snapshot = table.eager_snapshot().unwrap().clone();
+                snapshot.advance(vec![commit_data])?;
+                let state = DeltaTableState {
+                    app_transaction_version: HashMap::new(),
+                    snapshot,
+                };
+                create_checkpoint_for(version, &state, log_store.as_ref()).await?
+            }
+        }
+        Ok(())
+    }
+}
+
 /// A commit that successfully completed
 pub struct FinalizedCommit {
     /// The winning version number of the commit
@@ -554,6 +650,27 @@ impl FinalizedCommit {
     }
 }
 
+impl<'a> std::future::IntoFuture for PostCommit<'a> {
+    type Output = DeltaResult<FinalizedCommit>;
+    type IntoFuture = BoxFuture<'a, Self::Output>;
+
+    fn into_future(self) -> Self::IntoFuture {
+        let this = self;
+
+        Box::pin(async move {
+            match this.run_post_commit_hook(this.version, &this.data).await {
+                Ok(_) => {
+                    return Ok(FinalizedCommit {
+                        version: this.version,
+                        data: this.data,
+                    })
+                }
+                Err(err) => return Err(err),
+            };
+        })
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::{collections::HashMap, sync::Arc};

crates/core/src/protocol/checkpoints.rs

@@ -27,7 +27,6 @@ use crate::logstore::LogStore;
 use crate::table::state::DeltaTableState;
 use crate::table::{get_partition_col_data_types, CheckPoint, CheckPointBuilder};
 use crate::{open_table_with_version, DeltaTable};
-
 type SchemaPath = Vec<String>;
 
 /// Error returned when there is an error during creating a checkpoint.

crates/core/src/table/config.rs

@@ -274,7 +274,7 @@ impl<'a> TableConfig<'a> {
             DeltaConfigKey::CheckpointInterval,
             checkpoint_interval,
             i32,
-            10
+            100
         ),
     );
 
@@ -591,7 +591,7 @@ mod tests {
     fn get_long_from_metadata_test() {
         let md = dummy_metadata();
         let config = TableConfig(&md.configuration);
-        assert_eq!(config.checkpoint_interval(), 10,)
+        assert_eq!(config.checkpoint_interval(), 100,)
     }
 
     #[test]

crates/core/src/table/state.rs

@@ -24,7 +24,7 @@ use crate::{DeltaResult, DeltaTableError};
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct DeltaTableState {
-    app_transaction_version: HashMap<String, i64>,
+    pub(crate) app_transaction_version: HashMap<String, i64>,
     pub(crate) snapshot: EagerSnapshot,
 }