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 1ee2a104 [Core] Support caller snapshot properties in native commits
(#1104)
1ee2a104 is described below
commit 1ee2a104aeafd2c907f99ad95e9a29529d2b7e18
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 11 19:49:21 2026 +0800
[Core] Support caller snapshot properties in native commits (#1104)
---
.../python/python/pypaimon_rust/datafusion.pyi | 10 +-
bindings/python/src/write.rs | 31 +++-
bindings/python/tests/test_table_commit.py | 51 ++++++
crates/paimon/src/table/table_commit.rs | 16 ++
.../src/table/table_commit/property_tests.rs | 182 +++++++++++++++++++++
5 files changed, 279 insertions(+), 11 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 585f0f71..e0414953 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -425,7 +425,10 @@ class StreamTableWrite:
class BatchTableCommit:
def close(self) -> None: ...
- def commit(self, messages: Sequence[CommitMessage]) -> None:
+ def commit(
+ self, messages: Sequence[CommitMessage],
+ snapshot_properties: Optional[Dict[str, str]] = None,
+ ) -> None:
"""Commit once, using the builder's overwrite configuration and batch
identifier."""
...
def truncate_table(self) -> None:
@@ -440,7 +443,10 @@ class BatchTableCommit:
class StreamTableCommit:
def close(self) -> None: ...
- def commit(self, commit_identifier: int, messages:
Sequence[CommitMessage]) -> None:
+ def commit(
+ self, commit_identifier: int, messages: Sequence[CommitMessage],
+ snapshot_properties: Optional[Dict[str, str]] = None,
+ ) -> None:
"""Commit a checkpoint, including empty checkpoints, without filtering
retries."""
...
def filter_and_commit(
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 7f1d0fd8..fffe64b9 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -1324,6 +1324,14 @@ impl CommitContext {
py.detach(|| runtime().block_on(self.inner.abort(&messages)))
.map_err(to_py_err)
}
+
+ 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.
+ self.inner
+ .clone()
+ .with_snapshot_properties(properties.unwrap_or_default())
+ }
}
#[pyclass(name = "BatchTableCommit", module = "pypaimon_rust.datafusion")]
@@ -1351,18 +1359,24 @@ impl PyBatchTableCommit {
/// Rust committers have no background resources to shut down.
fn close(&self) {}
- fn commit(&mut self, py: Python<'_>, messages: &Bound<'_, PyAny>) ->
PyResult<()> {
+ #[pyo3(signature = (messages, snapshot_properties=None))]
+ fn commit(
+ &mut self,
+ py: Python<'_>,
+ messages: &Bound<'_, PyAny>,
+ snapshot_properties: Option<HashMap<String, String>>,
+ ) -> PyResult<()> {
let messages = self.context.messages(messages, "commit",
self.overwrite)?;
self.check_committed()?;
+ let committer = self.context.committer(snapshot_properties);
py.detach(|| {
runtime().block_on(async {
if self.overwrite {
- self.context
- .inner
+ committer
.overwrite(messages, self.static_partition.clone())
.await
} else {
- self.context.inner.commit(messages).await
+ committer.commit(messages).await
}
})
})
@@ -1403,19 +1417,18 @@ impl PyStreamTableCommit {
/// Rust committers have no background resources to shut down.
fn close(&self) {}
+ #[pyo3(signature = (commit_identifier, messages,
snapshot_properties=None))]
fn commit(
&self,
py: Python<'_>,
commit_identifier: i64,
messages: &Bound<'_, PyAny>,
+ snapshot_properties: Option<HashMap<String, String>>,
) -> PyResult<()> {
let messages = self.context.messages(messages, "commit", false)?;
+ let committer = self.context.committer(snapshot_properties);
py.detach(|| {
- runtime().block_on(
- self.context
- .inner
- .commit_with_identifier(messages, commit_identifier),
- )
+ runtime().block_on(committer.commit_with_identifier(messages,
commit_identifier))
})
.map_err(to_py_err)
}
diff --git a/bindings/python/tests/test_table_commit.py
b/bindings/python/tests/test_table_commit.py
index 00725b0e..ffb69285 100644
--- a/bindings/python/tests/test_table_commit.py
+++ b/bindings/python/tests/test_table_commit.py
@@ -155,6 +155,57 @@ def
test_deserialized_batch_commit_uses_target_builder(tmp_path):
assert _rows(table) == [1]
[email protected]("overwrite", [False, True])
[email protected]("properties", [None, {}, {"source": "capture",
"label": "中文"}])
+def test_batch_snapshot_properties(tmp_path, overwrite, properties):
+ table = _table(tmp_path)
+ _append(table, [1], [10])
+ builder = table.new_batch_write_builder()
+ if overwrite:
+ builder.with_overwrite()
+ messages = _roundtrip(_prepare(builder, [2], [20]))
+ committer = builder.new_commit()
+ committer.commit(messages, snapshot_properties=properties)
+ snapshot = _snapshot(table)
+ assert snapshot.get("properties") == (properties or None)
+ assert snapshot["commitKind"] == ("OVERWRITE" if overwrite else "APPEND")
+ assert _rows(table) == ([2] if overwrite else [1, 2])
+ with pytest.raises(RuntimeError, match="one-time"):
+ committer.commit(messages, snapshot_properties={"source": "retry"})
+ assert _snapshot(table)["id"] == snapshot["id"]
+
+
+def test_stream_snapshot_properties_are_per_checkpoint(tmp_path):
+ table = _table(tmp_path)
+ builder =
table.new_stream_write_builder().with_commit_user("properties-job")
+ committer = builder.new_commit()
+ properties = {"checkpoint": "first"}
+ for identifier, metadata in enumerate([properties, None, {},
{"checkpoint": "fourth"}], 1):
+ messages = _prepare(builder, [identifier], [identifier * 10],
identifier)
+ committer.commit(identifier, messages, snapshot_properties=metadata)
+ snapshot = _snapshot(table)
+ assert snapshot["commitIdentifier"] == identifier
+ assert snapshot.get("properties") == (metadata or None)
+ if not metadata:
+ assert "properties" not in snapshot
+ properties["checkpoint"] = "changed"
+ first = json.loads((Path(table.location()) / "snapshot" /
"snapshot-1").read_text())
+ assert first["properties"] == {"checkpoint": "first"}
+ assert _rows(table) == [1, 2, 3, 4]
+
+
+def test_empty_stream_snapshot_properties(tmp_path):
+ table = _table(tmp_path)
+ committer =
table.new_stream_write_builder().with_commit_user("empty-job").new_commit()
+ committer.commit(42, [], snapshot_properties={"source": "empty"})
+ snapshot = _snapshot(table)
+ assert snapshot["properties"] == {"source": "empty"}
+ assert snapshot["totalRecordCount"] == 0
+ assert snapshot["deltaRecordCount"] == 0
+ committer.commit(43, [])
+ assert "properties" not in _snapshot(table)
+
+
@pytest.mark.parametrize("overwrite", [False, True])
@pytest.mark.parametrize("serialized", [False, True])
def test_batch_bridge_preserves_external_commit_user(tmp_path, overwrite,
serialized):
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 698bea47..d4d2dba8 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -188,6 +188,7 @@ fn reject_compact_increment(messages: &[CommitMessage]) ->
Result<()> {
/// Table commit logic for Paimon write operations.
///
/// Provides atomic commit functionality including append, overwrite and
truncate
+#[derive(Clone)]
pub struct TableCommit {
table: Table,
snapshot_manager: SnapshotManager,
@@ -206,6 +207,7 @@ pub struct TableCommit {
row_tracking_enabled: bool,
data_evolution_enabled: bool,
partition_default_name: String,
+ snapshot_properties: Option<HashMap<String, String>>,
}
impl TableCommit {
@@ -247,6 +249,7 @@ impl TableCommit {
row_tracking_enabled,
data_evolution_enabled,
partition_default_name,
+ snapshot_properties: None,
}
}
@@ -256,6 +259,17 @@ impl TableCommit {
self
}
+ /// Attach caller properties to snapshots published by this committer.
+ ///
+ /// Like Java's ManifestCommittable properties, these describe the current
+ /// commit and are not inherited from the previous snapshot. Empty maps are
+ /// omitted. Clone the committer to configure metadata for a single call
+ /// while retaining its publication environment and retry configuration.
+ pub fn with_snapshot_properties(mut self, properties: HashMap<String,
String>) -> Self {
+ self.snapshot_properties =
(!properties.is_empty()).then_some(properties);
+ self
+ }
+
/// Java StreamTableCommit.filterAndCommit: sort identifiers and return the
/// number of groups remaining after filtering against committed snapshots.
pub async fn filter_and_commit(
@@ -1364,6 +1378,7 @@ impl TableCommit {
.statistics(statistics_file)
.next_row_id(next_row_id)
.index_manifest(resolved.index_manifest_name.clone())
+ .properties(self.snapshot_properties.clone())
.build();
let statistics =
self.generate_partition_statistics(&resolved.entries)?;
@@ -3600,6 +3615,7 @@ mod tests {
include!("table_commit/recovery_tests.rs");
include!("table_commit/row_tracking_tests.rs");
include!("table_commit/abort_tests.rs");
+ include!("table_commit/property_tests.rs");
}
#[tokio::test]
diff --git a/crates/paimon/src/table/table_commit/property_tests.rs
b/crates/paimon/src/table/table_commit/property_tests.rs
new file mode 100644
index 00000000..4d97e6af
--- /dev/null
+++ b/crates/paimon/src/table/table_commit/property_tests.rs
@@ -0,0 +1,182 @@
+// 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.
+
+#[tokio::test]
+async fn snapshot_properties_are_not_inherited_between_commits() {
+ let io = test_file_io();
+ let path = "memory:/snapshot-call-properties";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path).with_ignore_empty_commit(false);
+ let properties = HashMap::from([
+ ("source".into(), "capture".into()),
+ ("label".into(), "中文".into()),
+ ]);
+ let configured =
commit.clone().with_snapshot_properties(properties.clone());
+ configured
+ .commit_with_identifier(vec![append_message("data.parquet")], 1)
+ .await
+ .unwrap();
+ let first = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(first.properties(), Some(&properties));
+
+ // Empty snapshots also carry current-call metadata, not inherited values.
+ commit.commit_with_identifier(Vec::new(), 2).await.unwrap();
+ let second = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(second.properties(), None);
+ assert!(serde_json::to_value(&second)
+ .unwrap()
+ .get("properties")
+ .is_none());
+
+ configured
+ .clone()
+ .with_snapshot_properties(HashMap::new())
+ .commit_with_identifier(Vec::new(), 3)
+ .await
+ .unwrap();
+ assert_eq!(latest_snapshot(&io, path).await.unwrap().properties(), None);
+
+ configured
+ .commit_with_identifier(Vec::new(), 4)
+ .await
+ .unwrap();
+ let fourth = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(fourth.properties(), Some(&properties));
+ assert_eq!(fourth.delta_record_count(), Some(0));
+ assert_eq!(fourth.total_record_count(), first.total_record_count());
+}
+
+#[tokio::test]
+async fn snapshot_properties_overwrite_recovery_preserves_original_metadata() {
+ let io = test_file_io();
+ let path = "memory:/snapshot-overwrite-properties";
+ setup_dirs(&io, path).await;
+ let commit = setup_commit(&io, path);
+ commit
+ .commit_with_identifier(vec![append_message("old.parquet")], 1)
+ .await
+ .unwrap();
+ let properties = HashMap::from([("source".into(), "overwrite".into())]);
+ let configured =
commit.clone().with_snapshot_properties(properties.clone());
+ configured
+ .overwrite_with_identifier(Vec::new(), None, 2)
+ .await
+ .unwrap();
+ let snapshot = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(snapshot.properties(), Some(&properties));
+ assert_eq!(snapshot.commit_kind(), &CommitKind::OVERWRITE);
+ assert_eq!(snapshot.total_record_count(), Some(0));
+
+ let other = TableCommit::new(test_table(&io, path), "other-user".into());
+ other
+ .commit_with_identifier(vec![append_message("intervening.parquet")], 1)
+ .await
+ .unwrap();
+ // Recovered identities retain their original metadata and do not apply
+ // the caller's new properties or remove an intervening writer's data.
+ configured
+ .with_snapshot_properties(HashMap::from([("source".into(),
"retry".into())]))
+ .overwrite_with_identifier(Vec::new(), None, 2)
+ .await
+ .unwrap();
+ let latest = latest_snapshot(&io, path).await.unwrap();
+ assert_eq!(latest.id(), 3);
+ assert_eq!(latest.properties(), None);
+ assert_eq!(active_entries(&io, path, &latest).await.len(), 1);
+ assert_eq!(
+ commit
+ .snapshot_manager
+ .get_snapshot(2)
+ .await
+ .unwrap()
+ .properties(),
+ Some(&properties)
+ );
+}
+
+struct PropertyPublicationFault {
+ manager: SnapshotManager,
+ response_lost: bool,
+ calls: std::sync::atomic::AtomicUsize,
+ attempts: std::sync::Mutex<Vec<Snapshot>>,
+}
+
+#[async_trait::async_trait]
+impl SnapshotCommit for PropertyPublicationFault {
+ async fn commit(
+ &self,
+ _base_snapshot_uuid: Option<&str>,
+ snapshot: &Snapshot,
+ _branch: &str,
+ _statistics: &[PartitionStatistics],
+ ) -> Result<bool> {
+ self.attempts.lock().unwrap().push(snapshot.clone());
+ let first = self.calls.fetch_add(1,
std::sync::atomic::Ordering::SeqCst) == 0;
+ if first && !self.response_lost {
+ return Ok(false);
+ }
+ let published = self.manager.commit_snapshot(snapshot).await?;
+ if first && self.response_lost {
+ return Err(crate::Error::UnexpectedError {
+ message: "response lost after publication".into(),
+ source: None,
+ });
+ }
+ Ok(published)
+ }
+}
+
+#[tokio::test]
+async fn snapshot_properties_survive_conflict_and_uncertain_publication() {
+ for response_lost in [false, true] {
+ let io = test_file_io();
+ let path =
format!("memory:/snapshot-properties-fault-{response_lost}");
+ setup_dirs(&io, &path).await;
+ let mut commit = setup_commit(&io, &path);
+ commit.commit_min_retry_wait_ms = 0;
+ commit.commit_max_retry_wait_ms = 0;
+ let fault = Arc::new(PropertyPublicationFault {
+ manager: commit.snapshot_manager.clone(),
+ response_lost,
+ calls: std::sync::atomic::AtomicUsize::new(0),
+ attempts: std::sync::Mutex::new(Vec::new()),
+ });
+ commit.snapshot_commit = fault.clone();
+ let data_path = format!("{path}/bucket-0/prepared.parquet");
+ io.new_output(&data_path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"prepared file retained"))
+ .await
+ .unwrap();
+ let properties = HashMap::from([("source".into(), "retried".into())]);
+ commit
+ .with_snapshot_properties(properties.clone())
+ .commit_with_identifier(vec![append_message("prepared.parquet")],
7)
+ .await
+ .unwrap();
+ let snapshot = latest_snapshot(&io, &path).await.unwrap();
+ assert_eq!(snapshot.id(), 1);
+ assert_eq!(snapshot.properties(), Some(&properties));
+ assert_eq!(active_entries(&io, &path, &snapshot).await.len(), 1);
+ assert!(io.exists(&data_path).await.unwrap());
+ let attempts = fault.attempts.lock().unwrap();
+ assert_eq!(attempts.len(), if response_lost { 1 } else { 2 });
+ assert!(attempts
+ .iter()
+ .all(|attempt| attempt.properties() == Some(&properties)));
+ }
+}
