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 3ce22437 feat: Add Java-compatible CommitMessage serialization and
Python commit API (#912)
3ce22437 is described below
commit 3ce2243776b69ace304bb690ec8fdd6f75227c2d
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Sep 22 16:48:04 2026 +0800
feat: Add Java-compatible CommitMessage serialization and Python commit API
(#912)
---
bindings/python/README.md | 61 ++-
.../python/python/pypaimon_rust/datafusion.pyi | 79 ++-
bindings/python/src/context.rs | 9 +-
bindings/python/src/table.rs | 11 +-
bindings/python/src/write.rs | 604 ++++++++++++++++-----
bindings/python/tests/test_table_commit.py | 415 ++++++++++++++
bindings/python/tests/test_write.py | 46 +-
crates/paimon/src/spec/index_file_meta.rs | 217 ++++++++
crates/paimon/src/table/commit_message.rs | 258 ++++++++-
crates/paimon/src/table/data_file_writer.rs | 4 +-
.../src/table/dedicated_format_file_writer.rs | 10 +
crates/paimon/src/table/kv_file_writer.rs | 19 +
crates/paimon/src/table/mod.rs | 2 +-
crates/paimon/src/table/postpone_file_writer.rs | 16 +
crates/paimon/src/table/table_commit.rs | 241 +++++++-
crates/paimon/src/table/table_write.rs | 13 +
docs/src/python-binding.md | 6 +-
17 files changed, 1817 insertions(+), 194 deletions(-)
diff --git a/bindings/python/README.md b/bindings/python/README.md
index 53f3a8ee..c55c4010 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -90,7 +90,7 @@ batch = pa.record_batch(
[[3, 4], ["charlie", "diana"]],
schema=pa.schema([("id", pa.int32()), ("name", pa.utf8())]),
)
-write_builder = table.new_write_builder()
+write_builder = table.new_batch_write_builder()
writer = write_builder.new_write()
writer.write_arrow(batch)
commit_messages = writer.prepare_commit()
@@ -107,6 +107,65 @@ print(f"\nRead: {batches_tt[0].num_rows} rows")
print(batches_tt[0])
```
+### Native commit from serialized messages
+
+The Python binding follows Java's batch/stream builder structure. Use
+`table.new_batch_write_builder()` for batch writes and
+`table.new_stream_write_builder().with_commit_user("ingest-job")` for
streaming.
+Both create writers and committers with the same commit identity.
+
+```python
+from pypaimon_rust.datafusion import CommitMessage
+
+builder = table.new_stream_write_builder().with_commit_user("ingest-job")
+committer = builder.new_commit()
+messages = [
+ CommitMessage.deserialize(body, version=14)
+ for body in serialized_messages
+]
+committer.commit(42, messages)
+
+# After an uncertain result, restore the same user and retry checkpoint groups.
+restored =
table.new_stream_write_builder().with_commit_user("ingest-job").new_commit()
+committed_groups = restored.filter_and_commit({42: messages})
+```
+
+Stream identifiers increase monotonically per commit user. `filter_and_commit`
+sorts them and returns the number of groups committed after filtering. Empty
+stream checkpoints create snapshots recording their identifiers. Batch
+`commit(messages)` uses Java's batch identifier and permits one attempt per
+committer. Batch empty commits follow `snapshot.ignore-empty-commit` (default
+true). `truncate_table()` shares the batch commit guard; `truncate_partitions`
+accepts a nonempty list of partition specs as in Java.
+
+Configure overwrite on the batch builder:
+
+```python
+builder = table.new_batch_write_builder().with_overwrite()
+writer = builder.new_write()
+writer.write_arrow(batch)
+builder.new_commit().commit(writer.prepare_commit())
+```
+
+This configures both writer and committer. For partitioned tables,
+`dynamic-partition-overwrite=true` (the default) replaces touched partitions,
+including when a static spec was supplied; empty input deletes nothing.
+With that option false, `with_overwrite(spec)` replaces matching partitions and
+`with_overwrite()` replaces all. Unpartitioned empty overwrite truncates the
+whole table. Explicit `with_overwrite(None)` restores append. Partition values
+use schema-compatible Python values; `None` or the default partition name means
+null. Batch writers permit one `prepare_commit()` call; reusable stream writers
+use `prepare_commit(wait_compaction, commit_identifier)`.
+
+The Java v14 body has no version header, table identity, commit user, or
overwrite
+mode. `CommitMessage.deserialize(body, version=14)` decodes it without a table
+or builder. Submit decoded messages to their originating table; commit identity
+and overwrite mode come from the configured committer. Messages returned
directly
+by local writers retain their table and commit-user checks.
+Only v14 is supported. `abort(messages)` deletes newly written files and must
+only be used for messages known not to have committed. Compact increments
remain
+unsupported by the Rust committer and are rejected.
+
### Tables resolved outside the Rust catalog
`Table.from_resolved_schema(location, schema_json, *, database="default",
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index 4d0811ab..b3ebed81 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -163,7 +163,8 @@ class Table:
def location(self) -> str: ...
def schema(self) -> TableSchema: ...
def new_read_builder(self, options: Optional[Dict[str, str]] = None) ->
ReadBuilder: ...
- def new_write_builder(self) -> "WriteBuilder": ...
+ def new_batch_write_builder(self) -> "BatchWriteBuilder": ...
+ def new_stream_write_builder(self) -> "StreamWriteBuilder": ...
def latest_snapshot(self) -> Optional[Snapshot]:
"""
Warning: This method blocks on a DataFusion runtime.
@@ -197,22 +198,78 @@ class Table:
"""
...
-class CommitMessage: ...
+class CommitMessage:
+ @staticmethod
+ def deserialize(data: bytes, *, version: int = 14) -> "CommitMessage":
+ """Decode a Java body; table, commit user and overwrite mode come from
the committer."""
+ ...
+ def serialize(self) -> bytes:
+ """Java CommitMessageSerializer v14 body, without a version header."""
+ ...
+
+class BatchTableWrite:
+ def close(self) -> None: ...
+ def write_arrow(self, batch: pyarrow.RecordBatch) -> None: ...
+ def prepare_commit(self) -> List[CommitMessage]:
+ """Prepare once per instance, including empty or failed attempts."""
+ ...
-class TableWrite:
+class StreamTableWrite:
+ def close(self) -> None: ...
def write_arrow(self, batch: pyarrow.RecordBatch) -> None: ...
- def prepare_commit(self) -> List[CommitMessage]: ...
+ def prepare_commit(self, wait_compaction: bool, commit_identifier: int) ->
List[CommitMessage]:
+ """Flush a checkpoint; submit its messages with the same identifier.
-class TableCommit:
- def commit(self, messages: Sequence[CommitMessage]) -> None: ...
+ Rust currently flushes synchronously without background compaction.
+ """
+ ...
+
+class BatchTableCommit:
+ def close(self) -> None: ...
+ def commit(self, messages: Sequence[CommitMessage]) -> None:
+ """Commit once, using the builder's overwrite configuration and batch
identifier."""
+ ...
+ def truncate_table(self) -> None:
+ """Truncate all data; shares the one-time guard with commit()."""
+ ...
+ def truncate_partitions(self, partitions: Sequence[Dict[str, Any]]) ->
None:
+ """Truncate matching partitions. Reject empty input, as Java does."""
+ ...
def abort(self, messages: Sequence[CommitMessage]) -> None:
- """Delete the files the messages refer to. Best-effort: missing files
and
- storage errors are ignored. The messages must not be committed
afterwards."""
+ """Best-effort deletion of newly written files; only abort uncommitted
messages."""
+ ...
+
+class StreamTableCommit:
+ def close(self) -> None: ...
+ def commit(self, commit_identifier: int, messages:
Sequence[CommitMessage]) -> None:
+ """Commit a checkpoint, including empty checkpoints, without filtering
retries."""
+ ...
+ def filter_and_commit(
+ self, commit_identifiers_and_messages: Dict[int,
Sequence[CommitMessage]]
+ ) -> int:
+ """Sort identifiers, skip committed groups, and return the number
committed."""
+ ...
+ def abort(self, messages: Sequence[CommitMessage]) -> None: ...
+
+class BatchWriteBuilder:
+ def with_overwrite(self, static_partition: Optional[Dict[str, Any]] = {})
-> "BatchWriteBuilder":
+ """Configure both writer and committer. Explicit None restores append.
+
+ For partitioned tables dynamic-partition-overwrite defaults to true:
+ replace touched partitions, irrespective of the static spec; empty
input
+ removes nothing. With the option false, match the spec ({} means all).
+ An unpartitioned empty overwrite truncates the table. Partition values
+ use the schema's Python types; None or the default partition name is
null.
+ """
...
+ def new_write(self) -> BatchTableWrite: ...
+ def new_commit(self) -> BatchTableCommit: ...
-class WriteBuilder:
- def new_write(self) -> TableWrite: ...
- def new_commit(self) -> TableCommit: ...
+class StreamWriteBuilder:
+ def commit_user(self) -> str: ...
+ def with_commit_user(self, commit_user: str) -> "StreamWriteBuilder": ...
+ def new_write(self) -> StreamTableWrite: ...
+ def new_commit(self) -> StreamTableCommit: ...
class PaimonCatalog:
def __init__(self, catalog_options: Dict[str, str]) -> None: ...
diff --git a/bindings/python/src/context.rs b/bindings/python/src/context.rs
index 1e285fd1..de431c60 100644
--- a/bindings/python/src/context.rs
+++ b/bindings/python/src/context.rs
@@ -548,9 +548,12 @@ pub fn register_module(py: Python<'_>, m: &Bound<'_,
PyModule>) -> PyResult<()>
this.add_class::<crate::schema::PyDataField>()?;
this.add_class::<PyPythonScalarUDFObject>()?;
this.add_class::<PySQLContext>()?;
- this.add_class::<crate::write::PyWriteBuilder>()?;
- this.add_class::<crate::write::PyTableWrite>()?;
- this.add_class::<crate::write::PyTableCommit>()?;
+ this.add_class::<crate::write::PyBatchWriteBuilder>()?;
+ this.add_class::<crate::write::PyStreamWriteBuilder>()?;
+ this.add_class::<crate::write::PyBatchTableWrite>()?;
+ this.add_class::<crate::write::PyStreamTableWrite>()?;
+ this.add_class::<crate::write::PyBatchTableCommit>()?;
+ this.add_class::<crate::write::PyStreamTableCommit>()?;
this.add_class::<crate::write::PyCommitMessage>()?;
this.add_function(wrap_pyfunction!(udf, &this)?)?;
this.add_class::<crate::snapshot::PySnapshot>()?;
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 50ca8072..06c6316c 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -32,7 +32,7 @@ use crate::read::PyReadBuilder;
use crate::schema::PyTableSchema;
use crate::snapshot::PySnapshot;
use crate::tag::PyTag;
-use crate::write::PyWriteBuilder;
+use crate::write::{PyBatchWriteBuilder, PyStreamWriteBuilder};
#[pyclass(name = "Table", module = "pypaimon_rust.datafusion")]
pub struct PyTable {
@@ -133,9 +133,12 @@ impl PyTable {
}
}
- /// Create a [`PyWriteBuilder`] for the batch write loop.
- fn new_write_builder(&self) -> PyWriteBuilder {
- PyWriteBuilder::new(Arc::clone(&self.inner))
+ fn new_batch_write_builder(&self) -> PyBatchWriteBuilder {
+ PyBatchWriteBuilder::new(Arc::clone(&self.inner))
+ }
+
+ fn new_stream_write_builder(&self) -> PyStreamWriteBuilder {
+ PyStreamWriteBuilder::new(Arc::clone(&self.inner))
}
// ---------------- #285: observability ----------------
diff --git a/bindings/python/src/write.rs b/bindings/python/src/write.rs
index 07f0d7d3..4caa28a5 100644
--- a/bindings/python/src/write.rs
+++ b/bindings/python/src/write.rs
@@ -15,17 +15,23 @@
// specific language governing permissions and limitations
// under the License.
+use std::collections::HashMap;
use std::sync::Arc;
use arrow::datatypes::Schema as ArrowSchema;
use arrow::pyarrow::FromPyArrow;
use arrow::record_batch::RecordBatch;
-use paimon::table::{CommitMessage, Table, TableCommit, TableWrite};
+use paimon::spec::{CoreOptions, DataType, Datum};
+use paimon::table::{
+ CommitMessage, Table, TableCommit, TableWrite,
COMMIT_MESSAGE_SERIALIZER_VERSION,
+};
use paimon_datafusion::runtime::runtime;
-use pyo3::exceptions::{PyTypeError, PyValueError};
+use pyo3::exceptions::{PyRuntimeError, PyTypeError, PyValueError};
use pyo3::prelude::*;
+use pyo3::types::{PyBytes, PyDict, PyString};
use crate::error::to_py_err;
+use crate::predicate::py_to_datum;
/// Validate an incoming batch schema against the table's target Arrow schema:
/// field count, order, and names must match, and types must match exactly. The
@@ -60,128 +66,491 @@ fn validate_batch_schema(input: &ArrowSchema, target:
&ArrowSchema) -> PyResult<
Ok(())
}
-/// Builder for the batch write loop, created via
[`crate::table::PyTable::new_write_builder`].
-///
-/// Holds the owning table plus a single fixed `commit_user`, generated once
and
-/// shared by both `new_write()` and `new_commit()` so that writers and the
-/// committer agree on the commit user (Paimon uses it for duplicate-commit
-/// detection). Creating a fresh `WriteBuilder` per call would otherwise mint a
-/// new random UUID each time.
-#[pyclass(name = "WriteBuilder", module = "pypaimon_rust.datafusion")]
-pub struct PyWriteBuilder {
+type PartitionSpec = HashMap<String, Option<Datum>>;
+type PythonPartitionSpec = HashMap<String, Py<PyAny>>;
+
+/// Shared implementation state; public builders keep batch and stream
contracts separate.
+struct WriteContext {
table: Arc<Table>,
commit_user: String,
}
-impl PyWriteBuilder {
- pub fn new(table: Arc<Table>) -> Self {
+impl WriteContext {
+ fn new(table: Arc<Table>) -> Self {
let commit_user = table.new_write_builder().commit_user().to_string();
Self { table, commit_user }
}
-}
-#[pymethods]
-impl PyWriteBuilder {
- /// Create a writer for accumulating Arrow batches.
- fn new_write(&self) -> PyResult<PyTableWrite> {
+ fn new_write(&self, overwrite: bool) -> PyResult<WriteState> {
let builder = self
.table
.new_write_builder()
.with_commit_user(self.commit_user.clone())
.map_err(to_py_err)?;
- let target_schema =
paimon::arrow::build_target_arrow_schema(self.table.schema().fields())
- .map_err(to_py_err)?;
- Ok(PyTableWrite {
- inner: builder.new_write().map_err(to_py_err)?,
- target_schema,
+ let builder = if overwrite {
+ builder.with_overwrite()
+ } else {
+ builder
+ };
+ Ok(WriteState {
+ inner: Some(builder.new_write().map_err(to_py_err)?),
+ target_schema:
paimon::arrow::build_target_arrow_schema(self.table.schema().fields())
+ .map_err(to_py_err)?,
table_location: self.table.location().to_string(),
commit_user: self.commit_user.clone(),
})
}
+}
- /// Create a committer for persisting prepared commit messages.
- fn new_commit(&self) -> PyResult<PyTableCommit> {
- let builder = self
+fn boolean_option(table: &Table, key: &str, default: bool) -> PyResult<bool> {
+ match table.schema().options().get(key) {
+ None => Ok(default),
+ Some(value) if value.eq_ignore_ascii_case("true") => Ok(true),
+ Some(value) if value.eq_ignore_ascii_case("false") => Ok(false),
+ Some(value) => Err(PyValueError::new_err(format!(
+ "Invalid boolean option {key}: {value}"
+ ))),
+ }
+}
+
+/// Java partition specs may encode numeric values as strings. Other Python
+/// literals use the same schema-driven conversion as the predicate API.
+fn partition_value(value: &Bound<'_, PyAny>, data_type: &DataType) ->
PyResult<Datum> {
+ if let Ok(text) = value.cast::<PyString>() {
+ let text = text.to_str()?;
+ let invalid = || {
+ PyValueError::new_err(format!(
+ "Invalid partition value '{text}' for {data_type:?}"
+ ))
+ };
+ match data_type {
+ DataType::TinyInt(_) => return
text.parse().map(Datum::TinyInt).map_err(|_| invalid()),
+ DataType::SmallInt(_) => {
+ return text.parse().map(Datum::SmallInt).map_err(|_| invalid())
+ }
+ DataType::Int(_) => return
text.parse().map(Datum::Int).map_err(|_| invalid()),
+ DataType::BigInt(_) => return
text.parse().map(Datum::Long).map_err(|_| invalid()),
+ DataType::Float(_) => return
text.parse().map(Datum::Float).map_err(|_| invalid()),
+ DataType::Double(_) => return
text.parse().map(Datum::Double).map_err(|_| invalid()),
+ _ => {}
+ }
+ }
+ py_to_datum(value, data_type)
+}
+
+fn partition_spec(
+ py: Python<'_>,
+ table: &Table,
+ spec: PythonPartitionSpec,
+) -> PyResult<PartitionSpec> {
+ let fields = table.schema().partition_fields();
+ let default_name = CoreOptions::new(table.schema().options())
+ .partition_default_name()
+ .to_string();
+ spec.into_iter()
+ .map(|(key, value)| {
+ let field = fields
+ .iter()
+ .find(|field| field.name() == key)
+ .ok_or_else(|| {
+ PyValueError::new_err(format!(
+ "Partition spec key '{key}' is not a partition column"
+ ))
+ })?;
+ let value = value.bind(py);
+ let is_default = value
+ .cast::<PyString>()
+ .is_ok_and(|s| s.to_str().is_ok_and(|s| s == default_name));
+ let datum = if value.is_none() || is_default {
+ None
+ } else {
+ Some(partition_value(value, field.data_type())?)
+ };
+ Ok((key, datum))
+ })
+ .collect()
+}
+
+/// Java BatchWriteBuilder: the overwrite spec configures both writer and
committer.
+#[pyclass(name = "BatchWriteBuilder", module = "pypaimon_rust.datafusion")]
+pub struct PyBatchWriteBuilder {
+ context: WriteContext,
+ static_partition: Option<PartitionSpec>,
+}
+
+impl PyBatchWriteBuilder {
+ pub fn new(table: Arc<Table>) -> Self {
+ Self {
+ context: WriteContext::new(table),
+ static_partition: None,
+ }
+ }
+}
+
+#[pymethods]
+impl PyBatchWriteBuilder {
+ /// No argument enables overwrite with an empty spec; explicit None
restores append.
+ #[pyo3(signature = (static_partition=Some(HashMap::new())))]
+ fn with_overwrite<'py>(
+ mut slf: PyRefMut<'py, Self>,
+ py: Python<'py>,
+ static_partition: Option<PythonPartitionSpec>,
+ ) -> PyResult<PyRefMut<'py, Self>> {
+ slf.static_partition = static_partition
+ .map(|spec| partition_spec(py, &slf.context.table, spec))
+ .transpose()?;
+ Ok(slf)
+ }
+
+ fn new_write(&self) -> PyResult<PyBatchTableWrite> {
+ Ok(PyBatchTableWrite {
+ state: self.context.new_write(self.static_partition.is_some())?,
+ prepared: false,
+ })
+ }
+
+ fn new_commit(&self) -> PyResult<PyBatchTableCommit> {
+ let table = &self.context.table;
+ let ignore_empty = boolean_option(table,
"snapshot.ignore-empty-commit", true)?;
+ let dynamic = boolean_option(table, "dynamic-partition-overwrite",
true)?
+ && !table.schema().partition_keys().is_empty();
+ Ok(PyBatchTableCommit {
+ context: CommitContext::new(table, &self.context.commit_user,
ignore_empty)?,
+ overwrite: self.static_partition.is_some(),
+ static_partition: if dynamic {
+ None
+ } else {
+ self.static_partition.clone()
+ },
+ committed: false,
+ })
+ }
+}
+
+/// Java StreamWriteBuilder: stable commit identity belongs on the builder.
+#[pyclass(name = "StreamWriteBuilder", module = "pypaimon_rust.datafusion")]
+pub struct PyStreamWriteBuilder {
+ context: WriteContext,
+}
+
+impl PyStreamWriteBuilder {
+ pub fn new(table: Arc<Table>) -> Self {
+ Self {
+ context: WriteContext::new(table),
+ }
+ }
+}
+
+#[pymethods]
+impl PyStreamWriteBuilder {
+ fn commit_user(&self) -> &str {
+ &self.context.commit_user
+ }
+
+ fn with_commit_user(
+ mut slf: PyRefMut<'_, Self>,
+ commit_user: String,
+ ) -> PyResult<PyRefMut<'_, Self>> {
+ slf.context
.table
.new_write_builder()
- .with_commit_user(self.commit_user.clone())
+ .with_commit_user(commit_user.clone())
.map_err(to_py_err)?;
- Ok(PyTableCommit {
- inner: builder.new_commit(),
- table_location: self.table.location().to_string(),
- commit_user: self.commit_user.clone(),
+ slf.context.commit_user = commit_user;
+ Ok(slf)
+ }
+
+ fn new_write(&self) -> PyResult<PyStreamTableWrite> {
+ Ok(PyStreamTableWrite {
+ state: self.context.new_write(false)?,
+ })
+ }
+
+ fn new_commit(&self) -> PyResult<PyStreamTableCommit> {
+ Ok(PyStreamTableCommit {
+ context: CommitContext::new(&self.context.table,
&self.context.commit_user, false)?,
})
}
}
-/// A stateful writer that accumulates Arrow batches until `prepare_commit`.
-///
-/// Marked `unsendable`: the underlying `TableWrite` holds file writers that
are
-/// not `Sync`, so the object enforces single-thread access at runtime.
-#[pyclass(name = "TableWrite", module = "pypaimon_rust.datafusion",
unsendable)]
-pub struct PyTableWrite {
- inner: TableWrite,
- /// The table's target Arrow schema, used to validate incoming batches.
+struct WriteState {
+ inner: Option<TableWrite>,
target_schema: Arc<ArrowSchema>,
- /// The owning table's location, stamped onto produced commit messages so a
- /// committer can reject messages prepared for a different table.
table_location: String,
- /// The originating builder's `commit_user`, stamped onto produced
messages so
- /// a committer can reject messages prepared by a different `WriteBuilder`
- /// (writers and committers from the same builder must share one
commit_user;
- /// it drives snapshot duplicate detection and postpone-bucket file
naming).
commit_user: String,
}
-#[pymethods]
-impl PyTableWrite {
- /// Write a single PyArrow RecordBatch into the table's writers.
+impl WriteState {
fn write_arrow(&mut self, py: Python<'_>, batch: &Bound<'_, PyAny>) ->
PyResult<()> {
let batch = RecordBatch::from_pyarrow_bound(batch)?;
validate_batch_schema(&batch.schema(), &self.target_schema)?;
- let rt = runtime();
- py.detach(|| rt.block_on(async {
self.inner.write_arrow_batch(&batch).await }))
+ let inner = self
+ .inner
+ .as_mut()
+ .ok_or_else(|| PyRuntimeError::new_err("TableWrite is closed"))?;
+ py.detach(|| runtime().block_on(inner.write_arrow_batch(&batch)))
.map_err(to_py_err)
}
- /// Close writers and return the commit messages (opaque; pass to
commit()).
fn prepare_commit(&mut self, py: Python<'_>) ->
PyResult<Vec<PyCommitMessage>> {
- let rt = runtime();
+ let inner = self
+ .inner
+ .as_mut()
+ .ok_or_else(|| PyRuntimeError::new_err("TableWrite is closed"))?;
let messages = py
- .detach(|| rt.block_on(async { self.inner.prepare_commit().await
}))
+ .detach(|| runtime().block_on(inner.prepare_commit()))
.map_err(to_py_err)?;
Ok(messages
.into_iter()
.map(|inner| PyCommitMessage {
inner,
- table_location: self.table_location.clone(),
- commit_user: self.commit_user.clone(),
+ origin: Some(MessageOrigin {
+ table_location: self.table_location.clone(),
+ commit_user: self.commit_user.clone(),
+ }),
})
.collect())
}
}
-/// A committer that persists prepared commit messages as a snapshot.
-#[pyclass(name = "TableCommit", module = "pypaimon_rust.datafusion")]
-pub struct PyTableCommit {
+#[pyclass(
+ name = "BatchTableWrite",
+ module = "pypaimon_rust.datafusion",
+ unsendable
+)]
+pub struct PyBatchTableWrite {
+ state: WriteState,
+ prepared: bool,
+}
+
+#[pymethods]
+impl PyBatchTableWrite {
+ fn close(&mut self, py: Python<'_>) {
+ if let Some(mut writer) = self.state.inner.take() {
+ py.detach(|| runtime().block_on(writer.close()));
+ }
+ }
+
+ fn write_arrow(&mut self, py: Python<'_>, batch: &Bound<'_, PyAny>) ->
PyResult<()> {
+ self.state.write_arrow(py, batch)
+ }
+
+ fn prepare_commit(&mut self, py: Python<'_>) ->
PyResult<Vec<PyCommitMessage>> {
+ if self.prepared {
+ return Err(PyRuntimeError::new_err(
+ "BatchTableWrite only supports one-time committing.",
+ ));
+ }
+ self.prepared = true;
+ self.state.prepare_commit(py)
+ }
+}
+
+#[pyclass(
+ name = "StreamTableWrite",
+ module = "pypaimon_rust.datafusion",
+ unsendable
+)]
+pub struct PyStreamTableWrite {
+ state: WriteState,
+}
+
+#[pymethods]
+impl PyStreamTableWrite {
+ fn close(&mut self, py: Python<'_>) {
+ if let Some(mut writer) = self.state.inner.take() {
+ py.detach(|| runtime().block_on(writer.close()));
+ }
+ }
+
+ fn write_arrow(&mut self, py: Python<'_>, batch: &Bound<'_, PyAny>) ->
PyResult<()> {
+ self.state.write_arrow(py, batch)
+ }
+
+ /// Rust currently flushes synchronously and has no background compaction.
+ /// The identifier accompanies the returned messages in
StreamTableCommit.commit.
+ fn prepare_commit(
+ &mut self,
+ py: Python<'_>,
+ wait_compaction: bool,
+ commit_identifier: i64,
+ ) -> PyResult<Vec<PyCommitMessage>> {
+ let _ = (wait_compaction, commit_identifier);
+ self.state.prepare_commit(py)
+ }
+}
+
+struct CommitContext {
inner: TableCommit,
- /// The owning table's location, used to reject commit messages that were
- /// prepared for a different table (which would otherwise persist a
snapshot
- /// referencing data files written under another table).
- table_location: String,
- /// The committer's `commit_user`, used to reject messages prepared by a
- /// different `WriteBuilder` — even for the same table — since the writer
and
- /// committer must share one commit_user.
+ table: Arc<Table>,
commit_user: String,
}
+impl CommitContext {
+ fn new(table: &Arc<Table>, commit_user: &str, ignore_empty: bool) ->
PyResult<Self> {
+ let inner = table
+ .new_write_builder()
+ .with_commit_user(commit_user)
+ .map_err(to_py_err)?
+ .try_new_commit()
+ .map_err(to_py_err)?
+ .with_ignore_empty_commit(ignore_empty);
+ Ok(Self {
+ inner,
+ table: Arc::clone(table),
+ commit_user: commit_user.to_string(),
+ })
+ }
+
+ fn messages(
+ &self,
+ messages: &Bound<'_, PyAny>,
+ method: &str,
+ overwrite: bool,
+ ) -> PyResult<Vec<CommitMessage>> {
+ collect_and_validate_messages(
+ messages,
+ self.table.location(),
+ &self.commit_user,
+ method,
+ overwrite,
+ )
+ }
+
+ fn abort(&self, py: Python<'_>, messages: &Bound<'_, PyAny>) ->
PyResult<()> {
+ let messages = self.messages(messages, "abort", false)?;
+ py.detach(|| runtime().block_on(self.inner.abort(&messages)))
+ .map_err(to_py_err)
+ }
+}
+
+#[pyclass(name = "BatchTableCommit", module = "pypaimon_rust.datafusion")]
+pub struct PyBatchTableCommit {
+ context: CommitContext,
+ overwrite: bool,
+ static_partition: Option<PartitionSpec>,
+ committed: bool,
+}
+
+impl PyBatchTableCommit {
+ fn check_committed(&mut self) -> PyResult<()> {
+ if self.committed {
+ return Err(PyRuntimeError::new_err(
+ "BatchTableCommit only supports one-time committing.",
+ ));
+ }
+ self.committed = true;
+ Ok(())
+ }
+}
+
+#[pymethods]
+impl PyBatchTableCommit {
+ /// Rust committers have no background resources to shut down.
+ fn close(&self) {}
+
+ fn commit(&mut self, py: Python<'_>, messages: &Bound<'_, PyAny>) ->
PyResult<()> {
+ let messages = self.context.messages(messages, "commit",
self.overwrite)?;
+ self.check_committed()?;
+ py.detach(|| {
+ runtime().block_on(async {
+ if self.overwrite {
+ self.context
+ .inner
+ .overwrite(messages, self.static_partition.clone())
+ .await
+ } else {
+ self.context.inner.commit(messages).await
+ }
+ })
+ })
+ .map_err(to_py_err)
+ }
+
+ 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)
+ }
+
+ fn truncate_partitions(
+ &self,
+ py: Python<'_>,
+ partitions: Vec<PythonPartitionSpec>,
+ ) -> PyResult<()> {
+ let partitions = partitions
+ .into_iter()
+ .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)
+ }
+
+ fn abort(&self, py: Python<'_>, messages: &Bound<'_, PyAny>) ->
PyResult<()> {
+ self.context.abort(py, messages)
+ }
+}
+
+#[pyclass(name = "StreamTableCommit", module = "pypaimon_rust.datafusion")]
+pub struct PyStreamTableCommit {
+ context: CommitContext,
+}
+
+#[pymethods]
+impl PyStreamTableCommit {
+ /// Rust committers have no background resources to shut down.
+ fn close(&self) {}
+
+ fn commit(
+ &self,
+ py: Python<'_>,
+ commit_identifier: i64,
+ messages: &Bound<'_, PyAny>,
+ ) -> PyResult<()> {
+ let messages = self.context.messages(messages, "commit", false)?;
+ py.detach(|| {
+ runtime().block_on(
+ self.context
+ .inner
+ .commit_with_identifier(messages, commit_identifier),
+ )
+ })
+ .map_err(to_py_err)
+ }
+
+ /// Sort identifiers, filter already committed groups, and return the
number committed.
+ fn filter_and_commit(
+ &self,
+ py: Python<'_>,
+ commit_identifiers_and_messages: &Bound<'_, PyDict>,
+ ) -> PyResult<usize> {
+ let commits = commit_identifiers_and_messages
+ .iter()
+ .map(|(id, messages)| {
+ Ok((
+ id.extract::<i64>()?,
+ self.context
+ .messages(&messages, "filter_and_commit", false)?,
+ ))
+ })
+ .collect::<PyResult<Vec<_>>>()?;
+ py.detach(||
runtime().block_on(self.context.inner.filter_and_commit(commits)))
+ .map_err(to_py_err)
+ }
+
+ fn abort(&self, py: Python<'_>, messages: &Bound<'_, PyAny>) ->
PyResult<()> {
+ self.context.abort(py, messages)
+ }
+}
+
/// Collect and validate commit messages from a Python iterable, returning the
-/// inner Rust `CommitMessage` values. Shared by `commit` and `abort`.
+/// inner Rust `CommitMessage` values for every operation accepting messages.
fn collect_and_validate_messages<'py>(
messages: &Bound<'py, PyAny>,
table_location: &str,
commit_user: &str,
method: &str,
+ overwrite: bool,
) -> PyResult<Vec<CommitMessage>> {
let mut inner_messages = Vec::new();
let iter = messages.try_iter().map_err(|_| {
@@ -196,68 +565,61 @@ fn collect_and_validate_messages<'py>(
"{method}() expects a sequence of CommitMessage objects"
))
})?;
- if msg.table_location != table_location {
- return Err(PyValueError::new_err(format!(
- "commit message was prepared for a different table \
- (message table '{}', committer table '{}')",
- msg.table_location, table_location
- )));
+ let mut inner = msg.inner.clone();
+ if let Some(origin) = &msg.origin {
+ if origin.table_location != table_location {
+ return Err(PyValueError::new_err(format!(
+ "commit message was prepared for a different table \
+ (message table '{}', committer table '{}')",
+ origin.table_location, table_location
+ )));
+ }
+ if origin.commit_user != commit_user {
+ return Err(PyValueError::new_err(
+ "commit message has a different commit_user \
+ (writer and committer must share one commit_user)"
+ .to_string(),
+ ));
+ }
+ } else if overwrite {
+ // The Java body has no operation flag. Apply the target
committer's
+ // mode to imported messages without changing the Python object.
+ inner.mark_fixed_bucket_overwrite();
}
- if msg.commit_user != commit_user {
- return Err(PyValueError::new_err(
- "commit message was prepared by a different WriteBuilder \
- (writer and committer must come from the same \
- table.new_write_builder() so they share one commit_user)"
- .to_string(),
- ));
- }
- inner_messages.push(msg.inner.clone());
+ inner_messages.push(inner);
}
Ok(inner_messages)
}
-#[pymethods]
-impl PyTableCommit {
- /// Commit the given commit messages. Empty input is a no-op success.
- fn commit(&self, py: Python<'_>, messages: &Bound<'_, PyAny>) ->
PyResult<()> {
- let inner_messages = collect_and_validate_messages(
- messages,
- &self.table_location,
- &self.commit_user,
- "commit",
- )?;
- let rt = runtime();
- py.detach(|| rt.block_on(async {
self.inner.commit(inner_messages).await }))
- .map_err(to_py_err)
- }
-
- /// Abort a prepared commit by deleting newly written data, changelog and
- /// index files. Deletion is best-effort: missing files or storage errors
- /// are silently ignored so abort cleanup never masks the original write
- /// failure. After abort the data must not be committed — the files no
- /// longer exist.
- fn abort(&self, py: Python<'_>, messages: &Bound<'_, PyAny>) ->
PyResult<()> {
- let inner_messages = collect_and_validate_messages(
- messages,
- &self.table_location,
- &self.commit_user,
- "abort",
- )?;
- let rt = runtime();
- py.detach(|| rt.block_on(async {
self.inner.abort(&inner_messages).await }))
- .map_err(to_py_err)
- }
+/// Origin information retained for messages returned directly by a local
writer.
+struct MessageOrigin {
+ table_location: String,
+ commit_user: String,
}
-/// An opaque commit message produced by `prepare_commit`, consumed by
`commit`.
-/// PR1 supports same-process transfer only (no pickle/serialization).
-///
-/// Carries the originating table's location and builder `commit_user` so a
-/// committer can reject messages prepared for a different table or by a
-/// different `WriteBuilder`.
+/// A commit message produced by `prepare_commit` or decoded from the Java v14
body.
+/// Serialized messages contain no table, commit user, or operation context.
#[pyclass(name = "CommitMessage", module = "pypaimon_rust.datafusion")]
pub struct PyCommitMessage {
- pub(crate) inner: CommitMessage,
- pub(crate) table_location: String,
- pub(crate) commit_user: String,
+ inner: CommitMessage,
+ origin: Option<MessageOrigin>,
+}
+
+#[pymethods]
+impl PyCommitMessage {
+ /// Export the Java `CommitMessageSerializer` v14 body (no version header).
+ fn serialize<'py>(&self, py: Python<'py>) -> PyResult<Bound<'py, PyBytes>>
{
+ let bytes = self.inner.serialize().map_err(to_py_err)?;
+ Ok(PyBytes::new(py, &bytes))
+ }
+
+ /// Decode a Java body independently of a table or write builder.
+ #[staticmethod]
+ #[pyo3(signature = (data, *, version=COMMIT_MESSAGE_SERIALIZER_VERSION))]
+ fn deserialize(data: &Bound<'_, PyBytes>, version: i32) -> PyResult<Self> {
+ Ok(Self {
+ inner: CommitMessage::deserialize(version,
data.as_bytes()).map_err(to_py_err)?,
+ origin: None,
+ })
+ }
}
diff --git a/bindings/python/tests/test_table_commit.py
b/bindings/python/tests/test_table_commit.py
new file mode 100644
index 00000000..b168b224
--- /dev/null
+++ b/bindings/python/tests/test_table_commit.py
@@ -0,0 +1,415 @@
+# 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.
+
+import base64
+import json
+from pathlib import Path
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+import pytest
+
+from pypaimon_rust import datafusion
+from pypaimon_rust.datafusion import CommitMessage, PaimonCatalog, SQLContext
+
+
+def _table(path, partitioned=False, options=None, primary_key=False):
+ ctx = SQLContext()
+ ctx.register_catalog("paimon", {"warehouse": str(path)})
+ ctx.sql("CREATE SCHEMA paimon.db")
+ ddl = "CREATE TABLE paimon.db.t (id INT, pt INT"
+ ddl += ", PRIMARY KEY (id))" if primary_key else ")"
+ if partitioned:
+ ddl += " PARTITIONED BY (pt)"
+ if options:
+ ddl += " WITH (" + ", ".join(f"'{k}' = '{v}'" for k, v in
options.items()) + ")"
+ ctx.sql(ddl)
+ return PaimonCatalog({"warehouse": str(path)}).get_table("db.t")
+
+
+def _write(writer, ids, partitions):
+ writer.write_arrow(pa.record_batch(
+ [ids, partitions], schema=pa.schema([("id", pa.int32()), ("pt",
pa.int32())])
+ ))
+
+
+def _prepare(builder, ids, partitions, identifier=None):
+ writer = builder.new_write()
+ _write(writer, ids, partitions)
+ return writer.prepare_commit() if identifier is None else
writer.prepare_commit(True, identifier)
+
+
+def _append(table, ids, partitions):
+ builder = table.new_batch_write_builder()
+ builder.new_commit().commit(_prepare(builder, ids, partitions))
+
+
+def _roundtrip(messages):
+ return [CommitMessage.deserialize(message.serialize()) for message in
messages]
+
+
+def _rows(table):
+ reader = table.new_read_builder()
+ batches = reader.new_read().read(reader.new_scan().plan().splits())
+ return sorted(row["id"] for batch in batches for row in batch.to_pylist())
+
+
+def _snapshot(table):
+ return json.loads((Path(table.location()) / "snapshot" /
+ f"snapshot-{table.latest_snapshot().id()}").read_text())
+
+
+def test_public_api_separates_batch_and_stream(tmp_path):
+ table = _table(tmp_path)
+ assert not hasattr(table, "new_commit")
+ assert not hasattr(table, "new_write_builder")
+ for name in ("WriteBuilder", "TableWrite", "TableCommit"):
+ assert not hasattr(datafusion, name)
+ for factory in (table.new_batch_write_builder,
table.new_stream_write_builder):
+ for kwargs in ({"commit_user": "job"}, {"overwrite": True}):
+ with pytest.raises(TypeError):
+ factory(**kwargs)
+ batch = table.new_batch_write_builder()
+ stream = table.new_stream_write_builder()
+ assert not hasattr(batch, "with_commit_user")
+ assert not hasattr(stream, "with_overwrite")
+ for obj in (batch, stream, batch.new_commit(), stream.new_commit()):
+ 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")
+ with pytest.raises(TypeError):
+ batch.new_commit().commit([], commit_identifier=1)
+ with pytest.raises(TypeError):
+ stream.new_commit().commit([])
+
+
+def test_serialized_stream_commit_and_grouped_retry(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_stream_write_builder()
+ assert builder.with_commit_user("python-job") is builder
+ assert builder.commit_user() == "python-job"
+ writer = builder.new_write()
+ commit = builder.new_commit()
+ _write(writer, [1], [10])
+ first = _roundtrip(writer.prepare_commit(True, 7))
+ commit.commit(7, first)
+ _write(writer, [2], [20])
+ second = _roundtrip(writer.prepare_commit(False, 8))
+ _write(writer, [3], [30])
+ third = _roundtrip(writer.prepare_commit(True, 9))
+ restored =
table.new_stream_write_builder().with_commit_user("python-job").new_commit()
+ # Input order differs from commit order; count groups after filtering.
+ assert restored.filter_and_commit({9: third, 7: first, 8: second}) == 2
+ assert _rows(table) == [1, 2, 3]
+ snapshot = _snapshot(table)
+ assert snapshot["commitUser"] == "python-job"
+ assert snapshot["commitIdentifier"] == 9
+ assert restored.filter_and_commit({7: first, 8: second, 9: third}) == 0
+ assert table.latest_snapshot().id() == snapshot["id"]
+ assert restored.filter_and_commit({}) == 0
+
+
[email protected]("user", ["", "../job", "a/b"])
+def test_invalid_stream_commit_user(tmp_path, user):
+ builder = _table(tmp_path).new_stream_write_builder()
+ original = builder.commit_user()
+ with pytest.raises(ValueError):
+ builder.with_commit_user(user)
+ assert builder.commit_user() == original
+
+
+def test_static_deserialize_checks_version_and_payload():
+ # Java CommitMessageSerializer v14 fixture; no table or builder is needed.
+ body = base64.b64decode(
+
"AAAADAAAAAAAAAAAAAAAAAAAAAMAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAABw=="
+ )
+ with pytest.raises(NotImplementedError, match="version"):
+ CommitMessage.deserialize(body, version=13)
+ for invalid in (b"", body[:-1], body + b"extra"):
+ with pytest.raises(ValueError):
+ CommitMessage.deserialize(invalid)
+ assert CommitMessage.deserialize(body).serialize() == body
+ assert CommitMessage.deserialize(body, version=14).serialize() == body
+
+
+def test_deserialized_batch_commit_uses_target_builder(tmp_path):
+ table = _table(tmp_path)
+ messages = _roundtrip(_prepare(table.new_batch_write_builder(), [1], [10]))
+ table.new_batch_write_builder().new_commit().commit(messages)
+ assert _rows(table) == [1]
+
+
+def test_abort_serialized_messages_deletes_files(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ messages = _roundtrip(_prepare(builder, [1], [10]))
+ files = list(tmp_path.rglob("data-*.parquet"))
+ assert files
+ builder.new_commit().abort(messages)
+ assert not any(file.exists() for file in files)
+ assert table.latest_snapshot() is None
+
+
+def test_batch_writer_and_commit_are_one_shot(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ writer = builder.new_write()
+ _write(writer, [1], [10])
+ messages = writer.prepare_commit()
+ with pytest.raises(RuntimeError, match="one-time"):
+ writer.prepare_commit()
+ commit = builder.new_commit()
+ commit.commit(messages)
+ with pytest.raises(RuntimeError, match="one-time"):
+ commit.commit(messages)
+ with pytest.raises(RuntimeError, match="one-time"):
+ commit.truncate_table()
+ assert _rows(table) == [1]
+ truncate = builder.new_commit()
+ truncate.truncate_table()
+ with pytest.raises(RuntimeError, match="one-time"):
+ truncate.commit([])
+ assert _rows(table) == []
+
+
[email protected]("ignore", ["true", "false", "False"])
+def test_batch_empty_commit_honors_option(tmp_path, ignore):
+ table = _table(tmp_path, options={"snapshot.ignore-empty-commit": ignore})
+ commit = table.new_batch_write_builder().new_commit()
+ commit.commit([])
+ assert (table.latest_snapshot() is None) == (ignore.lower() == "true")
+ with pytest.raises(RuntimeError, match="one-time"):
+ commit.commit([])
+
+
+def test_stream_empty_checkpoint_is_recorded(tmp_path):
+ table = _table(tmp_path, options={"snapshot.ignore-empty-commit": "true"})
+ commit =
table.new_stream_write_builder().with_commit_user("job").new_commit()
+ commit.commit(1, [])
+ assert _snapshot(table)["commitIdentifier"] == 1
+ assert commit.filter_and_commit({3: [], 1: [], 2: []}) == 2
+ assert _snapshot(table)["commitIdentifier"] == 3
+ assert table.latest_snapshot().id() == 3
+ assert _rows(table) == []
+
+
[email protected]("spec", [{}, {"pt": 999}])
+def test_default_dynamic_overwrite_uses_touched_partitions(tmp_path, spec):
+ table = _table(tmp_path, partitioned=True)
+ _append(table, [1, 2], [10, 20])
+ builder = table.new_batch_write_builder()
+ assert builder.with_overwrite(spec) is builder
+ builder.new_commit().commit(_roundtrip(_prepare(builder, [3], [10])))
+ assert _rows(table) == [2, 3]
+ table.new_batch_write_builder().with_overwrite().new_commit().commit([])
+ assert _rows(table) == [2, 3]
+
+
[email protected]("null_value", [None, "__DEFAULT_PARTITION__"])
+def test_static_overwrite_and_empty_truncation(tmp_path, null_value):
+ table = _table(tmp_path, partitioned=True,
options={"dynamic-partition-overwrite": "FALSE"})
+ _append(table, [1, 2, 3], [10, 20, None])
+ builder = table.new_batch_write_builder().with_overwrite({"pt": 10})
+ builder.new_commit().commit(_prepare(builder, [4], [10]))
+ assert _rows(table) == [2, 3, 4]
+ table.new_batch_write_builder().with_overwrite({"pt":
null_value}).new_commit().commit([])
+ assert _rows(table) == [2, 4]
+ table.new_batch_write_builder().with_overwrite().new_commit().commit([])
+ assert _rows(table) == []
+
+
+def test_unpartitioned_empty_overwrite_truncates(tmp_path):
+ table = _table(tmp_path)
+ _append(table, [1], [10])
+ table.new_batch_write_builder().with_overwrite().new_commit().commit([])
+ assert _rows(table) == []
+
+
+def test_explicit_none_disables_overwrite_and_context_is_copied(tmp_path):
+ table = _table(tmp_path)
+ _append(table, [1], [10])
+ builder = table.new_batch_write_builder().with_overwrite()
+ overwrite = builder.new_commit()
+ builder.with_overwrite(None)
+ builder.new_commit().commit(_prepare(builder, [2], [20]))
+ assert _rows(table) == [1, 2]
+ overwrite.commit([])
+ assert _rows(table) == []
+
+
+def test_static_deserialize_uses_committer_overwrite_mode(tmp_path):
+ table = _table(tmp_path)
+ _append(table, [1], [10])
+ builder = table.new_batch_write_builder().with_overwrite()
+ body = _prepare(builder, [2], [20])[0].serialize()
+ # Model an external message carrying Java totalBuckets.
+ flag_offset = 4 + int.from_bytes(body[:4], "big") + 4
+ assert body[flag_offset] == 0
+ body = body[:flag_offset] + bytes([1]) + (1).to_bytes(4, "big") +
body[flag_offset + 1:]
+ commit = builder.new_commit()
+ message = CommitMessage.deserialize(body)
+ assert message.serialize() == body
+ commit.commit([message])
+ assert _rows(table) == [2]
+
+
+def test_overwrite_does_not_mutate_deserialized_message(tmp_path):
+ table = _table(tmp_path, partitioned=True,
options={"dynamic-partition-overwrite": "false"})
+ _append(table, [1], [10])
+ body = _prepare(table.new_batch_write_builder(), [2], [20])[0].serialize()
+ flag_offset = 4 + int.from_bytes(body[:4], "big") + 4
+ assert body[flag_offset] == 0
+ body = body[:flag_offset] + bytes([1]) + (1).to_bytes(4, "big") +
body[flag_offset + 1:]
+ message = CommitMessage.deserialize(body)
+ overwrite = table.new_batch_write_builder().with_overwrite({"pt":
10}).new_commit()
+ with pytest.raises(ValueError, match="does not belong"):
+ overwrite.commit([message])
+ # A failed overwrite must not stamp the object with its operation mode.
+ table.new_batch_write_builder().new_commit().commit([message])
+ assert _rows(table) == [1, 2]
+
+
+def test_failed_batch_commit_consumes_instance(tmp_path):
+ table = _table(tmp_path, partitioned=True,
options={"dynamic-partition-overwrite": "FALSE"})
+ _append(table, [1], [10])
+ builder = table.new_batch_write_builder().with_overwrite({"pt": 20})
+ messages = _prepare(builder, [2], [10])
+ commit = builder.new_commit()
+ with pytest.raises(ValueError, match="does not belong"):
+ commit.commit(messages)
+ with pytest.raises(RuntimeError, match="one-time"):
+ commit.commit(messages)
+ assert _rows(table) == [1]
+ assert len(list(tmp_path.rglob("data-*.parquet"))) == 2
+
+
+def test_partition_truncation_matches_java_lifecycle(tmp_path):
+ table = _table(tmp_path, partitioned=True)
+ _append(table, [1, 2], [10, 20])
+ commit = table.new_batch_write_builder().new_commit()
+ with pytest.raises(ValueError, match="Partitions list cannot be empty"):
+ commit.truncate_partitions([])
+ commit.truncate_partitions([{"pt": 10}])
+ assert _rows(table) == [2]
+ # Java truncatePartitions does not consume the batch commit guard.
+ commit.commit([])
+ commit.truncate_partitions([{"pt": 20}])
+ assert _rows(table) == []
+
+
[email protected]("spec", [{"id": 1}, {"pt": "bad"}, {"pt": True}])
+def test_partition_spec_validation_precedes_mutation(tmp_path, spec):
+ table = _table(tmp_path, partitioned=True)
+ _append(table, [1], [10])
+ snapshot_id = table.latest_snapshot().id()
+ with pytest.raises(ValueError):
+ table.new_batch_write_builder().with_overwrite(spec)
+ with pytest.raises(ValueError):
+
table.new_batch_write_builder().new_commit().truncate_partitions([spec])
+ assert table.latest_snapshot().id() == snapshot_id
+ assert _rows(table) == [1]
+
+
+def test_recovery_checks_all_pending_files_before_any_commit(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_stream_write_builder().with_commit_user("job")
+ first = _prepare(builder, [1], [10], 1)
+ before = set(tmp_path.rglob("data-*.parquet"))
+ second = _prepare(builder, [2], [20], 2)
+ for path in set(tmp_path.rglob("data-*.parquet")) - before:
+ path.unlink()
+ with pytest.raises(ValueError, match="does not exist"):
+ builder.new_commit().filter_and_commit({1: first, 2: second})
+ assert table.latest_snapshot() is None
+
+
+def test_recovery_skips_file_check_for_committed_checkpoints(tmp_path):
+ table = _table(tmp_path)
+ builder = table.new_stream_write_builder().with_commit_user("job")
+ messages = _prepare(builder, [1], [10], 1)
+ commit = builder.new_commit()
+ commit.commit(1, messages)
+ # Model retention after a later overwrite: a filtered checkpoint can
reference expired files.
+ for path in tmp_path.rglob("data-*.parquet"):
+ path.unlink()
+ assert commit.filter_and_commit({1: messages}) == 0
+ assert table.latest_snapshot().id() == 1
+
+
[email protected]("operation", ["overwrite", "truncate"])
+def test_empty_destructive_batch_operation_records_snapshot(tmp_path,
operation):
+ table = _table(tmp_path)
+ builder = table.new_batch_write_builder()
+ if operation == "overwrite":
+ builder.with_overwrite().new_commit().commit([])
+ else:
+ builder.new_commit().truncate_table()
+ assert _snapshot(table)["commitKind"] == "OVERWRITE"
+ assert _rows(table) == []
+
+
[email protected]("mode", ["batch", "stream"])
+def test_close_writer_preserves_prepared_files(tmp_path, mode):
+ table = _table(tmp_path)
+ builder = getattr(table, f"new_{mode}_write_builder")()
+ writer = builder.new_write()
+ _write(writer, [1], [10])
+ messages = writer.prepare_commit() if mode == "batch" else
writer.prepare_commit(True, 1)
+ writer.close()
+ writer.close()
+ with pytest.raises(RuntimeError, match="closed"):
+ _write(writer, [2], [20])
+ commit = builder.new_commit()
+ if mode == "batch":
+ commit.commit(messages)
+ else:
+ commit.commit(1, messages)
+ commit.close()
+ assert _rows(table) == [1]
+
+
+def test_static_overwrite_accepts_java_numeric_partition_strings(tmp_path):
+ table = _table(tmp_path, partitioned=True,
options={"dynamic-partition-overwrite": "false"})
+ _append(table, [1, 2], [10, 20])
+ table.new_batch_write_builder().with_overwrite({"pt":
"10"}).new_commit().commit([])
+ assert _rows(table) == [2]
+
+
[email protected]("bucket", [None, "1", "-2"])
+def test_close_cleans_unprepared_output_but_preserves_prepared_files(tmp_path,
bucket):
+ options = {"target-file-size": "1 b", "write.parquet-buffer-size": "1 b"}
+ if bucket is not None:
+ options["bucket"] = bucket
+ table = _table(tmp_path, options=options, primary_key=bucket is not None)
+ builder = table.new_stream_write_builder()
+ writer = builder.new_write()
+ _write(writer, [1], [10])
+ messages = writer.prepare_commit(True, 1)
+ prepared_paths = set(tmp_path.rglob("*.parquet"))
+ assert prepared_paths
+ _write(writer, [2], [20])
+ _write(writer, [3], [30])
+ assert set(tmp_path.rglob("*.parquet")) - prepared_paths
+ writer.close()
+ assert set(tmp_path.rglob("*.parquet")) == prepared_paths
+ builder.new_commit().commit(1, messages)
+ assert _snapshot(table)["totalRecordCount"] == 1
+ assert [row["id"] for path in prepared_paths
+ for row in pq.ParquetFile(path).read(columns=["id"]).to_pylist()]
== [1]
+ if bucket != "-2":
+ assert _rows(table) == [1]
diff --git a/bindings/python/tests/test_write.py
b/bindings/python/tests/test_write.py
index 98d9274d..f36fbfdd 100644
--- a/bindings/python/tests/test_write.py
+++ b/bindings/python/tests/test_write.py
@@ -52,7 +52,7 @@ def test_write_commit_read_roundtrip():
ctx = _make_empty_table(warehouse)
table = _get_table(warehouse)
batch = _batch([1, 2, 3], ["a", "b", "c"])
- wb = table.new_write_builder()
+ wb = table.new_batch_write_builder()
write = wb.new_write()
write.write_arrow(batch)
messages = write.prepare_commit()
@@ -68,7 +68,7 @@ def test_write_multiple_batches():
with tempfile.TemporaryDirectory() as warehouse:
ctx = _make_empty_table(warehouse)
table = _get_table(warehouse)
- wb = table.new_write_builder()
+ wb = table.new_batch_write_builder()
write = wb.new_write()
write.write_arrow(_batch([1], ["a"]))
write.write_arrow(_batch([2], ["b"]))
@@ -84,7 +84,7 @@ def test_prepare_commit_returns_messages():
with tempfile.TemporaryDirectory() as warehouse:
_make_empty_table(warehouse)
table = _get_table(warehouse)
- write = table.new_write_builder().new_write()
+ write = table.new_batch_write_builder().new_write()
write.write_arrow(_batch([1], ["a"]))
messages = write.prepare_commit()
assert len(messages) >= 1
@@ -95,7 +95,7 @@ def test_commit_empty_messages_noop():
with tempfile.TemporaryDirectory() as warehouse:
ctx = _make_empty_table(warehouse)
table = _get_table(warehouse)
- wb = table.new_write_builder()
+ wb = table.new_batch_write_builder()
messages = wb.new_write().prepare_commit() # no write
assert messages == []
wb.new_commit().commit(messages) # no-op success
@@ -107,7 +107,7 @@ def test_write_arrow_type_mismatch_raises():
with tempfile.TemporaryDirectory() as warehouse:
_make_empty_table(warehouse) # table (id INT, name STRING)
table = _get_table(warehouse)
- write = table.new_write_builder().new_write()
+ write = table.new_batch_write_builder().new_write()
bad = pa.record_batch([["x", "y"], ["a", "b"]], names=["id", "name"])
# id as STRING
with pytest.raises(ValueError):
write.write_arrow(bad)
@@ -123,7 +123,7 @@ def test_write_arrow_binary_family_mismatch_raises():
ctx.sql("CREATE SCHEMA paimon.wdb")
ctx.sql("CREATE TABLE paimon.wdb.bt (id INT, data BINARY)")
table = PaimonCatalog({"warehouse": warehouse}).get_table("wdb.bt")
- write = table.new_write_builder().new_write()
+ write = table.new_batch_write_builder().new_write()
schema = pa.schema([("id", pa.int32()), ("data", pa.large_binary())])
bad = pa.record_batch([[1], [b"x"]], schema=schema)
with pytest.raises(ValueError):
@@ -135,10 +135,10 @@ def test_commit_non_message_raises_typeerror():
_make_empty_table(warehouse)
table = _get_table(warehouse)
with pytest.raises(TypeError):
- table.new_write_builder().new_commit().commit([object()])
+ table.new_batch_write_builder().new_commit().commit([object()])
# A non-iterable argument also raises TypeError (not a raw PyO3 error).
with pytest.raises(TypeError):
- table.new_write_builder().new_commit().commit(42)
+ table.new_batch_write_builder().new_commit().commit(42)
def test_commit_cross_table_messages_raises():
@@ -159,26 +159,26 @@ def test_commit_cross_table_messages_raises():
[pa.array([1], pa.int32()), pa.array(["a"], pa.string())],
names=["id", "name"],
)
- w1 = t1.new_write_builder().new_write()
+ w1 = t1.new_batch_write_builder().new_write()
w1.write_arrow(batch)
messages = w1.prepare_commit()
with pytest.raises(ValueError):
- t2.new_write_builder().new_commit().commit(messages)
+ t2.new_batch_write_builder().new_commit().commit(messages)
def test_commit_different_builder_same_table_raises():
- # Even for the same table, a committer from a different WriteBuilder must
+ # Even for the same table, a committer from a different BatchWriteBuilder
must
# reject the messages: each builder mints its own commit_user, and writers
# and committers must share one (snapshot duplicate detection / postpone
# bucket file naming depend on it).
with tempfile.TemporaryDirectory() as warehouse:
_make_empty_table(warehouse)
table = _get_table(warehouse)
- write = table.new_write_builder().new_write()
+ write = table.new_batch_write_builder().new_write()
write.write_arrow(_batch([1], ["a"]))
messages = write.prepare_commit()
with pytest.raises(ValueError):
- table.new_write_builder().new_commit().commit(messages)
+ table.new_batch_write_builder().new_commit().commit(messages)
def test_abort_cleans_up_written_data():
@@ -187,7 +187,7 @@ def test_abort_cleans_up_written_data():
with tempfile.TemporaryDirectory() as warehouse:
ctx = _make_empty_table(warehouse)
table = _get_table(warehouse)
- wb = table.new_write_builder()
+ wb = table.new_batch_write_builder()
write = wb.new_write()
write.write_arrow(_batch([1, 2, 3], ["a", "b", "c"]))
messages = write.prepare_commit()
@@ -202,7 +202,7 @@ def test_abort_empty_messages_noop():
with tempfile.TemporaryDirectory() as warehouse:
ctx = _make_empty_table(warehouse)
table = _get_table(warehouse)
- wb = table.new_write_builder()
+ wb = table.new_batch_write_builder()
messages = wb.new_write().prepare_commit() # no write
assert messages == []
wb.new_commit().abort(messages) # no-op success
@@ -215,9 +215,9 @@ def test_abort_non_message_raises_typeerror():
_make_empty_table(warehouse)
table = _get_table(warehouse)
with pytest.raises(TypeError):
- table.new_write_builder().new_commit().abort([object()])
+ table.new_batch_write_builder().new_commit().abort([object()])
with pytest.raises(TypeError):
- table.new_write_builder().new_commit().abort(42)
+ table.new_batch_write_builder().new_commit().abort(42)
def test_abort_cross_table_messages_raises():
@@ -234,22 +234,22 @@ def test_abort_cross_table_messages_raises():
[pa.array([1], pa.int32()), pa.array(["a"], pa.string())],
names=["id", "name"],
)
- w1 = t1.new_write_builder().new_write()
+ w1 = t1.new_batch_write_builder().new_write()
w1.write_arrow(batch)
messages = w1.prepare_commit()
with pytest.raises(ValueError):
- t2.new_write_builder().new_commit().abort(messages)
+ t2.new_batch_write_builder().new_commit().abort(messages)
def test_abort_different_builder_same_table_raises():
with tempfile.TemporaryDirectory() as warehouse:
_make_empty_table(warehouse)
table = _get_table(warehouse)
- write = table.new_write_builder().new_write()
+ write = table.new_batch_write_builder().new_write()
write.write_arrow(_batch([1], ["a"]))
messages = write.prepare_commit()
with pytest.raises(ValueError):
- table.new_write_builder().new_commit().abort(messages)
+ table.new_batch_write_builder().new_commit().abort(messages)
@pytest.mark.parametrize("legacy,precision,unit,value,directory", [
@@ -270,7 +270,7 @@ def
test_timestamp_partition_writes_and_reads_use_java_paths(tmp_path, legacy, p
schema = pa.schema([("id", pa.int32()), ("ts", pa.timestamp(unit))])
for row_id in [1, 2]:
table = _get_table(str(tmp_path))
- builder = table.new_write_builder()
+ builder = table.new_batch_write_builder()
write = builder.new_write()
write.write_arrow(pa.record_batch([[row_id], [value]], schema=schema))
builder.new_commit().commit(write.prepare_commit())
@@ -304,7 +304,7 @@ def test_ltz_schema_alias_write_roundtrip(tmp_path,
precision, unit, micros, fra
)
value = datetime(2026, 9, 15, 20, 0, 0, micros, tzinfo=timezone.utc)
arrow_schema = pa.schema([("id", pa.int32()), ("ts", pa.timestamp(unit,
tz="UTC"))])
- builder = table.new_write_builder()
+ builder = table.new_batch_write_builder()
write = builder.new_write()
write.write_arrow(pa.record_batch([[1], [value]], schema=arrow_schema))
builder.new_commit().commit(write.prepare_commit())
diff --git a/crates/paimon/src/spec/index_file_meta.rs
b/crates/paimon/src/spec/index_file_meta.rs
index 7af98412..278f3fd6 100644
--- a/crates/paimon/src/spec/index_file_meta.rs
+++ b/crates/paimon/src/spec/index_file_meta.rs
@@ -18,8 +18,59 @@
use serde::{Deserialize, Serialize};
use std::fmt::{Display, Formatter};
+use crate::spec::{
+ deserialize_binary_array_int, deserialize_binary_array_rows, BinaryRow,
BinaryRowBuilder,
+};
use indexmap::IndexMap;
+fn invalid(message: impl Into<String>) -> crate::Error {
+ crate::Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+fn round_to_word(size: usize) -> usize {
+ (size + 7) & !7
+}
+
+fn array_header(size: usize) -> usize {
+ 4 + size.div_ceil(32) * 4
+}
+
+fn serialize_int_array(values: &[i32]) -> Vec<u8> {
+ let header = array_header(values.len());
+ let mut data = vec![0; round_to_word(header + values.len() * 4)];
+ data[..4].copy_from_slice(&(values.len() as i32).to_le_bytes());
+ for (i, value) in values.iter().enumerate() {
+ data[header + i * 4..header + i * 4 +
4].copy_from_slice(&value.to_le_bytes());
+ }
+ data
+}
+
+fn serialize_row_array(rows: &[Vec<u8>]) -> Vec<u8> {
+ let header = array_header(rows.len());
+ let mut data = vec![0; round_to_word(header + rows.len() * 8)];
+ data[..4].copy_from_slice(&(rows.len() as i32).to_le_bytes());
+ for (i, row) in rows.iter().enumerate() {
+ let offset = data.len();
+ data.extend_from_slice(row);
+ data.resize(data.len() + round_to_word(row.len()) - row.len(), 0);
+ let slot = ((offset as u64) << 32) | row.len() as u64;
+ data[header + i * 8..header + i * 8 +
8].copy_from_slice(&slot.to_le_bytes());
+ }
+ data
+}
+
+fn row_from_data(data: &[u8], arity: i32) -> crate::Result<BinaryRow> {
+ if data.len() < BinaryRow::cal_fix_part_size_in_bytes(arity) as usize {
+ return Err(invalid(format!(
+ "IndexFileMeta nested row is too short for arity {arity}"
+ )));
+ }
+ Ok(BinaryRow::from_bytes(arity, data.to_vec()))
+}
+
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DeletionVectorMeta {
pub offset: i32,
@@ -94,6 +145,133 @@ pub struct IndexFileMeta {
pub global_index_meta: Option<GlobalIndexMeta>,
}
+impl IndexFileMeta {
+ /// Java `IndexFileMetaSerializer` row body, without its i32 length prefix.
+ pub fn to_serialized_row_data(&self) -> crate::Result<Vec<u8>> {
+ let mut row = BinaryRowBuilder::new(7);
+ row.write_bytes(0, self.index_type.as_bytes());
+ row.write_bytes(1, self.file_name.as_bytes());
+ row.write_long(2, self.file_size);
+ row.write_long(3, self.row_count);
+ if let Some(ranges) = &self.deletion_vectors_ranges {
+ let rows = ranges
+ .iter()
+ .map(|(name, range)| {
+ let mut dv = BinaryRowBuilder::new(4);
+ dv.write_bytes(0, name.as_bytes());
+ dv.write_int(1, range.offset);
+ dv.write_int(2, range.length);
+ match range.cardinality {
+ Some(value) => dv.write_long(3, value),
+ None => dv.set_null_at(3),
+ }
+ dv.build_row_data()
+ })
+ .collect::<Vec<_>>();
+ row.write_bytes(4, &serialize_row_array(&rows));
+ } else {
+ row.set_null_at(4);
+ }
+ match &self.external_path {
+ Some(value) => row.write_bytes(5, value.as_bytes()),
+ None => row.set_null_at(5),
+ }
+ if let Some(meta) = &self.global_index_meta {
+ let mut global = BinaryRowBuilder::new(6);
+ global.write_long(0, meta.row_range_start);
+ global.write_long(1, meta.row_range_end);
+ global.write_int(2, meta.index_field_id);
+ match &meta.extra_field_ids {
+ Some(values) => global.write_bytes(3,
&serialize_int_array(values)),
+ None => global.set_null_at(3),
+ }
+ match &meta.index_meta {
+ Some(value) => global.write_bytes(4, value),
+ None => global.set_null_at(4),
+ }
+ match &meta.source_meta {
+ Some(value) => global.write_bytes(5, value),
+ None => global.set_null_at(5),
+ }
+ row.write_bytes(6, &global.build_row_data());
+ } else {
+ row.set_null_at(6);
+ }
+ Ok(row.build_row_data())
+ }
+
+ /// Reverse of `to_serialized_row_data` for the current Java row schema.
+ pub fn from_serialized_row_data(data: &[u8]) -> crate::Result<Self> {
+ let row = row_from_data(data, 7)?;
+ let name = |pos| -> crate::Result<String> {
+ String::from_utf8(row.get_binary(pos)?.to_vec())
+ .map_err(|_| invalid(format!("IndexFileMeta field {pos} is not
UTF-8")))
+ };
+ let ranges = if row.is_null_at(4) {
+ None
+ } else {
+ let mut ranges = IndexMap::new();
+ for raw in deserialize_binary_array_rows(row.get_binary(4)?)? {
+ let dv = row_from_data(raw, 4)?;
+ let name = String::from_utf8(dv.get_binary(0)?.to_vec())
+ .map_err(|_| invalid("deletion vector file name is not
UTF-8"))?;
+ ranges.insert(
+ name,
+ DeletionVectorMeta {
+ offset: dv.get_int(1)?,
+ length: dv.get_int(2)?,
+ cardinality: if dv.is_null_at(3) {
+ None
+ } else {
+ Some(dv.get_long(3)?)
+ },
+ },
+ );
+ }
+ Some(ranges)
+ };
+ let external_path = if row.is_null_at(5) {
+ None
+ } else {
+ Some(name(5)?)
+ };
+ let global_index_meta = if row.is_null_at(6) {
+ None
+ } else {
+ let global = row_from_data(row.get_binary(6)?, 6)?;
+ Some(GlobalIndexMeta {
+ row_range_start: global.get_long(0)?,
+ row_range_end: global.get_long(1)?,
+ index_field_id: global.get_int(2)?,
+ extra_field_ids: if global.is_null_at(3) {
+ None
+ } else {
+ Some(deserialize_binary_array_int(global.get_binary(3)?)?)
+ },
+ index_meta: if global.is_null_at(4) {
+ None
+ } else {
+ Some(global.get_binary(4)?.to_vec())
+ },
+ source_meta: if global.is_null_at(5) {
+ None
+ } else {
+ Some(global.get_binary(5)?.to_vec())
+ },
+ })
+ };
+ Ok(Self {
+ index_type: name(0)?,
+ file_name: name(1)?,
+ file_size: row.get_long(2)?,
+ row_count: row.get_long(3)?,
+ deletion_vectors_ranges: ranges,
+ external_path,
+ global_index_meta,
+ })
+ }
+}
+
impl Display for IndexFileMeta {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(
@@ -168,3 +346,42 @@ mod map_serde {
}
}
}
+
+#[cfg(test)]
+mod wire_tests {
+ use super::*;
+
+ #[test]
+ fn current_java_row_round_trip_with_nested_metadata() {
+ let mut ranges = IndexMap::new();
+ ranges.insert(
+ "data.parquet".into(),
+ DeletionVectorMeta {
+ offset: 3,
+ length: 12,
+ cardinality: Some(2),
+ },
+ );
+ let original = IndexFileMeta {
+ index_type: "DV".into(),
+ file_name: "index".into(),
+ file_size: 9,
+ row_count: 2,
+ deletion_vectors_ranges: Some(ranges),
+ external_path: Some("file:/index".into()),
+ global_index_meta: Some(GlobalIndexMeta {
+ row_range_start: 10,
+ row_range_end: 20,
+ index_field_id: 3,
+ extra_field_ids: Some(vec![1, 4]),
+ index_meta: Some(vec![1, 2]),
+ source_meta: Some(vec![3]),
+ }),
+ };
+ let bytes = original.to_serialized_row_data().unwrap();
+ assert_eq!(
+ IndexFileMeta::from_serialized_row_data(&bytes).unwrap(),
+ original
+ );
+ }
+}
diff --git a/crates/paimon/src/table/commit_message.rs
b/crates/paimon/src/table/commit_message.rs
index 55afbf64..f4e86ba0 100644
--- a/crates/paimon/src/table/commit_message.rs
+++ b/crates/paimon/src/table/commit_message.rs
@@ -15,8 +15,56 @@
// specific language governing permissions and limitations
// under the License.
-use crate::spec::DataFileMeta;
-use crate::spec::IndexFileMeta;
+use super::source::{read_i32, read_i64, take};
+use crate::spec::{BinaryRow, DataFileMeta, DataFileMetaRowLayout,
IndexFileMeta};
+
+/// Current Java `CommitMessageSerializer` body version. The version is carried
+/// by an enclosing serializer, not embedded in the body.
+pub const COMMIT_MESSAGE_SERIALIZER_VERSION: i32 = 14;
+
+fn invalid(message: impl Into<String>) -> crate::Error {
+ crate::Error::DataInvalid {
+ message: message.into(),
+ source: None,
+ }
+}
+
+fn write_rows<T>(
+ out: &mut Vec<u8>,
+ values: &[T],
+ encode: impl Fn(&T) -> crate::Result<Vec<u8>>,
+) -> crate::Result<()> {
+ out.extend_from_slice(&(values.len() as i32).to_be_bytes());
+ for value in values {
+ let row = encode(value)?;
+ out.extend_from_slice(&(row.len() as i32).to_be_bytes());
+ out.extend_from_slice(&row);
+ }
+ Ok(())
+}
+
+fn read_rows<T>(
+ cur: &mut &[u8],
+ decode: impl Fn(&[u8]) -> crate::Result<T>,
+) -> crate::Result<Vec<T>> {
+ let count = read_i32(cur)?;
+ if count < 0 || count as usize > cur.len() / 4 {
+ return Err(invalid(format!(
+ "invalid CommitMessage list count: {count}"
+ )));
+ }
+ let mut values = Vec::new();
+ for _ in 0..count {
+ let length = read_i32(cur)?;
+ if length < 0 {
+ return Err(invalid(format!(
+ "negative CommitMessage row length: {length}"
+ )));
+ }
+ values.push(decode(take(cur, length as usize)?)?);
+ }
+ Ok(values)
+}
/// A commit message representing new files to be committed for a specific
partition and bucket.
///
@@ -41,6 +89,13 @@ pub struct CommitMessage {
pub deleted_index_files: Vec<IndexFileMeta>,
/// Files to be deleted (copy-on-write rewrite: old files replaced by
new_files).
pub deleted_files: Vec<DataFileMeta>,
+ /// Files removed by compaction (Java's separate compact increment).
+ pub compact_before: Vec<DataFileMeta>,
+ /// Files produced by compaction.
+ pub compact_after: Vec<DataFileMeta>,
+ pub compact_changelog_files: Vec<DataFileMeta>,
+ pub compact_new_index_files: Vec<IndexFileMeta>,
+ pub compact_deleted_index_files: Vec<IndexFileMeta>,
fixed_bucket_overwrite: bool,
}
@@ -56,15 +111,212 @@ impl CommitMessage {
new_index_files: Vec::new(),
deleted_index_files: Vec::new(),
deleted_files: Vec::new(),
+ compact_before: Vec::new(),
+ compact_after: Vec::new(),
+ compact_changelog_files: Vec::new(),
+ compact_new_index_files: Vec::new(),
+ compact_deleted_index_files: Vec::new(),
fixed_bucket_overwrite: false,
}
}
- pub(crate) fn mark_fixed_bucket_overwrite(&mut self) {
+ /// Supply the overwrite operation when restoring a message from Java's
wire
+ /// format, which does not encode this flag. Use the target committer's
mode.
+ pub fn mark_fixed_bucket_overwrite(&mut self) {
self.fixed_bucket_overwrite = true;
}
pub(crate) fn is_fixed_bucket_overwrite(&self) -> bool {
self.fixed_bucket_overwrite
}
+
+ /// Write the unframed Java v14 `CommitMessageSerializer.serialize` body.
+ pub fn serialize(&self) -> crate::Result<Vec<u8>> {
+ let mut out = Vec::new();
+ // The partition normally is SerializationUtils.serializeBinaryRow:
+ // i32 arity followed by a raw BinaryRow. Internal unpartitioned
writers
+ // may still use an empty Vec, which represents BinaryRow.EMPTY_ROW.
+ let partition = if self.partition.is_empty() {
+ BinaryRow::new(0).to_serialized_bytes()
+ } else {
+ BinaryRow::from_serialized_bytes(&self.partition)?;
+ self.partition.clone()
+ };
+ out.extend_from_slice(&(partition.len() as i32).to_be_bytes());
+ out.extend_from_slice(&partition);
+ out.extend_from_slice(&self.bucket.to_be_bytes());
+ out.push(u8::from(self.total_buckets.is_some()));
+ if let Some(value) = self.total_buckets {
+ out.extend_from_slice(&value.to_be_bytes());
+ }
+ for files in [
+ &self.new_files,
+ &self.deleted_files,
+ &self.new_changelog_files,
+ ] {
+ write_rows(&mut out, files, DataFileMeta::to_serialized_row_data)?;
+ }
+ for files in [&self.new_index_files, &self.deleted_index_files] {
+ write_rows(&mut out, files,
IndexFileMeta::to_serialized_row_data)?;
+ }
+ for files in [
+ &self.compact_before,
+ &self.compact_after,
+ &self.compact_changelog_files,
+ ] {
+ write_rows(&mut out, files, DataFileMeta::to_serialized_row_data)?;
+ }
+ for files in [
+ &self.compact_new_index_files,
+ &self.compact_deleted_index_files,
+ ] {
+ write_rows(&mut out, files,
IndexFileMeta::to_serialized_row_data)?;
+ }
+ out.push(u8::from(self.check_from_snapshot.is_some()));
+ if let Some(value) = self.check_from_snapshot {
+ out.extend_from_slice(&value.to_be_bytes());
+ }
+ Ok(out)
+ }
+
+ /// Decode an unframed Java `CommitMessageSerializer` v14 body.
+ pub fn deserialize(version: i32, bytes: &[u8]) -> crate::Result<Self> {
+ if version != COMMIT_MESSAGE_SERIALIZER_VERSION {
+ return Err(crate::Error::Unsupported {
+ message: format!("CommitMessage serializer version {version}
is not supported"),
+ });
+ }
+ let mut cur = bytes;
+ let partition_len = read_i32(&mut cur)?;
+ if partition_len < 0 {
+ return Err(invalid("negative CommitMessage partition length"));
+ }
+ let partition = take(&mut cur, partition_len as usize)?.to_vec();
+ BinaryRow::from_serialized_bytes(&partition)?;
+ let bucket = read_i32(&mut cur)?;
+ let total_buckets = match take(&mut cur, 1)?[0] {
+ 0 => None,
+ 1 => Some(read_i32(&mut cur)?),
+ value => return Err(invalid(format!("invalid totalBuckets flag:
{value}"))),
+ };
+ let data_file = |raw: &[u8]| {
+ DataFileMeta::from_serialized_row_data(raw,
DataFileMetaRowLayout::CURRENT)
+ };
+ let new_files = read_rows(&mut cur, data_file)?;
+ let deleted_files = read_rows(&mut cur, data_file)?;
+ let new_changelog_files = read_rows(&mut cur, data_file)?;
+ let new_index_files = read_rows(&mut cur,
IndexFileMeta::from_serialized_row_data)?;
+ let deleted_index_files = read_rows(&mut cur,
IndexFileMeta::from_serialized_row_data)?;
+ let compact_before = read_rows(&mut cur, data_file)?;
+ let compact_after = read_rows(&mut cur, data_file)?;
+ let compact_changelog_files = read_rows(&mut cur, data_file)?;
+ let compact_new_index_files = read_rows(&mut cur,
IndexFileMeta::from_serialized_row_data)?;
+ let compact_deleted_index_files =
+ read_rows(&mut cur, IndexFileMeta::from_serialized_row_data)?;
+ let check_from_snapshot = match take(&mut cur, 1)?[0] {
+ 0 => None,
+ 1 => Some(read_i64(&mut cur)?),
+ value => return Err(invalid(format!("invalid checkFromSnapshot
flag: {value}"))),
+ };
+ if !cur.is_empty() {
+ return Err(invalid(format!(
+ "{} trailing bytes after CommitMessage",
+ cur.len()
+ )));
+ }
+ Ok(Self {
+ partition,
+ bucket,
+ total_buckets,
+ new_files,
+ deleted_files,
+ check_from_snapshot,
+ new_changelog_files,
+ new_index_files,
+ deleted_index_files,
+ compact_before,
+ compact_after,
+ compact_changelog_files,
+ compact_new_index_files,
+ compact_deleted_index_files,
+ fixed_bucket_overwrite: false,
+ })
+ }
+
+ /// Import a message for the fixed-bucket overwrite commit path. Java's
+ /// wire format does not encode the commit operation, so the caller must
+ /// supply that context when restoring a prepared overwrite message.
+ pub fn deserialize_for_fixed_bucket_overwrite(
+ version: i32,
+ bytes: &[u8],
+ ) -> crate::Result<Self> {
+ let mut message = Self::deserialize(version, bytes)?;
+ message.mark_fixed_bucket_overwrite();
+ Ok(message)
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use base64::Engine;
+
+ // Produced by Java CommitMessageSerializer v14 with BinaryRow.EMPTY_ROW,
+ // bucket 3, checkFromSnapshot 7. The second message also has totalBuckets
+ // 5 and one data-increment IndexFileMeta("I", "index", 9, 2).
+ const EMPTY: &str =
"AAAADAAAAAAAAAAAAAAAAAAAAAMAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAABw==";
+ const INDEXED: &str =
"AAAADAAAAAAAAAAAAAAAAAAAAAMBAAAABQAAAAAAAAAAAAAAAAAAAAEAAABAAHAAAAAAAABJAAAAAAAAgWluZGV4AACFCQAAAAAAAAACAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAEAAAAAAAAABw==";
+ const RICH_INDEXED: &str =
"AAAADAAAAAAAAAAAAAAAAAAAAAMBAAAABQAAAAAAAAAAAAAAAAAAAAEAAADgAAAAAAAAAABEVgAAAAAAgmluZGV4AACFCQAAAAAAAAACAAAAAAAAAEgAAABAAAAACwAAAIgAAABIAAAAmAAAAAEAAAAAAAAAOAAAABAAAAAAAAAAAAAAAAwAAAAoAAAAAwAAAAAAAAAMAAAAAAAAAAIAAAAAAAAAZGF0YS5wYXJxdWV0AAAAAGZpbGU6L2luZGV4AAAAAAAAAAAAAAAAAAoAAAAAAAAAFAAAAAAAAAADAAAAAAAAABAAAAA4AAAAAQIAAAAAAIIDAAAAAAAAgQIAAAAAAAAAAQAAAAQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAABAAAAAAAAAAc=";
+
+ #[test]
+ fn java_v14_golden_round_trip() {
+ for golden in [EMPTY, INDEXED, RICH_INDEXED] {
+ let bytes = base64::engine::general_purpose::STANDARD
+ .decode(golden)
+ .unwrap();
+ let message =
+ CommitMessage::deserialize(COMMIT_MESSAGE_SERIALIZER_VERSION,
&bytes).unwrap();
+ assert_eq!(message.bucket, 3);
+ assert_eq!(message.check_from_snapshot, Some(7));
+ assert_eq!(message.serialize().unwrap(), bytes);
+ }
+ let indexed = base64::engine::general_purpose::STANDARD
+ .decode(INDEXED)
+ .unwrap();
+ let message = CommitMessage::deserialize(14, &indexed).unwrap();
+ assert_eq!(message.total_buckets, Some(5));
+ assert_eq!(message.new_index_files[0].file_name, "index");
+ let overwrite =
+ CommitMessage::deserialize_for_fixed_bucket_overwrite(14,
&indexed).unwrap();
+ assert!(overwrite.is_fixed_bucket_overwrite());
+ let mut empty_partition = CommitMessage::new(Vec::new(), 3,
Vec::new());
+ empty_partition.check_from_snapshot = Some(7);
+ let empty = base64::engine::general_purpose::STANDARD
+ .decode(EMPTY)
+ .unwrap();
+ assert_eq!(empty_partition.serialize().unwrap(), empty);
+ let rich = base64::engine::general_purpose::STANDARD
+ .decode(RICH_INDEXED)
+ .unwrap();
+ let message = CommitMessage::deserialize(14, &rich).unwrap();
+ let index = &message.new_index_files[0];
+ assert_eq!(index.external_path.as_deref(), Some("file:/index"));
+ assert_eq!(index.deletion_vectors_ranges.as_ref().unwrap().len(), 1);
+ assert_eq!(
+ index.global_index_meta.as_ref().unwrap().extra_field_ids,
+ Some(vec![1, 4])
+ );
+ }
+
+ #[test]
+ fn rejects_truncated_or_unsupported_message() {
+ let bytes = base64::engine::general_purpose::STANDARD
+ .decode(EMPTY)
+ .unwrap();
+ assert!(CommitMessage::deserialize(13, &bytes).is_err());
+ assert!(CommitMessage::deserialize(14, &bytes[..bytes.len() -
1]).is_err());
+ let mut trailing = bytes;
+ trailing.push(0);
+ assert!(CommitMessage::deserialize(14, &trailing).is_err());
+ }
}
diff --git a/crates/paimon/src/table/data_file_writer.rs
b/crates/paimon/src/table/data_file_writer.rs
index e68ebc9a..2ace11e2 100644
--- a/crates/paimon/src/table/data_file_writer.rs
+++ b/crates/paimon/src/table/data_file_writer.rs
@@ -67,7 +67,7 @@ pub(crate) struct DataFileWriter {
current_row_count: i64,
index_options: Option<Arc<FileIndexOptions>>,
current_index: Option<DataFileIndexWriter>,
- /// Paths owned by this indexed write until prepare_commit hands them to
the caller.
+ /// Paths owned by this write until prepare_commit hands them to the
caller.
created_paths: Vec<String>,
}
@@ -177,8 +177,8 @@ impl DataFileWriter {
self.file_io.mkdirs(&format!("{bucket_dir}/")).await?;
let file_path = format!("{bucket_dir}/{file_name}");
+ self.created_paths.push(file_path.clone());
if self.index_options.is_some() {
- self.created_paths.push(file_path.clone());
self.created_paths.push(format!(
"{bucket_dir}/{}",
data_file_to_file_index_file_name(&file_name)
diff --git a/crates/paimon/src/table/dedicated_format_file_writer.rs
b/crates/paimon/src/table/dedicated_format_file_writer.rs
index 4f3ee315..b9fa2714 100644
--- a/crates/paimon/src/table/dedicated_format_file_writer.rs
+++ b/crates/paimon/src/table/dedicated_format_file_writer.rs
@@ -304,6 +304,16 @@ impl AppendDedicatedFormatFileWriter {
Ok(())
}
+ pub(crate) async fn abort(&mut self) {
+ self.normal_writer.abort().await;
+ for writer in &mut self.blob_writers {
+ writer.writer.abort().await;
+ }
+ if let Some(writer) = &mut self.vector_writer {
+ writer.writer.abort().await;
+ }
+ }
+
pub(crate) async fn prepare_commit(&mut self) -> Result<Vec<DataFileMeta>>
{
let mut results = self.normal_writer.prepare_commit().await?;
diff --git a/crates/paimon/src/table/kv_file_writer.rs
b/crates/paimon/src/table/kv_file_writer.rs
index a4ca6830..24f3e3af 100644
--- a/crates/paimon/src/table/kv_file_writer.rs
+++ b/crates/paimon/src/table/kv_file_writer.rs
@@ -890,6 +890,25 @@ impl KeyValueFileWriter {
Ok(result)
}
+ pub(crate) async fn abort(&mut self) {
+ self.buffer.clear();
+ self.buffer_bytes = 0;
+ let bucket_path = bucket_path_under(
+ &self.config.table_location,
+ &self.config.partition_path,
+ self.config.bucket,
+ );
+ for file in self
+ .written_files
+ .drain(..)
+ .chain(self.written_changelog_files.drain(..))
+ {
+ for path in file.collect_files(&bucket_path) {
+ let _ = self.file_io.delete_file(&path).await;
+ }
+ }
+ }
+
/// Flush remaining buffer and return all written file metadata.
pub(crate) async fn prepare_commit(&mut self) -> Result<PreparedFiles> {
self.flush().await?;
diff --git a/crates/paimon/src/table/mod.rs b/crates/paimon/src/table/mod.rs
index 5f8923c1..c7363f3d 100644
--- a/crates/paimon/src/table/mod.rs
+++ b/crates/paimon/src/table/mod.rs
@@ -132,7 +132,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_message::CommitMessage;
+pub use commit_message::{CommitMessage, COMMIT_MESSAGE_SERIALIZER_VERSION};
pub use consumer_manager::ConsumerManager;
pub use cow_writer::{CopyOnWriteMergeWriter, FileInfo};
pub use data_evolution_writer::{DataEvolutionDeleteWriter,
DataEvolutionWriter};
diff --git a/crates/paimon/src/table/postpone_file_writer.rs
b/crates/paimon/src/table/postpone_file_writer.rs
index e651b3ed..1e3b64a4 100644
--- a/crates/paimon/src/table/postpone_file_writer.rs
+++ b/crates/paimon/src/table/postpone_file_writer.rs
@@ -67,6 +67,7 @@ pub(crate) struct PostponeFileWriter {
/// Timestamp captured when the current file was opened (used for
deterministic replay order).
current_file_creation_time: DateTime<Utc>,
written_files: Vec<DataFileMeta>,
+ created_paths: Vec<String>,
/// Background file close tasks spawned during rolling.
in_flight_closes: JoinSet<Result<DataFileMeta>>,
}
@@ -83,6 +84,7 @@ impl PostponeFileWriter {
current_file_start_seq: 0,
current_file_creation_time: Utc::now(),
written_files: Vec::new(),
+ created_paths: Vec::new(),
in_flight_closes: JoinSet::new(),
}
}
@@ -155,6 +157,18 @@ impl PostponeFileWriter {
Ok(())
}
+ pub(crate) async fn abort(&mut self) {
+ if let Some(writer) = self.current_writer.take() {
+ let _ = writer.close().await;
+ }
+ self.current_file_name = None;
+ while self.in_flight_closes.join_next().await.is_some() {}
+ for path in self.created_paths.drain(..) {
+ let _ = self.file_io.delete_file(&path).await;
+ }
+ self.written_files.clear();
+ }
+
pub(crate) async fn prepare_commit(&mut self) -> Result<Vec<DataFileMeta>>
{
self.close_current_file().await?;
while let Some(result) = self.in_flight_closes.join_next().await {
@@ -164,6 +178,7 @@ impl PostponeFileWriter {
})??;
self.written_files.push(meta);
}
+ self.created_paths.clear();
Ok(std::mem::take(&mut self.written_files))
}
@@ -213,6 +228,7 @@ impl PostponeFileWriter {
self.file_io.mkdirs(&format!("{bucket_dir}/")).await?;
let physical_schema = build_physical_schema(&user_schema);
let file_path = format!("{bucket_dir}/{file_name}");
+ self.created_paths.push(file_path.clone());
let output = self.file_io.new_output(&file_path)?;
let writer = create_format_writer(
&output,
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 8ce382f0..b0b8ec2c 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -156,6 +156,22 @@ fn validate_fixed_bucket_commit_mode(messages:
&[CommitMessage], overwrite: bool
Ok(())
}
+fn reject_compact_increment(messages: &[CommitMessage]) -> Result<()> {
+ // Java writes this increment in a separate COMPACT snapshot.
+ if messages.iter().any(|message| {
+ !message.compact_before.is_empty()
+ || !message.compact_after.is_empty()
+ || !message.compact_changelog_files.is_empty()
+ || !message.compact_new_index_files.is_empty()
+ || !message.compact_deleted_index_files.is_empty()
+ }) {
+ return Err(crate::Error::Unsupported {
+ message: "Committing a compact increment requires a separate
COMPACT snapshot.".into(),
+ });
+ }
+ Ok(())
+}
+
/// Table commit logic for Paimon write operations.
///
/// Provides atomic commit functionality including append, overwrite and
truncate
@@ -164,6 +180,7 @@ pub struct TableCommit {
snapshot_manager: SnapshotManager,
snapshot_commit: Arc<dyn SnapshotCommit>,
commit_user: String,
+ ignore_empty_commit: bool,
total_buckets: i32,
// commit config
commit_max_retries: u32,
@@ -205,6 +222,7 @@ impl TableCommit {
snapshot_manager,
snapshot_commit,
commit_user,
+ ignore_empty_commit: true,
total_buckets,
commit_max_retries,
commit_timeout_ms,
@@ -219,6 +237,79 @@ impl TableCommit {
}
}
+ /// 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;
+ self
+ }
+
+ /// Java StreamTableCommit.filterAndCommit: sort identifiers and return the
+ /// number of groups remaining after filtering against committed snapshots.
+ pub async fn filter_and_commit(
+ &self,
+ mut commits: Vec<(i64, Vec<CommitMessage>)>,
+ ) -> Result<usize> {
+
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
+ self.table.ensure_not_branch_reference_for_write()?;
+ commits.sort_by_key(|(id, _)| *id);
+ let latest = self.snapshot_manager.get_latest_snapshot().await?;
+ let mut pending = Vec::new();
+ for (id, messages) in commits {
+ if !self.is_committed_identifier(&latest, id).await? {
+ pending.push((id, messages));
+ }
+ }
+ // Java checks every pending checkpoint before publishing any snapshot.
+ // Filtered checkpoints may reference files already expired by
retention.
+ for (_, messages) in &pending {
+ self.check_recovery_files(messages).await?;
+ }
+ let count = pending.len();
+ for (id, messages) in pending {
+ self.commit_with_identifier(messages, id).await?;
+ }
+ Ok(count)
+ }
+
+ async fn check_recovery_files(&self, messages: &[CommitMessage]) ->
Result<()> {
+ let index_in_bucket =
+
CoreOptions::new(self.table.schema().options()).index_file_in_data_file_dir();
+ for message in messages {
+ let bucket_path = self.bucket_path(&message.partition,
message.bucket)?;
+ let mut paths = Vec::new();
+ for file in message
+ .new_files
+ .iter()
+ .chain(&message.new_changelog_files)
+ .chain(&message.compact_after)
+ .chain(&message.compact_changelog_files)
+ {
+ paths.extend(file.collect_files(&bucket_path));
+ }
+ for file in message
+ .new_index_files
+ .iter()
+ .chain(&message.compact_new_index_files)
+ {
+ paths.push(committed_index_file_path(
+ self.table.location().trim_end_matches('/'),
+ &bucket_path,
+ index_in_bucket,
+ file,
+ ));
+ }
+ for path in paths {
+ if !self.table.file_io().exists(&path).await? {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Cannot recover commit: file '{path}'
does not exist"),
+ source: None,
+ });
+ }
+ }
+ }
+ Ok(())
+ }
+
/// Commit new files in APPEND mode.
pub async fn commit(&self, commit_messages: Vec<CommitMessage>) ->
Result<()> {
self.commit_with_identifier(commit_messages, BATCH_COMMIT_IDENTIFIER)
@@ -264,17 +355,18 @@ impl TableCommit {
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.table.ensure_not_branch_reference_for_write()?;
+ reject_compact_increment(&commit_messages)?;
validate_fixed_bucket_commit_mode(&commit_messages, false)?;
validate_bucket_ownership(&commit_messages)?;
- if commit_messages.is_empty() {
+ if commit_messages.is_empty() && self.ignore_empty_commit {
return Ok(());
}
let entries = self.messages_to_entries(&commit_messages);
let changelog_entries =
self.messages_to_changelog_entries(&commit_messages);
let new_index_entries =
self.messages_to_index_entries(&commit_messages);
- let check_from_snapshot =
Self::min_check_from_snapshot(&commit_messages);
+ let check_from_snapshot = Self::check_from_snapshot(&commit_messages)?;
self.try_commit(
CommitEntriesPlan::Direct {
entries,
@@ -311,6 +403,7 @@ impl TableCommit {
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.table.ensure_not_branch_reference_for_write()?;
+ reject_compact_increment(&commit_messages)?;
validate_fixed_bucket_commit_mode(&commit_messages, false)?;
validate_bucket_ownership(&commit_messages)?;
@@ -321,7 +414,7 @@ impl TableCommit {
let entries = self.messages_to_entries(&commit_messages);
let changelog_entries =
self.messages_to_changelog_entries(&commit_messages);
let new_index_entries =
self.messages_to_index_entries(&commit_messages);
- let check_from_snapshot =
Self::min_check_from_snapshot(&commit_messages);
+ let check_from_snapshot = Self::check_from_snapshot(&commit_messages)?;
let result = self
.try_commit(
CommitEntriesPlan::Direct {
@@ -396,6 +489,7 @@ impl TableCommit {
// A commit validates against the existing snapshot.
CoreOptions::new(self.table.schema().options()).ensure_read_authorized()?;
self.table.ensure_not_branch_reference_for_write()?;
+ reject_compact_increment(&commit_messages)?;
validate_fixed_bucket_commit_mode(&commit_messages, true)?;
validate_bucket_ownership(&commit_messages)?;
@@ -426,7 +520,7 @@ impl TableCommit {
}
}
- let check_from_snapshot =
Self::min_check_from_snapshot(&commit_messages);
+ let check_from_snapshot = Self::check_from_snapshot(&commit_messages)?;
self.try_commit(
CommitEntriesPlan::Overwrite {
@@ -789,6 +883,8 @@ impl TableCommit {
.new_files
.iter()
.chain(message.new_changelog_files.iter())
+ .chain(message.compact_after.iter())
+ .chain(message.compact_changelog_files.iter())
{
for path in file.collect_files(&bucket_path) {
let _ = self.table.file_io().delete_file(&path).await;
@@ -805,7 +901,11 @@ impl TableCommit {
// through `indexFileFactory(partition, bucket)`. Deleting is
// best-effort, so a wrong path leaks the file silently instead of
// failing.
- for file in &message.new_index_files {
+ for file in message
+ .new_index_files
+ .iter()
+ .chain(&message.compact_new_index_files)
+ {
let path = committed_index_file_path(
table_path,
&bucket_path,
@@ -853,11 +953,11 @@ impl TableCommit {
let mut duplicate_check_start_snapshot_id: Option<i64> = None;
let mut retry_state: Option<Box<RetryState>> = None;
let start_time_ms = current_time_millis();
- // An identified destructive no-op must still record its identifier.
- // Otherwise a retry after an intervening write can execute the
operation
- // for the first time and delete data which was not present originally.
- let commit_empty_overwrite =
- filter_committed && plan.commit_kind_hint() ==
CommitKind::OVERWRITE;
+ // Java records static overwrite/truncate operations even when no files
+ // match. Dynamic empty overwrite exits before constructing this plan.
+ let commit_empty_overwrite = plan.commit_kind_hint() ==
CommitKind::OVERWRITE;
+ let commit_empty_append =
+ !self.ignore_empty_commit && matches!(plan,
CommitEntriesPlan::Direct { .. });
let mut filter_committed = filter_committed;
let mut publication_uncertain = false;
@@ -896,6 +996,7 @@ impl TableCommit {
&& resolved.changelog_entries.is_empty()
&& !resolved.index_manifest_changed
&& !commit_empty_overwrite
+ && !commit_empty_append
{
break;
}
@@ -3014,12 +3115,42 @@ impl TableCommit {
Ok(spec)
}
- /// Earliest source snapshot requested by row-id conflict checks.
- fn min_check_from_snapshot(messages: &[CommitMessage]) -> Option<i64> {
- messages
- .iter()
- .filter_map(|message| message.check_from_snapshot)
- .min()
+ /// Check conflicts from the earliest writer snapshot, validating row-id
baselines.
+ fn check_from_snapshot(messages: &[CommitMessage]) -> Result<Option<i64>> {
+ let mut check_from_snapshot: Option<i64> = None;
+ for message in messages {
+ let Some(snapshot) = message.check_from_snapshot else {
+ continue;
+ };
+ if snapshot < 0 {
+ return Err(crate::Error::DataInvalid {
+ message: format!("Invalid row-id check snapshot:
{snapshot}"),
+ source: None,
+ });
+ }
+ check_from_snapshot =
+ Some(check_from_snapshot.map_or(snapshot, |previous|
previous.min(snapshot)));
+ }
+ if check_from_snapshot.is_some() {
+ for message in messages {
+ if message.check_from_snapshot.is_some() {
+ continue;
+ }
+ if message
+ .new_files
+ .iter()
+ .chain(&message.deleted_files)
+ .any(|file| file.first_row_id.is_some())
+ {
+ return Err(crate::Error::DataInvalid {
+ message: "A row-id commit message is missing its
check-from snapshot."
+ .into(),
+ source: None,
+ });
+ }
+ }
+ }
+ Ok(check_from_snapshot)
}
/// Convert commit messages to manifest entries (ADD/DELETE kind).
@@ -3365,6 +3496,55 @@ mod tests {
use apache_avro::types::Value;
use chrono::{DateTime, Utc};
+ #[test]
+ fn check_from_snapshot_uses_minimum_baseline() {
+ let mut tagged = CommitMessage::new(Vec::new(), 0, Vec::new());
+ tagged.check_from_snapshot = Some(7);
+ assert_eq!(
+ TableCommit::check_from_snapshot(&[tagged.clone()]).unwrap(),
+ Some(7)
+ );
+ assert_eq!(TableCommit::check_from_snapshot(&[]).unwrap(), None);
+
+ let mut different = tagged.clone();
+ different.check_from_snapshot = Some(8);
+ let untagged = CommitMessage::new(Vec::new(), 0, Vec::new());
+ for messages in [
+ vec![tagged.clone(), different.clone(), untagged.clone()],
+ vec![different, untagged.clone(), tagged.clone()],
+ ] {
+ assert_eq!(
+ TableCommit::check_from_snapshot(&messages).unwrap(),
+ Some(7)
+ );
+ }
+ assert_eq!(TableCommit::check_from_snapshot(&[untagged]).unwrap(),
None);
+ }
+
+ #[test]
+ fn check_from_snapshot_rejects_invalid_or_missing_baselines() {
+ let mut tagged = CommitMessage::new(Vec::new(), 0, Vec::new());
+ tagged.check_from_snapshot = Some(7);
+ let mut negative = tagged.clone();
+ negative.check_from_snapshot = Some(-1);
+ assert!(TableCommit::check_from_snapshot(&[negative]).is_err());
+ let mut missing = CommitMessage::new(Vec::new(), 0,
vec![test_data_file("x", 1)]);
+ missing.new_files[0].first_row_id = Some(1);
+ assert!(TableCommit::check_from_snapshot(&[tagged, missing]).is_err());
+ }
+
+ #[test]
+ fn compact_increment_requires_separate_snapshot() {
+ let mut message = CommitMessage::new(Vec::new(), 0, Vec::new());
+ message.compact_before.push(test_data_file("before", 1));
+ message.compact_after.push(test_data_file("after", 1));
+ message
+ .compact_changelog_files
+ .push(test_data_file("changelog", 1));
+ assert!(reject_compact_increment(&[message]).is_err());
+ assert!(reject_compact_increment(&[CommitMessage::new(Vec::new(), 0,
Vec::new())]).is_ok());
+ }
+
#[tokio::test]
async fn test_query_auth_table_refuses_commit_paths() {
let table = crate::table::query_auth_table();
@@ -5178,7 +5358,7 @@ mod tests {
}
#[tokio::test]
- async fn test_truncate_missing_partition_is_noop() {
+ async fn test_truncate_missing_partition_records_java_overwrite_snapshot()
{
let file_io = test_file_io();
let table_path = "memory:/test_truncate_missing_partition";
setup_dirs(&file_io, table_path).await;
@@ -5203,7 +5383,9 @@ mod tests {
let snap_manager = SnapshotManager::new(file_io.clone(),
table_path.to_string());
let snapshot =
snap_manager.get_latest_snapshot().await.unwrap().unwrap();
- assert_eq!(snapshot.id(), 1);
+ assert_eq!(snapshot.id(), 2);
+ assert_eq!(snapshot.commit_kind(), &CommitKind::OVERWRITE);
+ assert_eq!(snapshot.delta_record_count(), Some(0));
assert_eq!(snapshot.total_record_count(), Some(100));
}
@@ -5465,10 +5647,15 @@ mod tests {
second_partial.first_row_id = Some(0);
second_partial.file_source = Some(0);
second_partial.write_cols = Some(vec!["name".to_string()]);
- let mut second_message = CommitMessage::new(partition, 0,
vec![second_partial]);
+ let mut second_message = CommitMessage::new(partition.clone(), 0,
vec![second_partial]);
second_message.check_from_snapshot = Some(1);
+ // A newer writer in the same commit must not hide the stale update.
+ let mut fresh_message =
+ CommitMessage::new(partition, 0,
vec![test_data_file("fresh.parquet", 1)]);
+ fresh_message.check_from_snapshot = Some(2);
+ fresh_message.new_files[0].file_source = Some(0);
- let result = commit.commit(vec![second_message]).await;
+ let result = commit.commit(vec![fresh_message, second_message]).await;
assert!(result.is_err());
let err_msg = result.unwrap_err().to_string();
@@ -5476,6 +5663,7 @@ mod tests {
err_msg.contains("multiple MERGE INTO operations have encountered
conflicts"),
"expected row-id/column conflict, got: {err_msg}"
);
+ assert_eq!(latest_snapshot(&file_io, table_path).await.unwrap().id(),
2);
}
#[tokio::test]
@@ -5512,14 +5700,23 @@ mod tests {
id_partial.first_row_id = Some(0);
id_partial.file_source = Some(0);
id_partial.write_cols = Some(vec!["id".to_string()]);
- let mut id_message = CommitMessage::new(partition, 0,
vec![id_partial]);
+ let mut id_message = CommitMessage::new(partition.clone(), 0,
vec![id_partial]);
id_message.check_from_snapshot = Some(1);
+ let mut fresh_message =
+ CommitMessage::new(partition, 0,
vec![test_data_file("fresh.parquet", 1)]);
+ fresh_message.check_from_snapshot = Some(2);
+ fresh_message.new_files[0].file_source = Some(0);
- commit.commit(vec![id_message]).await.unwrap();
+ commit
+ .commit(vec![id_message, fresh_message])
+ .await
+ .unwrap();
let snap_manager = SnapshotManager::new(file_io.clone(),
table_path.to_string());
let snapshot =
snap_manager.get_latest_snapshot().await.unwrap().unwrap();
assert_eq!(snapshot.id(), 3);
+ // Snapshot counts include the two partial-column files.
+ assert_eq!(snapshot.total_record_count(), Some(301));
}
#[tokio::test]
diff --git a/crates/paimon/src/table/table_write.rs
b/crates/paimon/src/table/table_write.rs
index 4fc57a58..f23c83f3 100644
--- a/crates/paimon/src/table/table_write.rs
+++ b/crates/paimon/src/table/table_write.rs
@@ -880,6 +880,19 @@ impl TableWrite {
Ok(())
}
+ /// Close without preparing another commit, discarding only outstanding
output.
+ /// Files already returned by prepare_commit belong to the caller.
+ pub async fn close(&mut self) {
+ for (_, writer) in self.partition_writers.drain() {
+ match writer {
+ FileWriter::Append(mut writer) => writer.abort().await,
+ FileWriter::AppendDedicated(mut writer) =>
writer.abort().await,
+ FileWriter::KeyValue(mut writer) => writer.abort().await,
+ FileWriter::Postpone(mut writer) => writer.abort().await,
+ }
+ }
+ }
+
/// Close all writers and collect CommitMessages for use with TableCommit.
/// Writers are cleared after this call, allowing the TableWrite to be
reused.
pub async fn prepare_commit(&mut self) -> Result<Vec<CommitMessage>> {
diff --git a/docs/src/python-binding.md b/docs/src/python-binding.md
index 318cfb7f..6e5ce569 100644
--- a/docs/src/python-binding.md
+++ b/docs/src/python-binding.md
@@ -237,7 +237,7 @@ batch = pa.record_batch(
)
# Create a write builder (shared commit_user for writer and committer)
-wb = table.new_write_builder()
+wb = table.new_batch_write_builder()
# Write batches
write = wb.new_write()
@@ -265,7 +265,7 @@ ctx.sql("INSERT INTO paimon.default.my_table VALUES (1,
'alice'), (2, 'bob')")
The input batch schema is strictly validated against the table schema:
field count, order, names, and types must match exactly. A `ValueError` is
raised on mismatch.
!!! note "Write Builder Consistency"
- The writer and committer must come from the same `WriteBuilder` — they
share a `commit_user` for duplicate-commit detection. Passing messages from one
builder's writer to another builder's committer will raise a `ValueError`.
+ The writer and committer must come from the same `BatchWriteBuilder` —
they share a `commit_user` for duplicate-commit detection. Passing messages
from one builder's writer to another builder's committer will raise a
`ValueError`.
## Column Projection
@@ -539,7 +539,7 @@ batch = pa.record_batch(
[pa.array([1, 2, 3], pa.int32()), pa.array(["alice", "bob", "carol"],
pa.string())],
names=["id", "name"],
)
-wb = table.new_write_builder()
+wb = table.new_batch_write_builder()
write = wb.new_write()
write.write_arrow(batch)
wb.new_commit().commit(write.prepare_commit())
