feat: add post_commithook_properties to restore (delta-io#4336)

# Description
- replacing manual commit pipeline (into_prepared_commit_future + manual
write_commit_entry) in restore.rs into standard pipeline, all 3 stages
run automatically like other endpoints

- exposed `post_commithook_properties` in the Python binding, public
API, and type stub

- added tests verifying the parameter is accepted with post commimt hook

# Related Issue(s)
<!---
For example:

- closes #106
--->
closes delta-io#4251




<!---
Share links to useful documentation
--->

---------

Signed-off-by: Byeori Kim <bk.byeori.kim@gmail.com>
Co-authored-by: Ethan Urbanski <ethanurbanski@gmail.com>

Author: byuly

crates/core/src/operations/restore.rs

@@ -35,7 +35,7 @@ use serde::Serialize;
 use uuid::Uuid;
 
 use super::{CustomExecuteHandler, Operation};
-use crate::kernel::transaction::{CommitBuilder, CommitProperties, TransactionError};
+use crate::kernel::transaction::{CommitBuilder, CommitProperties};
 use crate::kernel::{
     Action, Add, EagerSnapshot, ProtocolExt as _, ProtocolInner, Remove, Version, resolve_snapshot,
 };
@@ -166,8 +166,9 @@ async fn execute(
     ignore_missing_files: bool,
     protocol_downgrade_allowed: bool,
     mut commit_properties: CommitProperties,
+    custom_execute_handler: Option<Arc<dyn CustomExecuteHandler>>,
     operation_id: Uuid,
-) -> DeltaResult<RestoreMetrics> {
+) -> DeltaResult<(RestoreMetrics, DeltaTableState)> {
     if !(version_to_restore
         .is_none()
         .bitxor(datetime_to_restore.is_none()))
@@ -294,30 +295,15 @@ async fn execute(
         datetime: datetime_to_restore.map(|time| -> i64 { time.timestamp_millis() }),
     };
 
-    let prepared_commit = CommitBuilder::from(commit_properties)
+    let commit = CommitBuilder::from(commit_properties)
         .with_actions(actions)
+        .with_max_retries(0)
+        .with_operation_id(operation_id)
+        .with_post_commit_hook_handler(custom_execute_handler)
         .build(Some(&snapshot), log_store.clone(), operation)
-        .into_prepared_commit_future()
         .await?;
 
-    let commit_version = snapshot.version() + 1;
-    let commit_bytes = prepared_commit.commit_or_bytes();
-    match log_store
-        .write_commit_entry(commit_version, commit_bytes.clone(), operation_id)
-        .await
-    {
-        Ok(_) => {}
-        Err(err @ TransactionError::VersionAlreadyExists(_)) => {
-            return Err(err.into());
-        }
-        Err(err) => {
-            log_store
-                .abort_commit_entry(commit_version, commit_bytes.clone(), operation_id)
-                .await?;
-            return Err(err.into());
-        }
-    }
-    Ok(metrics)
+    Ok((metrics, commit.snapshot()))
 }
 
 async fn check_files_available(
@@ -344,7 +330,7 @@ impl std::future::IntoFuture for RestoreBuilder {
     type IntoFuture = BoxFuture<'static, Self::Output>;
 
     fn into_future(self) -> Self::IntoFuture {
-        let this = self;
+        let mut this = self;
 
         Box::pin(async move {
             let snapshot =
@@ -353,24 +339,28 @@ impl std::future::IntoFuture for RestoreBuilder {
             let operation_id = this.get_operation_id();
             this.pre_execute(operation_id).await?;
 
-            let metrics = execute(
+            let handle = this.custom_execute_handler.take();
+            let (metrics, new_state) = execute(
                 this.log_store.clone(),
-                snapshot.clone(),
+                snapshot,
                 this.version_to_restore,
                 this.datetime_to_restore,
                 this.ignore_missing_files,
                 this.protocol_downgrade_allowed,
                 this.commit_properties.clone(),
+                handle.clone(),
                 operation_id,
             )
             .await?;
 
-            this.post_execute(operation_id).await?;
+            if let Some(handler) = handle {
+                handler.post_execute(&this.log_store, operation_id).await?;
+            }
 
-            let mut table =
-                DeltaTable::new_with_state(this.log_store, DeltaTableState::new(snapshot));
-            table.update_state().await?;
-            Ok((table, metrics))
+            Ok((
+                DeltaTable::new_with_state(this.log_store, new_state),
+                metrics,
+            ))
         })
     }
 }

python/deltalake/_internal.pyi

@@ -189,6 +189,7 @@ class RawDeltaTable:
         ignore_missing_files: bool,
         protocol_downgrade_allowed: bool,
         commit_properties: CommitProperties | None,
+        post_commithook_properties: PostCommitHookProperties | None,
     ) -> str: ...
     def history(self, limit: int | None) -> list[str]: ...
     def update_incremental(self) -> None: ...

python/deltalake/table.py

@@ -892,6 +892,7 @@ def restore(
         ignore_missing_files: bool = False,
         protocol_downgrade_allowed: bool = False,
         commit_properties: CommitProperties | None = None,
+        post_commithook_properties: PostCommitHookProperties | None = None,
     ) -> dict[str, Any]:
         """
         Restores table to a given version or datetime.
@@ -901,6 +902,7 @@ def restore(
             ignore_missing_files: whether the operation carry on when some data files missing.
             protocol_downgrade_allowed: whether the operation when protocol version upgraded.
             commit_properties: properties of the transaction commit. If None, default values are used.
+            post_commithook_properties: properties for the post commit hook. If None, default values are used.
 
         Returns:
             the metrics from restore.
@@ -911,13 +913,15 @@ def restore(
                 ignore_missing_files=ignore_missing_files,
                 protocol_downgrade_allowed=protocol_downgrade_allowed,
                 commit_properties=commit_properties,
+                post_commithook_properties=post_commithook_properties,
             )
         else:
             metrics = self._table.restore(
                 target,
                 ignore_missing_files=ignore_missing_files,
                 protocol_downgrade_allowed=protocol_downgrade_allowed,
                 commit_properties=commit_properties,
+                post_commithook_properties=post_commithook_properties,
             )
         return json.loads(metrics)
 

python/src/lib.rs

@@ -1241,14 +1241,15 @@ impl RawDeltaTable {
 
     // Run the restore command on the Delta Table: restore table to a given version or datetime
     #[pyo3(signature = (
-        target, *, ignore_missing_files = false, protocol_downgrade_allowed = false, commit_properties=None
+        target, *, ignore_missing_files = false, protocol_downgrade_allowed = false, commit_properties=None, post_commithook_properties=None
     ))]
     pub fn restore(
         &self,
         target: Option<&Bound<'_, PyAny>>,
         ignore_missing_files: bool,
         protocol_downgrade_allowed: bool,
         commit_properties: Option<PyCommitProperties>,
+        post_commithook_properties: Option<PyPostCommitHookProperties>,
     ) -> PyResult<String> {
         let table = self._table.lock().map_err(to_rt_err)?.clone();
         let mut cmd = table.restore();
@@ -1268,7 +1269,9 @@ impl RawDeltaTable {
         cmd = cmd.with_ignore_missing_files(ignore_missing_files);
         cmd = cmd.with_protocol_downgrade_allowed(protocol_downgrade_allowed);
 
-        if let Some(commit_properties) = maybe_create_commit_properties(commit_properties, None) {
+        if let Some(commit_properties) =
+            maybe_create_commit_properties(commit_properties, post_commithook_properties)
+        {
             cmd = cmd.with_commit_properties(commit_properties);
         }
 

python/tests/test_restore.py

@@ -4,7 +4,12 @@
 import pytest
 from arro3.core import Table
 
-from deltalake import CommitProperties, DeltaTable, write_deltalake
+from deltalake import (
+    CommitProperties,
+    DeltaTable,
+    PostCommitHookProperties,
+    write_deltalake,
+)
 
 
 @pytest.mark.parametrize("use_relative", [True, False])
@@ -88,3 +93,29 @@ def test_restore_with_datetime(
     last_action = dt.history(1)[0]
     assert last_action["operation"] == "RESTORE"
     assert dt.version() == old_version + 1
+
+
+def test_restore_with_post_commithook_properties(
+    tmp_path: pathlib.Path,
+    sample_table: Table,
+):
+    write_deltalake(str(tmp_path), sample_table, mode="append")
+    write_deltalake(str(tmp_path), sample_table, mode="append")
+    write_deltalake(str(tmp_path), sample_table, mode="append")
+
+    dt = DeltaTable(str(tmp_path))
+    old_version = dt.version()
+    dt.restore(
+        1,
+        post_commithook_properties=PostCommitHookProperties(
+            create_checkpoint=False,
+            cleanup_expired_logs=False,
+        ),
+    )
+    last_action = dt.history(1)[0]
+    assert last_action["operation"] == "RESTORE"
+    assert dt.version() == old_version + 1
+
+    log_dir = tmp_path / "_delta_log"
+    checkpoint_files = list(log_dir.glob("*.checkpoint.parquet"))
+    assert len(checkpoint_files) == 0