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 c270543f [python] Preserve external batch commit identity (#916)
c270543f is described below
commit c270543f20570772b91ce5804ffe39e6aa34a4ac
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 22 23:05:54 2026 +0800
[python] Preserve external batch commit identity (#916)
---
.../python/python/pypaimon_rust/datafusion.pyi | 3 ++
bindings/python/src/write.rs | 14 +++++++++
bindings/python/tests/test_table_commit.py | 33 +++++++++++++++++++++-
3 files changed, 49 insertions(+), 1 deletion(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index b3ebed81..8c40c14d 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -252,6 +252,9 @@ class StreamTableCommit:
def abort(self, messages: Sequence[CommitMessage]) -> None: ...
class BatchWriteBuilder:
+ def _with_commit_user(self, commit_user: str) -> "BatchWriteBuilder":
+ """Internal PyPaimon bridge for commits produced by an external Python
writer."""
+ ...
def with_overwrite(self, static_partition: Optional[Dict[str, Any]] = {})
-> "BatchWriteBuilder":
"""Configure both writer and committer. Explicit None restores append.
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 4caa28a5..258676e6 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -189,6 +189,20 @@ impl PyBatchWriteBuilder {
#[pymethods]
impl PyBatchWriteBuilder {
+ /// Internal PyPaimon bridge: retain the identity of the external Python
writer.
+ fn _with_commit_user(
+ mut slf: PyRefMut<'_, Self>,
+ commit_user: String,
+ ) -> PyResult<PyRefMut<'_, Self>> {
+ slf.context
+ .table
+ .new_write_builder()
+ .with_commit_user(commit_user.clone())
+ .map_err(to_py_err)?;
+ slf.context.commit_user = commit_user;
+ Ok(slf)
+ }
+
/// No argument enables overwrite with an empty spec; explicit None
restores append.
#[pyo3(signature = (static_partition=Some(HashMap::new())))]
fn with_overwrite<'py>(
diff --git a/bindings/python/tests/test_table_commit.py
b/bindings/python/tests/test_table_commit.py
index 12b6f666..2af7a926 100644
--- a/bindings/python/tests/test_table_commit.py
+++ b/bindings/python/tests/test_table_commit.py
@@ -91,7 +91,8 @@ def test_public_api_separates_batch_and_stream(tmp_path):
assert not hasattr(obj, "deserialize_commit_message")
assert not hasattr(batch.new_commit(), "filter_and_commit")
assert not hasattr(batch.new_commit(), "overwrite")
- assert not hasattr(stream.new_commit(), "truncate_table")
+ for method in ("overwrite", "_overwrite", "truncate_table"):
+ assert not hasattr(stream.new_commit(), method)
with pytest.raises(TypeError):
batch.new_commit().commit([], commit_identifier=1)
with pytest.raises(TypeError):
@@ -154,6 +155,36 @@ def
test_deserialized_batch_commit_uses_target_builder(tmp_path):
assert _rows(table) == [1]
[email protected]("overwrite", [False, True])
[email protected]("serialized", [False, True])
+def test_batch_bridge_preserves_external_commit_user(tmp_path, overwrite,
serialized):
+ table = _table(tmp_path, primary_key=True, options={"bucket": "1"})
+ _append(table, [1], [10])
+ builder = table.new_batch_write_builder()
+ assert builder._with_commit_user("python-batch-job") is builder
+ if overwrite:
+ builder.with_overwrite()
+ messages = _prepare(builder, [2], [20])
+ if serialized:
+ messages = _roundtrip(messages)
+ builder.new_commit().commit(messages)
+ assert _rows(table) == ([2] if overwrite else [1, 2])
+ snapshot = _snapshot(table)
+ assert snapshot["commitUser"] == "python-batch-job"
+ assert snapshot["commitIdentifier"] == 2**63 - 1
+
+
[email protected]("user", ["", "../job", "a/b"])
+def test_invalid_batch_bridge_commit_user_preserves_identity(tmp_path, user):
+ table = _table(tmp_path)
+ builder =
table.new_batch_write_builder()._with_commit_user("python-batch-job")
+ with pytest.raises(ValueError):
+ builder._with_commit_user(user)
+ builder.new_commit().commit(_prepare(builder, [1], [10]))
+ assert _snapshot(table)["commitUser"] == "python-batch-job"
+ assert _rows(table) == [1]
+
+
def test_abort_serialized_messages_deletes_files(tmp_path):
table = _table(tmp_path)
builder = table.new_batch_write_builder()
