You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/10/12 08:55:48 UTC
(paimon-rust) branch main updated: [core] Add confirmed snapshot commit callbacks (#1108)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 9745230b [core] Add confirmed snapshot commit callbacks (#1108)
9745230b is described below
commit 9745230b1525c6e0ed4416dbb53617973634622a
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Oct 12 16:53:46 2026 +0800
[core] Add confirmed snapshot commit callbacks (#1108)
---
.../python/python/pypaimon_rust/datafusion.pyi | 105 ++++++++
bindings/python/src/blob_uri_reader.rs | 4 +-
bindings/python/src/commit_callback.rs | 267 ++++++++++++++++++++
bindings/python/src/context.rs | 4 +
bindings/python/src/lib.rs | 1 +
bindings/python/src/snapshot.rs | 26 ++
bindings/python/src/write.rs | 121 ++++++++-
bindings/python/tests/test_table_commit.py | 140 +++++++++++
crates/paimon/src/spec/manifest_entry.rs | 4 +-
crates/paimon/src/spec/mod.rs | 1 +
crates/paimon/src/table/commit_callback.rs | 45 ++++
crates/paimon/src/table/mod.rs | 2 +
crates/paimon/src/table/table_commit.rs | 67 ++++-
.../src/table/table_commit/callback_tests.rs | 275 +++++++++++++++++++++
14 files changed, 1035 insertions(+), 27 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi b/bindings/python/python/pypaimon_rust/datafusion.pyi
index abc262e2..3b80fb45 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+from datetime import datetime
from os import PathLike
from typing import (
Any, BinaryIO, Callable, Dict, Iterable, List, Literal, Optional, Protocol, Sequence, Tuple,
@@ -175,6 +176,12 @@ class Snapshot:
def total_record_count(self) -> Optional[int]: ...
def delta_record_count(self) -> Optional[int]: ...
def commit_kind(self) -> str: ...
+ def schema_id(self) -> int: ...
+ def commit_user(self) -> str: ...
+ def commit_identifier(self) -> int: ...
+ def next_row_id(self) -> Optional[int]: ...
+ def properties(self) -> Optional[Dict[str, str]]: ...
+ def uuid(self) -> Optional[str]: ...
class Tag:
def name(self) -> str: ...
@@ -435,7 +442,98 @@ class StreamTableWrite:
"""
...
+class BinaryTableStats:
+ """Read-only core statistics. Min/max values are serialized core BinaryRows."""
+ @property
+ def min_values(self) -> bytes: ...
+ @property
+ def max_values(self) -> bytes: ...
+ @property
+ def null_counts(self) -> List[Optional[int]]: ...
+
+class DataFileMeta:
+ """Read-only view of the core data-file metadata."""
+ @property
+ def file_name(self) -> str: ...
+ @property
+ def file_size(self) -> int: ...
+ @property
+ def row_count(self) -> int: ...
+ @property
+ def min_key(self) -> bytes: ...
+ @property
+ def max_key(self) -> bytes: ...
+ @property
+ def key_stats(self) -> BinaryTableStats: ...
+ @property
+ def value_stats(self) -> BinaryTableStats: ...
+ @property
+ def min_sequence_number(self) -> int: ...
+ @property
+ def max_sequence_number(self) -> int: ...
+ @property
+ def schema_id(self) -> int: ...
+ @property
+ def level(self) -> int: ...
+ @property
+ def extra_files(self) -> List[str]: ...
+ @property
+ def creation_time(self) -> Optional[datetime]: ...
+ @property
+ def delete_row_count(self) -> Optional[int]: ...
+ @property
+ def embedded_index(self) -> Optional[bytes]: ...
+ @property
+ def file_source(self) -> Optional[int]: ...
+ @property
+ def value_stats_cols(self) -> Optional[List[str]]: ...
+ @property
+ def external_path(self) -> Optional[str]: ...
+ @property
+ def first_row_id(self) -> Optional[int]: ...
+ @property
+ def write_cols(self) -> Optional[List[str]]: ...
+ @property
+ def column_max_sequence_numbers(self) -> Optional[List[int]]: ...
+
+class ManifestEntry:
+ """Read-only core file delta. kind is 0 (ADD) or 1 (DELETE)."""
+ @property
+ def kind(self) -> int: ...
+ @property
+ def partition(self) -> bytes:
+ """Core serialized BinaryRow bytes for the partition."""
+ ...
+ @property
+ def bucket(self) -> int: ...
+ @property
+ def total_buckets(self) -> int: ...
+ @property
+ def file(self) -> DataFileMeta: ...
+ def _to_commit_message(self) -> CommitMessage:
+ """Internal PyPaimon file metadata bridge; excludes the entry kind."""
+ ...
+
+class CommitCallbackContext:
+ """Read-only core notification, retainable after the callback returns."""
+ @property
+ def snapshot(self) -> Snapshot: ...
+ @property
+ def commit_entries(self) -> List[ManifestEntry]: ...
+ @property
+ def identifier(self) -> int: ...
+ def _snapshot_json(self) -> str:
+ """Internal lossless PyPaimon Snapshot conversion."""
+ ...
+
class BatchTableCommit:
+ def add_commit_callback(self, callback: Callable[[CommitCallbackContext], None]) -> None:
+ """Register an idempotent callback for confirmed snapshots, in registration order.
+
+ Errors propagate after publication; committed files are retained. The
+ caller owns callback resources. close() releases the callback reference.
+ """
+ ...
def close(self) -> None: ...
def commit(
self, messages: Sequence[CommitMessage],
@@ -454,6 +552,13 @@ class BatchTableCommit:
...
class StreamTableCommit:
+ def add_commit_callback(self, callback: Callable[[CommitCallbackContext], None]) -> None:
+ """Register an idempotent callback for confirmed snapshots, in registration order.
+
+ Errors propagate after publication; committed files are retained. The
+ caller owns callback resources. close() releases the callback reference.
+ """
+ ...
def close(self) -> None: ...
def commit(
self, commit_identifier: int, messages: Sequence[CommitMessage],
diff --git a/bindings/python/src/blob_uri_reader.rs b/bindings/python/src/blob_uri_reader.rs
index 39c21c16..95b9b38d 100644
--- a/bindings/python/src/blob_uri_reader.rs
+++ b/bindings/python/src/blob_uri_reader.rs
@@ -35,14 +35,14 @@ impl std::fmt::Debug for PythonCallbackError {
impl std::fmt::Display for PythonCallbackError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- f.write_str("Python Blob callback failed")
+ f.write_str("Python callback failed")
}
}
impl std::error::Error for PythonCallbackError {}
pub(crate) fn callback_error(error: PyErr) -> paimon::Error {
paimon::Error::UnexpectedError {
- message: "Python Blob callback failed".into(),
+ message: "Python callback failed".into(),
source: Some(Box::new(PythonCallbackError(error))),
}
}
diff --git a/bindings/python/src/commit_callback.rs b/bindings/python/src/commit_callback.rs
new file mode 100644
index 00000000..7bb744c3
--- /dev/null
+++ b/bindings/python/src/commit_callback.rs
@@ -0,0 +1,267 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Typed views of core commit notifications. Wire conversion is private to the
+//! PyPaimon adapter; callback users receive domain objects.
+
+use chrono::{DateTime, Utc};
+use paimon::spec::{BinaryTableStats, DataFileMeta, FileKind, ManifestEntry};
+use paimon::table::{CommitCallbackContext, CommitMessage};
+use pyo3::prelude::*;
+use pyo3::types::PyBytes;
+
+use crate::snapshot::PySnapshot;
+use crate::write::PyCommitMessage;
+
+/// Own a copy of the core notification so Python can retain it after call().
+#[pyclass(
+ name = "CommitCallbackContext",
+ module = "pypaimon_rust.datafusion",
+ frozen
+)]
+pub struct PyCommitCallbackContext {
+ inner: CommitCallbackContext,
+}
+
+impl PyCommitCallbackContext {
+ pub(crate) fn new(inner: CommitCallbackContext) -> Self {
+ Self { inner }
+ }
+}
+
+#[pymethods]
+impl PyCommitCallbackContext {
+ #[getter]
+ fn snapshot(&self) -> PySnapshot {
+ PySnapshot::new(self.inner.snapshot.clone())
+ }
+
+ #[getter]
+ fn commit_entries(&self) -> Vec<PyManifestEntry> {
+ self.inner
+ .commit_entries
+ .iter()
+ .cloned()
+ .map(|inner| PyManifestEntry { inner })
+ .collect()
+ }
+
+ #[getter]
+ fn identifier(&self) -> i64 {
+ self.inner.identifier
+ }
+
+ /// Internal lossless Snapshot conversion for the PyPaimon adapter.
+ fn _snapshot_json(&self) -> PyResult<String> {
+ serde_json::to_string(&self.inner.snapshot).map_err(|error| {
+ pyo3::exceptions::PyValueError::new_err(format!(
+ "Cannot serialize callback snapshot: {error}"
+ ))
+ })
+ }
+}
+
+#[pyclass(name = "ManifestEntry", module = "pypaimon_rust.datafusion", frozen)]
+pub struct PyManifestEntry {
+ inner: ManifestEntry,
+}
+
+#[pymethods]
+impl PyManifestEntry {
+ #[getter]
+ fn kind(&self) -> i32 {
+ match self.inner.kind() {
+ FileKind::Add => 0,
+ FileKind::Delete => 1,
+ }
+ }
+
+ /// Core BinaryRow bytes; callers can inspect metadata without decoding them.
+ #[getter]
+ fn partition<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
+ PyBytes::new(py, self.inner.partition())
+ }
+
+ #[getter]
+ fn bucket(&self) -> i32 {
+ self.inner.bucket()
+ }
+
+ #[getter]
+ fn total_buckets(&self) -> i32 {
+ self.inner.total_buckets()
+ }
+
+ #[getter]
+ fn file(&self) -> PyDataFileMeta {
+ PyDataFileMeta {
+ inner: self.inner.file().clone(),
+ }
+ }
+
+ /// Internal v14 bridge preserves the complete file metadata for PyPaimon.
+ fn _to_commit_message(&self) -> PyCommitMessage {
+ PyCommitMessage::new(CommitMessage::new(
+ self.inner.partition().to_vec(),
+ self.inner.bucket(),
+ vec![self.inner.file().clone()],
+ ))
+ }
+}
+
+#[pyclass(name = "DataFileMeta", module = "pypaimon_rust.datafusion", frozen)]
+pub struct PyDataFileMeta {
+ inner: DataFileMeta,
+}
+
+#[pymethods]
+impl PyDataFileMeta {
+ #[getter]
+ fn file_name(&self) -> String {
+ self.inner.file_name.clone()
+ }
+
+ #[getter]
+ fn file_size(&self) -> i64 {
+ self.inner.file_size
+ }
+
+ #[getter]
+ fn row_count(&self) -> i64 {
+ self.inner.row_count
+ }
+
+ #[getter]
+ fn min_sequence_number(&self) -> i64 {
+ self.inner.min_sequence_number
+ }
+
+ #[getter]
+ fn max_sequence_number(&self) -> i64 {
+ self.inner.max_sequence_number
+ }
+
+ #[getter]
+ fn schema_id(&self) -> i64 {
+ self.inner.schema_id
+ }
+
+ #[getter]
+ fn level(&self) -> i32 {
+ self.inner.level
+ }
+
+ #[getter]
+ fn extra_files(&self) -> Vec<String> {
+ self.inner.extra_files.clone()
+ }
+
+ #[getter]
+ fn creation_time(&self) -> Option<DateTime<Utc>> {
+ self.inner.creation_time
+ }
+
+ #[getter]
+ fn delete_row_count(&self) -> Option<i64> {
+ self.inner.delete_row_count
+ }
+
+ #[getter]
+ fn file_source(&self) -> Option<i32> {
+ self.inner.file_source
+ }
+
+ #[getter]
+ fn value_stats_cols(&self) -> Option<Vec<String>> {
+ self.inner.value_stats_cols.clone()
+ }
+
+ #[getter]
+ fn external_path(&self) -> Option<String> {
+ self.inner.external_path.clone()
+ }
+
+ #[getter]
+ fn first_row_id(&self) -> Option<i64> {
+ self.inner.first_row_id
+ }
+
+ #[getter]
+ fn write_cols(&self) -> Option<Vec<String>> {
+ self.inner.write_cols.clone()
+ }
+
+ #[getter]
+ fn column_max_sequence_numbers(&self) -> Option<Vec<i64>> {
+ self.inner.column_max_sequence_numbers.clone()
+ }
+
+ #[getter]
+ fn min_key<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
+ PyBytes::new(py, &self.inner.min_key)
+ }
+
+ #[getter]
+ fn max_key<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
+ PyBytes::new(py, &self.inner.max_key)
+ }
+
+ #[getter]
+ fn embedded_index<'py>(&self, py: Python<'py>) -> Option<Bound<'py, PyBytes>> {
+ self.inner
+ .embedded_index
+ .as_ref()
+ .map(|bytes| PyBytes::new(py, bytes))
+ }
+
+ #[getter]
+ fn key_stats(&self) -> PyBinaryTableStats {
+ PyBinaryTableStats {
+ inner: self.inner.key_stats.clone(),
+ }
+ }
+
+ #[getter]
+ fn value_stats(&self) -> PyBinaryTableStats {
+ PyBinaryTableStats {
+ inner: self.inner.value_stats.clone(),
+ }
+ }
+}
+
+#[pyclass(name = "BinaryTableStats", module = "pypaimon_rust.datafusion", frozen)]
+pub struct PyBinaryTableStats {
+ inner: BinaryTableStats,
+}
+
+#[pymethods]
+impl PyBinaryTableStats {
+ #[getter]
+ fn min_values<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
+ PyBytes::new(py, self.inner.min_values())
+ }
+
+ #[getter]
+ fn max_values<'py>(&self, py: Python<'py>) -> Bound<'py, PyBytes> {
+ PyBytes::new(py, self.inner.max_values())
+ }
+
+ #[getter]
+ fn null_counts(&self) -> Vec<Option<i64>> {
+ self.inner.null_counts().clone()
+ }
+}
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 1b9c5397..37277513 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -836,6 +836,10 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_, PyModule>) -> PyResult<()>
this.add_class::<crate::write::PyBatchTableCommit>()?;
this.add_class::<crate::write::PyStreamTableCommit>()?;
this.add_class::<crate::write::PyCommitMessage>()?;
+ this.add_class::<crate::commit_callback::PyCommitCallbackContext>()?;
+ this.add_class::<crate::commit_callback::PyManifestEntry>()?;
+ this.add_class::<crate::commit_callback::PyDataFileMeta>()?;
+ this.add_class::<crate::commit_callback::PyBinaryTableStats>()?;
this.add_function(wrap_pyfunction!(udf, &this)?)?;
this.add_class::<crate::snapshot::PySnapshot>()?;
this.add_class::<crate::tag::PyTag>()?;
diff --git a/bindings/python/src/lib.rs b/bindings/python/src/lib.rs
index 86e7341b..6b2fcc54 100644
--- a/bindings/python/src/lib.rs
+++ b/bindings/python/src/lib.rs
@@ -19,6 +19,7 @@ use pyo3::prelude::*;
mod blob;
mod blob_uri_reader;
+mod commit_callback;
mod context;
mod error;
mod full_text_search;
diff --git a/bindings/python/src/snapshot.rs b/bindings/python/src/snapshot.rs
index 58505188..11c40c98 100644
--- a/bindings/python/src/snapshot.rs
+++ b/bindings/python/src/snapshot.rs
@@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
+use std::collections::HashMap;
+
use paimon::spec::Snapshot;
use pyo3::prelude::*;
@@ -47,6 +49,30 @@ impl PySnapshot {
self.inner.delta_record_count()
}
+ fn schema_id(&self) -> i64 {
+ self.inner.schema_id()
+ }
+
+ fn commit_user(&self) -> String {
+ self.inner.commit_user().to_string()
+ }
+
+ fn commit_identifier(&self) -> i64 {
+ self.inner.commit_identifier()
+ }
+
+ fn next_row_id(&self) -> Option<i64> {
+ self.inner.next_row_id()
+ }
+
+ fn properties(&self) -> Option<HashMap<String, String>> {
+ self.inner.properties().cloned()
+ }
+
+ fn uuid(&self) -> Option<String> {
+ self.inner.uuid().map(str::to_string)
+ }
+
fn commit_kind(&self) -> String {
self.inner.commit_kind().to_string()
}
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index be06b18f..312990d0 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -23,15 +23,18 @@ use arrow::pyarrow::FromPyArrow;
use arrow::record_batch::RecordBatch;
use paimon::spec::{BlobConsumer, BlobDescriptor, CoreOptions, DataType, Datum, RowType};
use paimon::table::{
- CommitMessage, PostponeBucketPlan, PostponeFixedBucketTableWrite, Table, TableCommit,
- TableUpdate, TableUpdateByRowId, TableWrite, COMMIT_MESSAGE_SERIALIZER_VERSION,
+ CommitCallback, CommitCallbackContext, CommitMessage, PostponeBucketPlan,
+ PostponeFixedBucketTableWrite, Table, TableCommit, TableUpdate, TableUpdateByRowId, TableWrite,
+ COMMIT_MESSAGE_SERIALIZER_VERSION,
};
use paimon_datafusion::runtime::runtime;
use pyo3::exceptions::{PyRuntimeError, PyTypeError, PyValueError};
use pyo3::prelude::*;
use pyo3::types::{PyBytes, PyDict, PyString};
+use pyo3::{PyTraverseError, PyVisit};
use crate::blob_uri_reader::{callback_error, optional_uri_reader_factory, to_callback_error};
+use crate::commit_callback::PyCommitCallbackContext;
use crate::error::to_py_err;
use crate::predicate::{dict_to_table_predicate, py_to_datum};
@@ -1320,8 +1323,41 @@ impl PyStreamTableWrite {
}
}
+/// Transport only: publication and retry notification belong to Rust core.
+struct PythonCommitCallback {
+ // One Python reference shared by the core hook and the GC-visible wrapper.
+ callback: Mutex<Option<Py<PyAny>>>,
+}
+
+#[async_trait::async_trait]
+impl CommitCallback for PythonCommitCallback {
+ async fn call(&self, context: &CommitCallbackContext) -> paimon::Result<()> {
+ // Suspend the runtime context so Python can call Native again, for
+ // example to read the snapshot which triggered this callback.
+ tokio::task::block_in_place(|| {
+ Python::attach(|py| {
+ let callback = self
+ .callback
+ .lock()
+ .unwrap()
+ .as_ref()
+ .map(|callback| callback.clone_ref(py));
+ let Some(callback) = callback else {
+ return Ok(());
+ };
+ callback
+ .bind(py)
+ .call1((PyCommitCallbackContext::new(context.clone()),))?;
+ Ok(())
+ })
+ })
+ .map_err(callback_error)
+ }
+}
+
struct CommitContext {
inner: TableCommit,
+ callbacks: Vec<Arc<PythonCommitCallback>>,
table: Arc<Table>,
commit_user: String,
}
@@ -1337,6 +1373,7 @@ impl CommitContext {
.with_ignore_empty_commit(ignore_empty);
Ok(Self {
inner,
+ callbacks: Vec::new(),
table: Arc::clone(table),
commit_user: commit_user.to_string(),
})
@@ -1363,6 +1400,38 @@ impl CommitContext {
.map_err(to_py_err)
}
+ fn add_commit_callback(&mut self, py: Python<'_>, callback: Py<PyAny>) -> PyResult<()> {
+ if !callback.bind(py).is_callable() {
+ return Err(PyTypeError::new_err("Commit callback must be callable"));
+ }
+ let callback = Arc::new(PythonCommitCallback {
+ callback: Mutex::new(Some(callback)),
+ });
+ self.inner
+ .add_commit_callback(callback.clone())
+ .map_err(to_py_err)?;
+ self.callbacks.push(callback);
+ Ok(())
+ }
+
+ fn traverse_callbacks(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> {
+ for callback in &self.callbacks {
+ visit.call(&*callback.callback.lock().unwrap())?;
+ }
+ Ok(())
+ }
+
+ fn clear_callbacks(&self) {
+ let references = self
+ .callbacks
+ .iter()
+ .map(|callback| callback.callback.lock().unwrap().take())
+ .collect::<Vec<_>>();
+ // Destructors can re-enter Native. Release every lock before dropping
+ // references. PyPaimon owns close() on the original callback objects.
+ drop(references);
+ }
+
fn committer(&self, properties: Option<HashMap<String, String>>) -> TableCommit {
// Properties belong to this call, including retries. The reusable
// stream context must not carry them into a later checkpoint.
@@ -1394,8 +1463,23 @@ impl PyBatchTableCommit {
#[pymethods]
impl PyBatchTableCommit {
- /// Rust committers have no background resources to shut down.
- fn close(&self) {}
+ fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> {
+ self.context.traverse_callbacks(visit)
+ }
+
+ fn __clear__(&mut self) {
+ self.context.clear_callbacks();
+ }
+
+ /// Register an idempotent callback receiving the committed core context.
+ fn add_commit_callback(&mut self, py: Python<'_>, callback: Py<PyAny>) -> PyResult<()> {
+ self.context.add_commit_callback(py, callback)
+ }
+
+ /// Release callback references; callback resources belong to the caller.
+ fn close(&self) {
+ self.context.clear_callbacks();
+ }
#[pyo3(signature = (messages, snapshot_properties=None))]
fn commit(
@@ -1418,13 +1502,13 @@ impl PyBatchTableCommit {
}
})
})
- .map_err(to_py_err)
+ .map_err(to_callback_error)
}
fn truncate_table(&mut self, py: Python<'_>) -> PyResult<()> {
self.check_committed()?;
py.detach(|| runtime().block_on(self.context.inner.truncate_table()))
- .map_err(to_py_err)
+ .map_err(to_callback_error)
}
fn truncate_partitions(
@@ -1437,7 +1521,7 @@ impl PyBatchTableCommit {
.map(|spec| partition_spec(py, &self.context.table, spec))
.collect::<PyResult<Vec<_>>>()?;
py.detach(|| runtime().block_on(self.context.inner.drop_partitions(partitions)))
- .map_err(to_py_err)
+ .map_err(to_callback_error)
}
fn abort(&self, py: Python<'_>, messages: &Bound<'_, PyAny>) -> PyResult<()> {
@@ -1452,8 +1536,23 @@ pub struct PyStreamTableCommit {
#[pymethods]
impl PyStreamTableCommit {
- /// Rust committers have no background resources to shut down.
- fn close(&self) {}
+ fn __traverse__(&self, visit: PyVisit<'_>) -> Result<(), PyTraverseError> {
+ self.context.traverse_callbacks(visit)
+ }
+
+ fn __clear__(&mut self) {
+ self.context.clear_callbacks();
+ }
+
+ /// Register an idempotent callback receiving the committed core context.
+ fn add_commit_callback(&mut self, py: Python<'_>, callback: Py<PyAny>) -> PyResult<()> {
+ self.context.add_commit_callback(py, callback)
+ }
+
+ /// Release callback references; callback resources belong to the caller.
+ fn close(&self) {
+ self.context.clear_callbacks();
+ }
#[pyo3(signature = (commit_identifier, messages, snapshot_properties=None))]
fn commit(
@@ -1468,7 +1567,7 @@ impl PyStreamTableCommit {
py.detach(|| {
runtime().block_on(committer.commit_with_identifier(messages, commit_identifier))
})
- .map_err(to_py_err)
+ .map_err(to_callback_error)
}
/// Sort identifiers, filter already committed groups, and return the number committed.
@@ -1488,7 +1587,7 @@ impl PyStreamTableCommit {
})
.collect::<PyResult<Vec<_>>>()?;
py.detach(|| runtime().block_on(self.context.inner.filter_and_commit(commits)))
- .map_err(to_py_err)
+ .map_err(to_callback_error)
}
fn abort(&self, py: Python<'_>, messages: &Bound<'_, PyAny>) -> PyResult<()> {
diff --git a/bindings/python/tests/test_table_commit.py b/bindings/python/tests/test_table_commit.py
index ffb69285..2de95d8b 100644
--- a/bindings/python/tests/test_table_commit.py
+++ b/bindings/python/tests/test_table_commit.py
@@ -506,3 +506,143 @@ def test_close_cleans_unprepared_output_but_preserves_prepared_files(tmp_path, b
for row in pq.ParquetFile(path).read(columns=["id"]).to_pylist()] == [1]
if bucket != "-2":
assert _rows(table) == [1]
+
+
+@pytest.mark.parametrize('stream', [False, True])
+@pytest.mark.parametrize('close', [False, True])
+@pytest.mark.parametrize('bound_method', [False, True])
+def test_commit_callback_owner_cycle_is_collected(tmp_path, stream, close, bound_method):
+ import gc
+ import weakref
+
+ table = _table(tmp_path)
+ builder = table.new_stream_write_builder() if stream else table.new_batch_write_builder()
+ commit = builder.new_commit()
+
+ class Owner:
+ def __init__(self, committer):
+ self.committer = committer
+
+ def __call__(self, context):
+ pass
+
+ owner = Owner(commit)
+ reference = weakref.ref(owner)
+ callback = owner.__call__ if bound_method else owner
+ commit.add_commit_callback(callback)
+ # Register more than one reference so GC traversal must count each edge.
+ commit.add_commit_callback(callback)
+ if close:
+ commit.close()
+ del callback, owner, commit
+ gc.collect()
+ assert reference() is None
+
+
+@pytest.mark.parametrize('stream', [False, True])
+def test_commit_callback_bridge_preserves_error_and_publication(tmp_path, stream):
+ table = _table(tmp_path)
+ builder = table.new_stream_write_builder() if stream else table.new_batch_write_builder()
+ commit = builder.new_commit()
+ seen = []
+ error = LookupError('notification failed')
+
+ def callback(context):
+ assert isinstance(context, datafusion.CommitCallbackContext)
+ assert context.snapshot.id() == 1
+ assert context.identifier == (7 if stream else (2 ** 63 - 1))
+ assert len(context.commit_entries) == 1
+ entry = context.commit_entries[0]
+ assert isinstance(entry, datafusion.ManifestEntry)
+ assert entry.kind == 0
+ assert entry.total_buckets == -1
+ assert entry.bucket == 0
+ assert isinstance(entry.file, datafusion.DataFileMeta)
+ assert entry.file.row_count == 1
+ assert entry.file.file_name.endswith('.parquet')
+ assert _rows(table) == [1] # Re-enter the Native runtime during notification.
+ seen.append(context)
+ raise error
+
+ commit.add_commit_callback(callback)
+ messages = _prepare(builder, [1], [10], 7 if stream else None)
+ with pytest.raises(LookupError) as raised:
+ commit.commit(7, messages) if stream else commit.commit(messages)
+ assert raised.value is error
+ assert len(seen) == 1
+ assert _rows(table) == [1]
+ assert table.latest_snapshot().id() == 1
+ commit.close()
+
+
+def test_commit_callback_rejects_non_callable_before_publication(tmp_path):
+ table = _table(tmp_path)
+ commit = table.new_batch_write_builder().new_commit()
+ with pytest.raises(TypeError, match='must be callable'):
+ commit.add_commit_callback(42)
+ assert table.latest_snapshot() is None
+
+
+@pytest.mark.parametrize('stream', [False, True])
+def test_public_commit_callback_retains_typed_core_context(tmp_path, stream):
+ import gc
+
+ table = _table(tmp_path, options={
+ 'row-tracking.enabled': 'true', 'data-evolution.enabled': 'true'})
+ builder = table.new_stream_write_builder() if stream else table.new_batch_write_builder()
+ if stream:
+ builder.with_commit_user('callback-user')
+ commit = builder.new_commit()
+ contexts = []
+ commit.add_commit_callback(contexts.append)
+ assert not hasattr(commit, '_add_commit_callback')
+ messages = _prepare(builder, [1, 2], [10, 20], 7 if stream else None)
+ if stream:
+ commit.commit(7, messages, snapshot_properties={'source': 'typed-callback'})
+ else:
+ commit.commit(messages, snapshot_properties={'source': 'typed-callback'})
+ commit.close()
+ del messages, builder, commit
+ gc.collect()
+
+ assert len(contexts) == 1
+ context = contexts[0]
+ assert isinstance(context, datafusion.CommitCallbackContext)
+ assert isinstance(context.snapshot, datafusion.Snapshot)
+ assert context.snapshot.id() == 1
+ assert context.snapshot.schema_id() == 0
+ assert context.snapshot.commit_identifier() == context.identifier
+ assert context.snapshot.next_row_id() == 2
+ assert context.snapshot.properties() == {'source': 'typed-callback'}
+ if stream:
+ assert context.snapshot.commit_user() == 'callback-user'
+ entry = context.commit_entries[0]
+ file = entry.file
+ assert isinstance(entry.partition, bytes)
+ assert file.row_count == 2
+ assert file.file_size > 0
+ assert file.first_row_id == 0
+ assert file.min_sequence_number == file.max_sequence_number == 1
+ assert file.schema_id == 0
+ assert file.file_source == 0
+ assert file.external_path is None
+ assert file.write_cols is None
+ assert isinstance(file.value_stats, datafusion.BinaryTableStats)
+ assert file.value_stats.null_counts == [0, 0]
+ assert file.value_stats.min_values != file.value_stats.max_values
+ assert file.creation_time is not None
+ assert file.delete_row_count == 0
+ # Views are read-only; returned Python lists/maps cannot mutate core metadata.
+ with pytest.raises(AttributeError):
+ context.identifier = 99
+ with pytest.raises(AttributeError):
+ entry.kind = 1
+ with pytest.raises(AttributeError):
+ file.row_count = 99
+ context.commit_entries.clear()
+ file.extra_files.append('not-a-real-file')
+ context.snapshot.properties()['source'] = 'changed'
+ assert len(context.commit_entries) == 1
+ assert file.extra_files == []
+ assert context.snapshot.properties() == {'source': 'typed-callback'}
+ assert _rows(table) == [1, 2]
diff --git a/crates/paimon/src/spec/manifest_entry.rs b/crates/paimon/src/spec/manifest_entry.rs
index 484f2bd9..d978ae4a 100644
--- a/crates/paimon/src/spec/manifest_entry.rs
+++ b/crates/paimon/src/spec/manifest_entry.rs
@@ -77,7 +77,7 @@ pub struct ManifestEntry {
#[allow(dead_code)]
impl ManifestEntry {
- pub(crate) fn kind(&self) -> &FileKind {
+ pub fn kind(&self) -> &FileKind {
&self.kind
}
@@ -86,7 +86,7 @@ impl ManifestEntry {
&self.partition
}
- pub(crate) fn bucket(&self) -> i32 {
+ pub fn bucket(&self) -> i32 {
self.bucket
}
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index 8ef2af95..7969f9b3 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -113,6 +113,7 @@ pub(crate) use objects_file::{
};
pub(crate) mod avro;
pub(crate) mod stats;
+pub use stats::BinaryTableStats;
mod types;
pub use types::*;
mod partition;
diff --git a/crates/paimon/src/table/commit_callback.rs b/crates/paimon/src/table/commit_callback.rs
new file mode 100644
index 00000000..8a9ee0ed
--- /dev/null
+++ b/crates/paimon/src/table/commit_callback.rs
@@ -0,0 +1,45 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Post-publication notifications for Paimon table commits.
+
+use crate::spec::{ManifestEntry, Snapshot};
+use crate::Result;
+use async_trait::async_trait;
+
+/// The committed snapshot and its data-file delta, after row tracking assignment.
+///
+/// Entries include both ADD and DELETE records, but not changelog files. This is
+/// the snapshot / deltaFiles portion of Java CommitCallback.Context.
+#[derive(Debug, Clone)]
+pub struct CommitCallbackContext {
+ pub snapshot: Snapshot,
+ pub commit_entries: Vec<ManifestEntry>,
+ pub identifier: i64,
+}
+
+/// Invoked in registration order after snapshot publication is confirmed.
+///
+/// Implementations must be idempotent: a lost response can cause notification
+/// to be replayed from the committed snapshot during in-process recovery.
+/// Filtering previously committed identifiers does not invoke this notification.
+/// Errors are propagated without undoing publication or deleting files. The caller owns callback resources;
+/// the committer holds shared references, also across clones.
+#[async_trait]
+pub trait CommitCallback: Send + Sync {
+ async fn call(&self, context: &CommitCallbackContext) -> Result<()>;
+}
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index fe7f8ced..d1bb0057 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -40,6 +40,7 @@ mod bucket_assigner_fixed;
mod bucket_filter;
mod bucket_function;
mod chunk_shuffle;
+mod commit_callback;
mod commit_message;
mod consumer_manager;
pub(crate) mod cow_writer;
@@ -182,6 +183,7 @@ pub use audit_log_table::AuditLogTable;
pub use batch_vector_search_builder::BatchVectorSearchBuilder;
pub use blob_resolver::{BlobReader, BlobStream};
pub use branch_manager::BranchManager;
+pub use commit_callback::{CommitCallback, CommitCallbackContext};
pub use commit_message::{CommitMessage, COMMIT_MESSAGE_SERIALIZER_VERSION};
pub use consumer_manager::ConsumerManager;
pub use cow_writer::{CopyOnWriteMergeWriter, FileInfo};
diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs
index 317ac221..b71c8cee 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -30,6 +30,7 @@ use crate::spec::{
ManifestFileMeta, ManifestList, PartitionComputer, PartitionStatistics, Predicate, Snapshot,
EMPTY_SERIALIZED_ROW, MANIFEST_ENTRY_SCHEMA, POSTPONE_BUCKET,
};
+use crate::table::commit_callback::{CommitCallback, CommitCallbackContext};
use crate::table::commit_message::CommitMessage;
use crate::table::format_table_commit::FormatTableCommit;
use crate::table::global_index_build_common::same_extra_field_ids;
@@ -208,6 +209,7 @@ pub struct TableCommit {
data_evolution_enabled: bool,
partition_default_name: String,
snapshot_properties: Option<HashMap<String, String>>,
+ commit_callbacks: Vec<Arc<dyn CommitCallback>>,
}
impl TableCommit {
@@ -250,9 +252,37 @@ impl TableCommit {
data_evolution_enabled,
partition_default_name,
snapshot_properties: None,
+ commit_callbacks: Vec::new(),
}
}
+ /// Register an idempotent notification for confirmed Paimon snapshots.
+ pub fn add_commit_callback(&mut self, callback: Arc<dyn CommitCallback>) -> Result<()> {
+ if self.table.is_format_table() {
+ return Err(crate::Error::Unsupported {
+ message: "Commit callbacks require a Paimon table snapshot".into(),
+ });
+ }
+ self.commit_callbacks.push(callback);
+ Ok(())
+ }
+
+ async fn notify_commit_callbacks(
+ &self,
+ snapshot: Snapshot,
+ commit_entries: Vec<ManifestEntry>,
+ ) -> Result<()> {
+ let context = CommitCallbackContext {
+ identifier: snapshot.commit_identifier(),
+ snapshot,
+ commit_entries,
+ };
+ for callback in &self.commit_callbacks {
+ callback.call(&context).await?;
+ }
+ Ok(())
+ }
+
/// Control empty APPEND snapshots. Java stream commits set this to false.
pub fn with_ignore_empty_commit(mut self, ignore_empty_commit: bool) -> Self {
self.ignore_empty_commit = ignore_empty_commit;
@@ -1017,8 +1047,8 @@ impl TableCommit {
filter_committed = false;
}
if let Some(start_snapshot_id) = duplicate_check_start_snapshot_id {
- if self
- .is_duplicate_commit(
+ if let Some(snapshot) = self
+ .find_duplicate_commit(
start_snapshot_id,
&latest_snapshot,
commit_identifier,
@@ -1026,6 +1056,10 @@ impl TableCommit {
)
.await?
{
+ if !self.commit_callbacks.is_empty() {
+ let entries = self.read_delta_entries(None, &snapshot).await?;
+ self.notify_commit_callbacks(snapshot, entries).await?;
+ }
break;
}
}
@@ -1195,7 +1229,13 @@ impl TableCommit {
)
.await
{
- Ok(true) => return Ok(CommitAttemptResult::Success),
+ Ok(true) => {
+ if !self.commit_callbacks.is_empty() {
+ self.notify_commit_callbacks(snapshot, resolved.entries)
+ .await?;
+ }
+ return Ok(CommitAttemptResult::Success);
+ }
Ok(false) => None,
Err(error) => Some(error),
};
@@ -1680,13 +1720,13 @@ impl TableCommit {
}
/// Check if this commit was already completed during an in-process retry.
- async fn is_duplicate_commit(
+ async fn find_duplicate_commit(
&self,
start_snapshot_id: i64,
latest_snapshot: &Option<Snapshot>,
commit_identifier: i64,
commit_kind: &CommitKind,
- ) -> Result<bool> {
+ ) -> Result<Option<Snapshot>> {
if let Some(latest) = latest_snapshot {
for snapshot_id in (start_snapshot_id..=latest.id()).rev() {
let snap = if snapshot_id == latest.id() {
@@ -1698,11 +1738,11 @@ impl TableCommit {
&& snap.commit_identifier() == commit_identifier
&& snap.commit_kind() == commit_kind
{
- return Ok(true);
+ return Ok(Some(snap));
}
}
}
- Ok(false)
+ Ok(None)
}
/// Resolve commit entries and merge index entries based on the plan type.
@@ -3604,6 +3644,7 @@ mod tests {
include!("table_commit/row_tracking_tests.rs");
include!("table_commit/abort_tests.rs");
include!("table_commit/property_tests.rs");
+ include!("table_commit/callback_tests.rs");
}
#[tokio::test]
@@ -4374,13 +4415,15 @@ mod tests {
let snap_manager = SnapshotManager::new(file_io.clone(), table_path.to_string());
let latest = snap_manager.get_latest_snapshot().await.unwrap();
assert!(commit
- .is_duplicate_commit(1, &latest, 7, &CommitKind::APPEND)
+ .find_duplicate_commit(1, &latest, 7, &CommitKind::APPEND)
.await
- .unwrap());
- assert!(!commit
- .is_duplicate_commit(1, &latest, 8, &CommitKind::APPEND)
+ .unwrap()
+ .is_some());
+ assert!(commit
+ .find_duplicate_commit(1, &latest, 8, &CommitKind::APPEND)
.await
- .unwrap());
+ .unwrap()
+ .is_none());
}
#[tokio::test]
diff --git a/crates/paimon/src/table/table_commit/callback_tests.rs b/crates/paimon/src/table/table_commit/callback_tests.rs
new file mode 100644
index 00000000..a3e55daa
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/callback_tests.rs
@@ -0,0 +1,275 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Mutex;
+
+#[derive(Default)]
+struct RecordingCommitCallback {
+ contexts: Mutex<Vec<CommitCallbackContext>>,
+ fail: bool,
+}
+
+#[async_trait::async_trait]
+impl CommitCallback for RecordingCommitCallback {
+ async fn call(&self, context: &CommitCallbackContext) -> Result<()> {
+ self.contexts.lock().unwrap().push(context.clone());
+ if self.fail {
+ return Err(crate::Error::DataInvalid {
+ message: "callback rejected notification".into(),
+ source: None,
+ });
+ }
+ Ok(())
+ }
+}
+
+struct CallbackTestPublisher {
+ manager: SnapshotManager,
+ behavior: &'static str,
+ attempts: AtomicUsize,
+}
+
+#[async_trait::async_trait]
+impl SnapshotCommit for CallbackTestPublisher {
+ async fn commit(
+ &self,
+ _: Option<&str>,
+ snapshot: &Snapshot,
+ _: &str,
+ _: &[PartitionStatistics],
+ ) -> Result<bool> {
+ let attempt = self.attempts.fetch_add(1, Ordering::SeqCst);
+ if self.behavior == "reject" || (self.behavior == "conflict" && attempt == 0) {
+ return Ok(false);
+ }
+ assert!(self.manager.commit_snapshot(snapshot).await?);
+ if self.behavior == "lose-response" {
+ return Err(crate::Error::UnexpectedError {
+ message: "publication response lost".into(),
+ source: None,
+ });
+ }
+ Ok(true)
+ }
+}
+
+#[tokio::test]
+async fn commit_callback_gets_assigned_metadata_after_publication() {
+ for behavior in ["success", "lose-response", "conflict"] {
+ let io = test_file_io();
+ let path = format!("memory:/callback-{behavior}");
+ setup_dirs(&io, &path).await;
+ let table = test_table_with_options(
+ &io,
+ &path,
+ HashMap::from([("row-tracking.enabled".into(), "true".into())]),
+ );
+ let mut commit = TableCommit::new(table, "callback-user".into());
+ commit.commit_min_retry_wait_ms = 0;
+ commit.commit_max_retry_wait_ms = 0;
+ let publisher = Arc::new(CallbackTestPublisher {
+ manager: commit.snapshot_manager.clone(),
+ behavior,
+ attempts: AtomicUsize::new(0),
+ });
+ commit.snapshot_commit = publisher.clone();
+ let callback = Arc::new(RecordingCommitCallback::default());
+ commit.add_commit_callback(callback.clone()).unwrap();
+ let mut message = append_message("data.parquet");
+ message.partition = vec![0, 0, 0, 0];
+ message.new_files[0].file_source = Some(0);
+ let mut min = BinaryRowBuilder::new(2);
+ min.write_int(0, 1);
+ min.write_string(1, "first");
+ let mut max = BinaryRowBuilder::new(2);
+ max.write_int(0, 10);
+ max.write_string(1, "last");
+ let expected_stats = BinaryTableStats::new(
+ min.build().to_serialized_bytes(),
+ max.build().to_serialized_bytes(),
+ vec![Some(0), Some(1)],
+ );
+ message.new_files[0].value_stats = expected_stats.clone();
+ message
+ .new_changelog_files
+ .push(test_data_file("changelog.parquet", 10));
+ let properties = HashMap::from([("source".into(), "test".into())]);
+ commit
+ .clone()
+ .with_snapshot_properties(properties.clone())
+ .commit_with_identifier(vec![message], 7)
+ .await
+ .unwrap();
+ let latest = commit
+ .snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ let persisted_entries = commit.read_delta_entries(None, &latest).await.unwrap();
+ let contexts = callback.contexts.lock().unwrap();
+ assert_eq!(contexts.len(), 1);
+ let context = &contexts[0];
+ assert_eq!(context.snapshot, latest);
+ assert_eq!(context.identifier, 7);
+ assert_eq!(context.snapshot.properties(), Some(&properties));
+ assert_eq!(
+ context.commit_entries.len(),
+ 1,
+ "changelog is not a data-file delta"
+ );
+ let file = context.commit_entries[0].file();
+ assert_eq!(file.value_stats, expected_stats);
+ assert_eq!(file.first_row_id, Some(0));
+ assert_eq!(file.min_sequence_number, 1);
+ assert_eq!(file.max_sequence_number, 1);
+ assert_eq!(context.commit_entries, persisted_entries);
+ assert_eq!(
+ publisher.attempts.load(Ordering::SeqCst),
+ if behavior == "conflict" { 2 } else { 1 }
+ );
+ }
+}
+
+#[tokio::test]
+async fn commit_callback_overwrite_and_truncate_include_deletes() {
+ let io = test_file_io();
+ let path = "memory:/callback-overwrite";
+ setup_dirs(&io, path).await;
+ let mut commit = setup_commit(&io, path);
+ commit
+ .commit_with_identifier(vec![append_message("old.parquet")], 1)
+ .await
+ .unwrap();
+ let callback = Arc::new(RecordingCommitCallback::default());
+ commit.add_commit_callback(callback.clone()).unwrap();
+ commit
+ .overwrite(vec![append_message("new.parquet")], None)
+ .await
+ .unwrap();
+ commit.truncate_table().await.unwrap();
+ let contexts = callback.contexts.lock().unwrap();
+ assert_eq!(contexts.len(), 2);
+ let delta = &contexts[0].commit_entries;
+ assert_eq!(contexts[0].snapshot.commit_kind(), &CommitKind::OVERWRITE);
+ assert_eq!(delta.len(), 2);
+ assert_eq!(*delta[0].kind(), FileKind::Delete);
+ assert_eq!(delta[0].file().file_name, "old.parquet");
+ assert_eq!(*delta[1].kind(), FileKind::Add);
+ assert_eq!(delta[1].file().file_name, "new.parquet");
+ assert_eq!(contexts[1].commit_entries.len(), 1);
+ assert_eq!(*contexts[1].commit_entries[0].kind(), FileKind::Delete);
+}
+
+#[tokio::test]
+async fn commit_callback_error_preserves_published_files_and_stops_later_callbacks() {
+ for behavior in ["success", "lose-response"] {
+ let io = test_file_io();
+ let path = format!("memory:/callback-error-{behavior}");
+ setup_dirs(&io, &path).await;
+ let data_path = format!("{path}/bucket-0/data.parquet");
+ io.new_output(&data_path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"data"))
+ .await
+ .unwrap();
+ let mut commit = setup_commit(&io, &path);
+ commit.commit_min_retry_wait_ms = 0;
+ commit.commit_max_retry_wait_ms = 0;
+ let publisher = Arc::new(CallbackTestPublisher {
+ manager: commit.snapshot_manager.clone(),
+ behavior,
+ attempts: AtomicUsize::new(0),
+ });
+ commit.snapshot_commit = publisher.clone();
+ let first = Arc::new(RecordingCommitCallback {
+ fail: true,
+ ..Default::default()
+ });
+ let second = Arc::new(RecordingCommitCallback::default());
+ commit.add_commit_callback(first.clone()).unwrap();
+ commit.add_commit_callback(second.clone()).unwrap();
+ let error = commit
+ .commit_with_identifier(vec![append_message("data.parquet")], 7)
+ .await
+ .unwrap_err();
+ assert!(format!("{error:?}").contains("callback rejected notification"));
+ let latest = commit
+ .snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(latest.id(), 1);
+ assert_eq!(
+ commit
+ .read_delta_entries(None, &latest)
+ .await
+ .unwrap()
+ .len(),
+ 1
+ );
+ assert!(io.exists(&data_path).await.unwrap());
+ assert_eq!(first.contexts.lock().unwrap().len(), 1);
+ assert!(second.contexts.lock().unwrap().is_empty());
+ assert_eq!(publisher.attempts.load(Ordering::SeqCst), 1);
+ }
+}
+
+#[tokio::test]
+async fn commit_callback_only_runs_for_confirmed_snapshots() {
+ let io = test_file_io();
+ let path = "memory:/callback-empty-rejected";
+ setup_dirs(&io, path).await;
+ let mut commit = setup_commit(&io, path);
+ let callback = Arc::new(RecordingCommitCallback::default());
+ commit.add_commit_callback(callback.clone()).unwrap();
+ commit.commit_with_identifier(Vec::new(), 1).await.unwrap();
+ assert!(callback.contexts.lock().unwrap().is_empty());
+ commit
+ .clone()
+ .with_ignore_empty_commit(false)
+ .commit_with_identifier(Vec::new(), 2)
+ .await
+ .unwrap();
+ assert_eq!(callback.contexts.lock().unwrap().len(), 1);
+ assert!(callback.contexts.lock().unwrap()[0]
+ .commit_entries
+ .is_empty());
+ commit.commit_max_retries = 0;
+ commit.snapshot_commit = Arc::new(CallbackTestPublisher {
+ manager: commit.snapshot_manager.clone(),
+ behavior: "reject",
+ attempts: AtomicUsize::new(0),
+ });
+ assert!(commit
+ .commit_with_identifier(vec![append_message("rejected.parquet")], 3)
+ .await
+ .is_err());
+ assert_eq!(callback.contexts.lock().unwrap().len(), 1);
+ assert_eq!(
+ commit
+ .snapshot_manager
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ 1
+ );
+}