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 0d927e1 feat(c): add write/commit C FFI bindings (#522)
0d927e1 is described below
commit 0d927e18a3b22cf71941cf44a949871b5de725d6
Author: Socrates <[email protected]>
AuthorDate: Fri Jul 17 00:22:50 2026 +0800
feat(c): add write/commit C FFI bindings (#522)
Add write-side C FFI bindings following the existing read-side patterns:
API surface (14 functions):
- WriteBuilder: new_write_builder, free, with_overwrite, new_write,
new_commit
- TableWrite: free, write_arrow_batch (via Arrow C Data Interface),
prepare_commit
- TableCommit: free, commit, overwrite, truncate_table, abort
- CommitMessages: free
---
bindings/c/Cargo.toml | 4 +-
bindings/c/src/error.rs | 1 +
bindings/c/src/lib.rs | 3 +
bindings/c/src/result.rs | 26 +
bindings/c/src/tests.rs | 1508 +++++++++++++++++++++++++++++++
bindings/c/src/types.rs | 54 +-
bindings/c/src/write.rs | 830 +++++++++++++++++
crates/paimon/src/table/table_commit.rs | 461 +++++++++-
8 files changed, 2858 insertions(+), 29 deletions(-)
diff --git a/bindings/c/Cargo.toml b/bindings/c/Cargo.toml
index a6e27cf..6eb34c1 100644
--- a/bindings/c/Cargo.toml
+++ b/bindings/c/Cargo.toml
@@ -25,11 +25,13 @@ license.workspace = true
version.workspace = true
[lib]
-crate-type = ["cdylib", "staticlib"]
+crate-type = ["cdylib", "staticlib", "rlib"]
doc = false
[dependencies]
paimon = { path = "../../crates/paimon" }
tokio = { workspace = true, features = ["rt-multi-thread"] }
futures = "0.3"
+arrow = { workspace = true }
arrow-array = { workspace = true }
+arrow-schema = { workspace = true }
diff --git a/bindings/c/src/error.rs b/bindings/c/src/error.rs
index e8d0757..7b0a88f 100644
--- a/bindings/c/src/error.rs
+++ b/bindings/c/src/error.rs
@@ -58,6 +58,7 @@ impl paimon_error {
| paimon::Error::ColumnAlreadyExist { .. } =>
PaimonErrorCode::AlreadyExists,
paimon::Error::ConfigInvalid { .. }
| paimon::Error::DataTypeInvalid { .. }
+ | paimon::Error::DataInvalid { .. }
| paimon::Error::IdentifierInvalid { .. } =>
PaimonErrorCode::InvalidInput,
paimon::Error::IoUnexpected { .. } => PaimonErrorCode::IoError,
_ => PaimonErrorCode::Unexpected,
diff --git a/bindings/c/src/lib.rs b/bindings/c/src/lib.rs
index 19942c6..3e1367e 100644
--- a/bindings/c/src/lib.rs
+++ b/bindings/c/src/lib.rs
@@ -24,7 +24,10 @@ mod error;
mod identifier;
mod result;
mod table;
+#[cfg(test)]
+mod tests;
mod types;
+mod write;
use std::sync::OnceLock;
use tokio::runtime::Runtime;
diff --git a/bindings/c/src/result.rs b/bindings/c/src/result.rs
index 19d523c..216509f 100644
--- a/bindings/c/src/result.rs
+++ b/bindings/c/src/result.rs
@@ -77,3 +77,29 @@ pub struct paimon_result_next_batch {
pub batch: paimon_arrow_batch,
pub error: *mut paimon_error,
}
+
+// === Write/Commit result types ===
+
+#[repr(C)]
+pub struct paimon_result_write_builder {
+ pub write_builder: *mut paimon_write_builder,
+ pub error: *mut paimon_error,
+}
+
+#[repr(C)]
+pub struct paimon_result_table_write {
+ pub write: *mut paimon_table_write,
+ pub error: *mut paimon_error,
+}
+
+#[repr(C)]
+pub struct paimon_result_table_commit {
+ pub commit: *mut paimon_table_commit,
+ pub error: *mut paimon_error,
+}
+
+#[repr(C)]
+pub struct paimon_result_prepare_commit {
+ pub messages: *mut paimon_commit_messages,
+ pub error: *mut paimon_error,
+}
diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs
new file mode 100644
index 0000000..72ba582
--- /dev/null
+++ b/bindings/c/src/tests.rs
@@ -0,0 +1,1508 @@
+// 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.
+
+//! Tests for the paimon-c FFI bindings.
+//!
+//! Covers: read path (table, scan, plan, predicates, record batch streaming),
+//! write path (write builder, write_arrow_batch, prepare_commit, commit,
+//! overwrite, truncate, abort), and full write->read roundtrip.
+//!
+//! IMPORTANT: C FFI functions internally use `runtime().block_on()`. Tests
+//! must NOT wrap C FFI calls inside another `block_on`. Use the global
+//! runtime only for Rust-API setup (write_data_rust, setup_table_dirs).
+
+use std::ffi::{c_void, CString};
+use std::mem::ManuallyDrop;
+use std::process::Command;
+use std::ptr;
+use std::sync::Arc;
+
+use arrow::buffer::NullBuffer;
+use arrow_array::ffi::{FFI_ArrowArray, FFI_ArrowSchema};
+use arrow_array::{Array, Int32Array, RecordBatch, StringArray, StructArray};
+use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
+use paimon::catalog::Identifier;
+use paimon::io::FileIOBuilder;
+use paimon::spec::{DataType, IntType, Schema, TableSchema, VarCharType};
+use paimon::table::{SnapshotManager, Table};
+
+use crate::error::*;
+use crate::table::*;
+use crate::types::*;
+use crate::write::*;
+
+// =========================================================================
+// Helpers
+// =========================================================================
+
+fn memory_file_io() -> paimon::io::FileIO {
+ FileIOBuilder::new("memory").build().unwrap()
+}
+
+fn simple_table_schema() -> TableSchema {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .build()
+ .unwrap();
+ TableSchema::new(0, &schema)
+}
+
+fn not_null_table_schema() -> TableSchema {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .build()
+ .unwrap();
+ TableSchema::new(0, &schema)
+}
+
+unsafe fn wrap_table(table: Table) -> *mut paimon_table {
+ let inner = Box::into_raw(Box::new(table)) as *mut c_void;
+ Box::into_raw(Box::new(paimon_table { inner }))
+}
+
+unsafe fn unwrap_table(table: *mut paimon_table) {
+ let wrapper = Box::from_raw(table);
+ if !wrapper.inner.is_null() {
+ drop(Box::from_raw(wrapper.inner as *mut Table));
+ }
+}
+
+fn make_batch(ids: Vec<i32>, names: Vec<&str>) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, false),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(StringArray::from(names)),
+ ],
+ )
+ .unwrap()
+}
+
+fn make_type_mismatch_batch(ids: Vec<&str>, names: Vec<&str>) -> RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Utf8, false),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(StringArray::from(ids)),
+ Arc::new(StringArray::from(names)),
+ ],
+ )
+ .unwrap()
+}
+
+fn make_nullable_id_batch(ids: Vec<Option<i32>>, names: Vec<&str>) ->
RecordBatch {
+ let schema = Arc::new(ArrowSchema::new(vec![
+ ArrowField::new("id", ArrowDataType::Int32, true),
+ ArrowField::new("name", ArrowDataType::Utf8, true),
+ ]));
+ RecordBatch::try_new(
+ schema,
+ vec![
+ Arc::new(Int32Array::from(ids)),
+ Arc::new(StringArray::from(names)),
+ ],
+ )
+ .unwrap()
+}
+
+fn export_batch_to_ffi(
+ batch: RecordBatch,
+) -> (
+ Box<ManuallyDrop<FFI_ArrowArray>>,
+ Box<ManuallyDrop<FFI_ArrowSchema>>,
+) {
+ let struct_array = StructArray::from(batch);
+ let data = struct_array.to_data();
+ let ffi_array = FFI_ArrowArray::new(&data);
+ let ffi_schema = FFI_ArrowSchema::try_from(data.data_type()).unwrap();
+ (
+ Box::new(ManuallyDrop::new(ffi_array)),
+ Box::new(ManuallyDrop::new(ffi_schema)),
+ )
+}
+
+fn export_array_to_ffi(
+ array: &dyn Array,
+) -> (
+ Box<ManuallyDrop<FFI_ArrowArray>>,
+ Box<ManuallyDrop<FFI_ArrowSchema>>,
+) {
+ let data = array.to_data();
+ (
+ Box::new(ManuallyDrop::new(FFI_ArrowArray::new(&data))),
+ Box::new(ManuallyDrop::new(
+ FFI_ArrowSchema::try_from(data.data_type()).unwrap(),
+ )),
+ )
+}
+
+fn run_current_test_in_child(test_name: &str, env_name: &str, env_value: &str)
-> bool {
+ Command::new(std::env::current_exe().unwrap())
+ .arg("--exact")
+ .arg(test_name)
+ .arg("--nocapture")
+ .env(env_name, env_value)
+ .status()
+ .unwrap()
+ .success()
+}
+
+/// Use Rust API to write data (runs on global runtime via block_on).
+fn write_data_rust(table: &Table, batches: &[RecordBatch]) {
+ crate::runtime().block_on(async {
+ let wb = table.new_write_builder();
+ let mut tw = wb.new_write().unwrap();
+ for batch in batches {
+ tw.write_arrow_batch(batch).await.unwrap();
+ }
+ wb.new_commit()
+ .commit(tw.prepare_commit().await.unwrap())
+ .await
+ .unwrap();
+ });
+}
+
+/// Create directories needed by paimon (runs on global runtime via block_on).
+fn setup_table_dirs(file_io: &paimon::io::FileIO, path: &str) {
+ crate::runtime().block_on(async {
+ file_io.mkdirs(&format!("{path}/snapshot/")).await.unwrap();
+ file_io.mkdirs(&format!("{path}/manifest/")).await.unwrap();
+ });
+}
+
+/// Collect (id, name) rows from a C FFI record batch reader.
+/// Called OUTSIDE of any block_on — the C FFI functions use block_on
internally.
+unsafe fn collect_rows(reader: *mut paimon_record_batch_reader) -> Vec<(i32,
String)> {
+ let mut rows = Vec::new();
+ loop {
+ let result = paimon_record_batch_reader_next(reader);
+ assert!(result.error.is_null(), "reader_next should not error");
+ if result.batch.array.is_null() {
+ // End of stream — still need to free the empty batch structs
+ break;
+ }
+
+ // Take ownership of FFI structs via ptr::read (bitwise copy).
+ let ffi_array = ptr::read(result.batch.array as *const FFI_ArrowArray);
+ let ffi_schema = ptr::read(result.batch.schema as *const
FFI_ArrowSchema);
+
+ // from_ffi consumes ffi_array (by value), borrows ffi_schema.
+ let data = arrow_array::ffi::from_ffi(ffi_array, &ffi_schema).unwrap();
+
+ // Zero out the original Box allocations so release is a no-op.
+ // The data ownership was transferred to `data` via from_ffi.
+ ptr::write(
+ result.batch.array as *mut FFI_ArrowArray,
+ FFI_ArrowArray::empty(),
+ );
+ ptr::write(
+ result.batch.schema as *mut FFI_ArrowSchema,
+ FFI_ArrowSchema::empty(),
+ );
+ paimon_arrow_batch_free(result.batch);
+
+ let struct_array = StructArray::from(data);
+ let batch = RecordBatch::from(struct_array);
+
+ let id_arr = batch
+ .column_by_name("id")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<Int32Array>()
+ .unwrap();
+ let name_arr = batch
+ .column_by_name("name")
+ .unwrap()
+ .as_any()
+ .downcast_ref::<StringArray>()
+ .unwrap();
+
+ for i in 0..batch.num_rows() {
+ let name = if name_arr.is_null(i) {
+ String::new()
+ } else {
+ name_arr.value(i).to_string()
+ };
+ rows.push((id_arr.value(i), name));
+ }
+
+ // ffi_schema drops here — its release frees schema-owned memory
+ // (format string, private_data), which is correct per the C Data
+ // Interface spec.
+ }
+ rows.sort_by_key(|r| r.0);
+ rows
+}
+
+/// Full read via C FFI: read_builder -> scan -> plan -> read -> stream ->
rows.
+/// Called OUTSIDE of any block_on — the C FFI functions use block_on
internally.
+unsafe fn read_rows_ffi(table: *const paimon_table) -> Vec<(i32, String)> {
+ let rb_result = paimon_table_new_read_builder(table);
+ assert!(rb_result.error.is_null());
+ let rb = rb_result.read_builder;
+
+ let scan_result = paimon_read_builder_new_scan(rb);
+ assert!(scan_result.error.is_null());
+ let scan = scan_result.scan;
+
+ let plan_result = paimon_table_scan_plan(scan);
+ assert!(plan_result.error.is_null());
+ let plan = plan_result.plan;
+
+ let read_result = paimon_read_builder_new_read(rb);
+ assert!(read_result.error.is_null());
+ let read = read_result.read;
+
+ let reader_result = paimon_table_read_to_arrow(read, plan, 0, usize::MAX);
+ assert!(reader_result.error.is_null());
+ let reader = reader_result.reader;
+
+ let rows = collect_rows(reader);
+
+ paimon_record_batch_reader_free(reader);
+ paimon_table_read_free(read);
+ paimon_plan_free(plan);
+ paimon_table_scan_free(scan);
+ paimon_read_builder_free(rb);
+
+ rows
+}
+
+// =========================================================================
+// Read path tests
+// =========================================================================
+
+#[test]
+fn test_read_empty_table() {
+ let path = "memory:/test_read_empty";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+ let rows = unsafe { read_rows_ffi(handle) };
+ assert!(rows.is_empty());
+ unsafe { unwrap_table(handle) };
+}
+
+#[test]
+fn test_read_with_data() {
+ let path = "memory:/test_read_with_data";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ write_data_rust(&table, &[make_batch(vec![1, 2, 3], vec!["a", "b", "c"])]);
+ let handle = unsafe { wrap_table(table) };
+ let rows = unsafe { read_rows_ffi(handle) };
+ assert_eq!(
+ rows,
+ vec![(1, "a".into()), (2, "b".into()), (3, "c".into())]
+ );
+ unsafe { unwrap_table(handle) };
+}
+
+#[test]
+fn test_read_with_projection() {
+ let path = "memory:/test_read_proj";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ write_data_rust(&table, &[make_batch(vec![1, 2], vec!["x", "y"])]);
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let rb_result = paimon_table_new_read_builder(handle);
+ assert!(rb_result.error.is_null());
+ let rb = rb_result.read_builder;
+
+ let col = CString::new("id").unwrap();
+ let cols = [col.as_ptr(), ptr::null()];
+ let err = paimon_read_builder_with_projection(rb, cols.as_ptr());
+ assert!(err.is_null());
+
+ let scan_result = paimon_read_builder_new_scan(rb);
+ assert!(scan_result.error.is_null());
+ let scan = scan_result.scan;
+
+ let plan_result = paimon_table_scan_plan(scan);
+ assert!(plan_result.error.is_null());
+ let plan = plan_result.plan;
+
+ let read_result = paimon_read_builder_new_read(rb);
+ assert!(read_result.error.is_null());
+ let read = read_result.read;
+
+ let reader_result = paimon_table_read_to_arrow(read, plan, 0,
usize::MAX);
+ assert!(reader_result.error.is_null());
+ let reader = reader_result.reader;
+
+ let result = paimon_record_batch_reader_next(reader);
+ assert!(result.error.is_null());
+ assert!(!result.batch.array.is_null());
+
+ let ffi_schema = ptr::read(result.batch.schema as *const
FFI_ArrowSchema);
+ let arrow_schema: ArrowSchema = (&ffi_schema).try_into().unwrap();
+ assert_eq!(arrow_schema.fields().len(), 1);
+ assert_eq!(arrow_schema.field(0).name(), "id");
+ std::mem::forget(ffi_schema);
+
+ paimon_arrow_batch_free(result.batch);
+
+ paimon_record_batch_reader_free(reader);
+ paimon_table_read_free(read);
+ paimon_plan_free(plan);
+ paimon_table_scan_free(scan);
+ paimon_read_builder_free(rb);
+ }
+
+ unsafe { unwrap_table(handle) };
+}
+
+// =========================================================================
+// Predicate tests
+// =========================================================================
+
+unsafe fn build_predicate_equal(
+ table: *const paimon_table,
+ column: &str,
+ int_val: i32,
+) -> *mut paimon_predicate {
+ let col = CString::new(column).unwrap();
+ let datum = paimon_datum {
+ tag: 3,
+ int_val: int_val as i64,
+ double_val: 0.0,
+ str_data: ptr::null(),
+ str_len: 0,
+ int_val2: 0,
+ uint_val: 0,
+ uint_val2: 0,
+ };
+ let result = paimon_predicate_equal(table, col.as_ptr(), datum);
+ assert!(result.error.is_null());
+ result.predicate
+}
+
+#[test]
+fn test_predicate_basics() {
+ let path = "memory:/test_predicate_basics";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ write_data_rust(&table, &[make_batch(vec![1, 2, 3], vec!["a", "b", "c"])]);
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let pred = build_predicate_equal(handle, "id", 1);
+ assert!(!pred.is_null());
+ paimon_predicate_free(pred);
+
+ let col = CString::new("name").unwrap();
+ let nn_result = paimon_predicate_is_not_null(handle, col.as_ptr());
+ assert!(nn_result.error.is_null());
+ paimon_predicate_free(nn_result.predicate);
+
+ let idcol = CString::new("id").unwrap();
+ let datum1 = paimon_datum {
+ tag: 3,
+ int_val: 1,
+ double_val: 0.0,
+ str_data: ptr::null(),
+ str_len: 0,
+ int_val2: 0,
+ uint_val: 0,
+ uint_val2: 0,
+ };
+ let datum2 = paimon_datum {
+ tag: 3,
+ int_val: 2,
+ double_val: 0.0,
+ str_data: ptr::null(),
+ str_len: 0,
+ int_val2: 0,
+ uint_val: 0,
+ uint_val2: 0,
+ };
+ let datums = [datum1, datum2];
+ let in_result = paimon_predicate_is_in(handle, idcol.as_ptr(),
datums.as_ptr(), 2);
+ assert!(in_result.error.is_null());
+ paimon_predicate_free(in_result.predicate);
+ }
+
+ unsafe { unwrap_table(handle) };
+}
+
+#[test]
+fn test_predicate_scan_filter() {
+ let path = "memory:/test_predicate_filter";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ write_data_rust(
+ &table,
+ &[make_batch(vec![1, 2, 3, 4], vec!["a", "b", "c", "d"])],
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let rb_result = paimon_table_new_read_builder(handle);
+ assert!(rb_result.error.is_null());
+ let rb = rb_result.read_builder;
+
+ let col = CString::new("id").unwrap();
+ let datum = paimon_datum {
+ tag: 3,
+ int_val: 3,
+ double_val: 0.0,
+ str_data: ptr::null(),
+ str_len: 0,
+ int_val2: 0,
+ uint_val: 0,
+ uint_val2: 0,
+ };
+ let pred_result = paimon_predicate_less_than(handle, col.as_ptr(),
datum);
+ assert!(pred_result.error.is_null());
+
+ let err = paimon_read_builder_with_filter(rb, pred_result.predicate);
+ assert!(err.is_null());
+
+ let scan_result = paimon_read_builder_new_scan(rb);
+ assert!(scan_result.error.is_null());
+ let scan = scan_result.scan;
+
+ let plan_result = paimon_table_scan_plan(scan);
+ assert!(plan_result.error.is_null());
+ let plan = plan_result.plan;
+
+ let read_result = paimon_read_builder_new_read(rb);
+ assert!(read_result.error.is_null());
+ let read = read_result.read;
+
+ let reader_result = paimon_table_read_to_arrow(read, plan, 0,
usize::MAX);
+ assert!(reader_result.error.is_null());
+ let reader = reader_result.reader;
+
+ let rows = collect_rows(reader);
+ assert_eq!(rows, vec![(1, "a".into()), (2, "b".into())]);
+
+ paimon_record_batch_reader_free(reader);
+ paimon_table_read_free(read);
+ paimon_plan_free(plan);
+ paimon_table_scan_free(scan);
+ paimon_read_builder_free(rb);
+ }
+
+ unsafe { unwrap_table(handle) };
+}
+
+#[test]
+fn test_predicate_and_or_not() {
+ let path = "memory:/test_predicate_combinators";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ write_data_rust(&table, &[make_batch(vec![1, 2, 3], vec!["a", "b", "c"])]);
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let col = CString::new("id").unwrap();
+ let datum1 = paimon_datum {
+ tag: 3,
+ int_val: 1,
+ double_val: 0.0,
+ str_data: ptr::null(),
+ str_len: 0,
+ int_val2: 0,
+ uint_val: 0,
+ uint_val2: 0,
+ };
+ let datum2 = paimon_datum {
+ tag: 3,
+ int_val: 3,
+ double_val: 0.0,
+ str_data: ptr::null(),
+ str_len: 0,
+ int_val2: 0,
+ uint_val: 0,
+ uint_val2: 0,
+ };
+
+ let p1 = paimon_predicate_greater_than(handle, col.as_ptr(), datum1);
+ let p2 = paimon_predicate_less_than(handle, col.as_ptr(), datum2);
+ assert!(p1.error.is_null() && p2.error.is_null());
+
+ let p_and = paimon_predicate_and(p1.predicate, p2.predicate);
+ assert!(!p_and.is_null());
+
+ let rb_result = paimon_table_new_read_builder(handle);
+ let rb = rb_result.read_builder;
+ paimon_read_builder_with_filter(rb, p_and);
+
+ let scan = paimon_read_builder_new_scan(rb);
+ let plan = paimon_table_scan_plan(scan.scan);
+ let read = paimon_read_builder_new_read(rb);
+ let reader = paimon_table_read_to_arrow(read.read, plan.plan, 0,
usize::MAX);
+
+ let rows = collect_rows(reader.reader);
+ assert_eq!(rows, vec![(2, "b".into())]);
+
+ paimon_record_batch_reader_free(reader.reader);
+ paimon_table_read_free(read.read);
+ paimon_plan_free(plan.plan);
+ paimon_table_scan_free(scan.scan);
+ paimon_read_builder_free(rb);
+ }
+
+ unsafe { unwrap_table(handle) };
+}
+
+// =========================================================================
+// Write path tests
+// =========================================================================
+
+#[test]
+fn test_write_new_builder_and_free() {
+ let path = "memory:/test_write_builder";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+ unsafe {
+ let result = paimon_table_new_write_builder(handle);
+ assert!(result.error.is_null());
+ assert!(!result.write_builder.is_null());
+ paimon_write_builder_free(result.write_builder);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_write_commit_read_roundtrip() {
+ let path = "memory:/test_write_roundtrip";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb_result = paimon_table_new_write_builder(handle);
+ assert!(wb_result.error.is_null());
+ let wb = wb_result.write_builder;
+
+ let tw_result = paimon_write_builder_new_write(wb);
+ assert!(tw_result.error.is_null());
+ let tw = tw_result.write;
+
+ let batch = make_batch(vec![1, 2, 3], vec!["a", "b", "c"]);
+ let (array_box, schema_box) = export_batch_to_ffi(batch);
+ let array_ptr = (&**array_box) as *const FFI_ArrowArray as *mut c_void;
+ let schema_ptr = (&**schema_box) as *const FFI_ArrowSchema as *mut
c_void;
+
+ let err = paimon_table_write_write_arrow_batch(tw, array_ptr,
schema_ptr);
+ assert!(err.is_null());
+
+ let pc_result = paimon_table_write_prepare_commit(tw);
+ assert!(pc_result.error.is_null());
+ assert!(!pc_result.messages.is_null());
+
+ let tc_result = paimon_write_builder_new_commit(wb);
+ assert!(tc_result.error.is_null());
+ let tc = tc_result.commit;
+
+ let err = paimon_table_commit_commit(tc, pc_result.messages);
+ assert!(err.is_null());
+ paimon_commit_messages_free(pc_result.messages);
+
+ let rows = read_rows_ffi(handle);
+ assert_eq!(
+ rows,
+ vec![(1, "a".into()), (2, "b".into()), (3, "c".into())]
+ );
+
+ paimon_table_commit_free(tc);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_write_arrow_batch_moves_ffi_structs() {
+ let path = "memory:/test_write_arrow_batch_moves_ffi_structs";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb = paimon_table_new_write_builder(handle).write_builder;
+ let tw = paimon_write_builder_new_write(wb).write;
+ let (array, schema) = export_batch_to_ffi(make_batch(vec![1],
vec!["a"]));
+
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**array) as *const FFI_ArrowArray as *mut c_void,
+ (&**schema) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(err.is_null());
+ assert!(array.is_released(), "import must clear ArrowArray.release");
+ assert!(
+ schema.release.is_none(),
+ "import must clear ArrowSchema.release"
+ );
+
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_write_arrow_batch_rejects_table_schema_mismatch() {
+ let path = "memory:/test_write_arrow_batch_rejects_table_schema_mismatch";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb = paimon_table_new_write_builder(handle).write_builder;
+ let tw = paimon_write_builder_new_write(wb).write;
+ let (array, schema) =
export_batch_to_ffi(make_type_mismatch_batch(vec!["1"], vec!["a"]));
+
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**array) as *const FFI_ArrowArray as *mut c_void,
+ (&**schema) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(!err.is_null(), "schema mismatch must be rejected");
+ paimon_error_free(err);
+
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_write_arrow_batch_rejects_null_for_not_null_field() {
+ let path =
"memory:/test_write_arrow_batch_rejects_null_for_not_null_field";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ not_null_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb = paimon_table_new_write_builder(handle).write_builder;
+ let tw = paimon_write_builder_new_write(wb).write;
+ let (array, schema) =
export_batch_to_ffi(make_nullable_id_batch(vec![None], vec!["a"]));
+
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**array) as *const FFI_ArrowArray as *mut c_void,
+ (&**schema) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(!err.is_null(), "NULL must be rejected for a NOT NULL field");
+ paimon_error_free(err);
+
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_write_arrow_batch_rejects_invalid_root_arrays_without_aborting() {
+ const CHILD_ENV: &str = "PAIMON_C_INVALID_ROOT_CHILD";
+ if let Ok(mode) = std::env::var(CHILD_ENV) {
+ let path = format!("memory:/test_invalid_root_{mode}");
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, &path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path,
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb = paimon_table_new_write_builder(handle).write_builder;
+ let tw = paimon_write_builder_new_write(wb).write;
+ let (array, schema) = if mode == "non_struct" {
+ let array = Int32Array::from(vec![1]);
+ export_array_to_ffi(&array)
+ } else {
+ let fields =
+ vec![Arc::new(ArrowField::new("id", ArrowDataType::Int32,
false))].into();
+ let array = StructArray::new(
+ fields,
+ vec![Arc::new(Int32Array::from(vec![1]))],
+ Some(NullBuffer::new_null(1)),
+ );
+ export_array_to_ffi(&array)
+ };
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**array) as *const FFI_ArrowArray as *mut c_void,
+ (&**schema) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(!err.is_null(), "invalid root array must return an error");
+ paimon_error_free(err);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ unwrap_table(handle);
+ }
+ return;
+ }
+
+ for mode in ["non_struct", "nullable_struct"] {
+ assert!(
+ run_current_test_in_child(
+
"tests::test_write_arrow_batch_rejects_invalid_root_arrays_without_aborting",
+ CHILD_ENV,
+ mode,
+ ),
+ "{mode} input must return an error instead of aborting the process"
+ );
+ }
+}
+
+#[test]
+fn test_commit_rejects_messages_from_another_table() {
+ let file_io = memory_file_io();
+ let source_path = "memory:/test_commit_provenance_source";
+ let target_path = "memory:/test_commit_provenance_target";
+ setup_table_dirs(&file_io, source_path);
+ setup_table_dirs(&file_io, target_path);
+ let source = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "source"),
+ source_path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let target = Table::new(
+ file_io,
+ Identifier::new("default", "target"),
+ target_path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let source_handle = unsafe { wrap_table(source) };
+ let target_handle = unsafe { wrap_table(target) };
+
+ unsafe {
+ let source_wb =
paimon_table_new_write_builder(source_handle).write_builder;
+ let source_tw = paimon_write_builder_new_write(source_wb).write;
+ let (array, schema) = export_batch_to_ffi(make_batch(vec![1],
vec!["a"]));
+ let err = paimon_table_write_write_arrow_batch(
+ source_tw,
+ (&**array) as *const FFI_ArrowArray as *mut c_void,
+ (&**schema) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(err.is_null());
+ let messages = paimon_table_write_prepare_commit(source_tw).messages;
+
+ let target_wb =
paimon_table_new_write_builder(target_handle).write_builder;
+ let target_commit = paimon_write_builder_new_commit(target_wb).commit;
+ let err = paimon_table_commit_commit(target_commit, messages);
+ assert!(
+ !err.is_null(),
+ "a committer must reject messages prepared for another table"
+ );
+ paimon_error_free(err);
+
+ paimon_commit_messages_free(messages);
+ paimon_table_commit_free(target_commit);
+ paimon_write_builder_free(target_wb);
+ paimon_table_write_free(source_tw);
+ paimon_write_builder_free(source_wb);
+ unwrap_table(target_handle);
+ unwrap_table(source_handle);
+ }
+}
+
+#[test]
+fn test_commit_rejects_messages_from_different_builder_identity() {
+ let path = "memory:/test_commit_builder_provenance";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let writer_wb = paimon_table_new_write_builder(handle).write_builder;
+ let tw = paimon_write_builder_new_write(writer_wb).write;
+ let (array, schema) = export_batch_to_ffi(make_batch(vec![1],
vec!["a"]));
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**array) as *const FFI_ArrowArray as *mut c_void,
+ (&**schema) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(err.is_null());
+ let messages = paimon_table_write_prepare_commit(tw).messages;
+
+ let other_wb = paimon_table_new_write_builder(handle).write_builder;
+ let wrong_commit = paimon_write_builder_new_commit(other_wb).commit;
+ let err = paimon_table_commit_commit(wrong_commit, messages);
+ assert!(
+ !err.is_null(),
+ "messages from another commit_user must be rejected"
+ );
+ paimon_error_free(err);
+
+ let correct_commit = paimon_write_builder_new_commit(writer_wb).commit;
+ let err = paimon_table_commit_commit(correct_commit, messages);
+ assert!(err.is_null(), "rejected messages must remain reusable");
+
+ paimon_commit_messages_free(messages);
+ paimon_table_commit_free(correct_commit);
+ paimon_table_commit_free(wrong_commit);
+ paimon_write_builder_free(other_wb);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(writer_wb);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_commit_messages_live_until_explicit_free() {
+ const CHILD_ENV: &str = "PAIMON_C_MESSAGES_LIFETIME_CHILD";
+ if std::env::var_os(CHILD_ENV).is_some() {
+ let path = "memory:/test_commit_messages_lifetime";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb = paimon_table_new_write_builder(handle).write_builder;
+ let tw = paimon_write_builder_new_write(wb).write;
+ let (array, schema) = export_batch_to_ffi(make_batch(vec![1],
vec!["a"]));
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**array) as *const FFI_ArrowArray as *mut c_void,
+ (&**schema) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(err.is_null());
+ let messages = paimon_table_write_prepare_commit(tw).messages;
+ let commit = paimon_write_builder_new_commit(wb).commit;
+ let err = paimon_table_commit_commit(commit, messages);
+ assert!(err.is_null());
+
+ paimon_commit_messages_free(messages);
+ paimon_table_commit_free(commit);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ unwrap_table(handle);
+ }
+ return;
+ }
+
+ assert!(
+ run_current_test_in_child(
+ "tests::test_commit_messages_live_until_explicit_free",
+ CHILD_ENV,
+ "1",
+ ),
+ "commit must not destroy a handle that callers are required to free"
+ );
+}
+
+#[test]
+fn test_caller_supplied_commit_identity_is_shared_and_persisted() {
+ let path = "memory:/test_caller_supplied_commit_identity";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io.clone(),
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+ let commit_user = CString::new("doris-load-job-42").unwrap();
+
+ unsafe {
+ let writer_wb =
+ paimon_table_new_write_builder_with_commit_user(handle,
commit_user.as_ptr())
+ .write_builder;
+ let committer_wb =
+ paimon_table_new_write_builder_with_commit_user(handle,
commit_user.as_ptr())
+ .write_builder;
+ let tw = paimon_write_builder_new_write(writer_wb).write;
+ let (array, schema) = export_batch_to_ffi(make_batch(vec![1],
vec!["a"]));
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**array) as *const FFI_ArrowArray as *mut c_void,
+ (&**schema) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(err.is_null());
+ let messages = paimon_table_write_prepare_commit(tw).messages;
+ let commit = paimon_write_builder_new_commit(committer_wb).commit;
+ let err = paimon_table_commit_commit_with_identifier(commit, messages,
42);
+ assert!(err.is_null());
+
+ let retry_wb =
+ paimon_table_new_write_builder_with_commit_user(handle,
commit_user.as_ptr())
+ .write_builder;
+ let retry_commit = paimon_write_builder_new_commit(retry_wb).commit;
+ let err =
paimon_table_commit_filter_and_commit_with_identifier(retry_commit, messages,
42);
+ assert!(
+ err.is_null(),
+ "retrying the same identity must be idempotent"
+ );
+
+ paimon_commit_messages_free(messages);
+ paimon_table_commit_free(retry_commit);
+ paimon_write_builder_free(retry_wb);
+ paimon_table_commit_free(commit);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(committer_wb);
+ paimon_write_builder_free(writer_wb);
+ unwrap_table(handle);
+ }
+
+ let snapshot = crate::runtime()
+ .block_on(SnapshotManager::new(file_io,
path.to_string()).get_latest_snapshot())
+ .unwrap()
+ .unwrap();
+ assert_eq!(snapshot.commit_user(), "doris-load-job-42");
+ assert_eq!(snapshot.commit_identifier(), 42);
+ assert_eq!(snapshot.id(), 1, "retry must not create another snapshot");
+}
+
+#[test]
+fn test_commit_messages_merge_preserves_all_writer_files() {
+ let path = "memory:/test_commit_messages_merge";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+ let commit_user = CString::new("distributed-job-7").unwrap();
+
+ unsafe {
+ let wb1 = paimon_table_new_write_builder_with_commit_user(handle,
commit_user.as_ptr())
+ .write_builder;
+ let wb2 = paimon_table_new_write_builder_with_commit_user(handle,
commit_user.as_ptr())
+ .write_builder;
+ let tw1 = paimon_write_builder_new_write(wb1).write;
+ let tw2 = paimon_write_builder_new_write(wb2).write;
+
+ for (tw, ids, names) in [(tw1, vec![1], vec!["a"]), (tw2, vec![2],
vec!["b"])] {
+ let (array, schema) = export_batch_to_ffi(make_batch(ids, names));
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**array) as *const FFI_ArrowArray as *mut c_void,
+ (&**schema) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(err.is_null());
+ }
+
+ let messages1 = paimon_table_write_prepare_commit(tw1).messages;
+ let messages2 = paimon_table_write_prepare_commit(tw2).messages;
+ let err = paimon_commit_messages_merge(messages1, messages2);
+ assert!(err.is_null());
+
+ let commit = paimon_write_builder_new_commit(wb1).commit;
+ let err = paimon_table_commit_commit_with_identifier(commit,
messages1, 7);
+ assert!(err.is_null());
+ assert_eq!(
+ read_rows_ffi(handle),
+ vec![(1, "a".into()), (2, "b".into())]
+ );
+
+ paimon_commit_messages_free(messages2);
+ paimon_commit_messages_free(messages1);
+ paimon_table_commit_free(commit);
+ paimon_table_write_free(tw2);
+ paimon_table_write_free(tw1);
+ paimon_write_builder_free(wb2);
+ paimon_write_builder_free(wb1);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_write_multiple_batches() {
+ let path = "memory:/test_write_multi_batch";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb_result = paimon_table_new_write_builder(handle);
+ let wb = wb_result.write_builder;
+
+ let tw_result = paimon_write_builder_new_write(wb);
+ let tw = tw_result.write;
+
+ for (ids, names) in [(vec![1], vec!["a"]), (vec![2], vec!["b"])] {
+ let batch = make_batch(ids, names);
+ let (ab, sb) = export_batch_to_ffi(batch);
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**ab) as *const FFI_ArrowArray as *mut c_void,
+ (&**sb) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(err.is_null());
+ }
+
+ let pc_result = paimon_table_write_prepare_commit(tw);
+ assert!(pc_result.error.is_null());
+
+ let tc_result = paimon_write_builder_new_commit(wb);
+ let tc = tc_result.commit;
+ let err = paimon_table_commit_commit(tc, pc_result.messages);
+ assert!(err.is_null());
+ paimon_commit_messages_free(pc_result.messages);
+
+ let rows = read_rows_ffi(handle);
+ assert_eq!(rows, vec![(1, "a".into()), (2, "b".into())]);
+
+ paimon_table_commit_free(tc);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_commit_empty_messages_noop() {
+ let path = "memory:/test_commit_empty";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb_result = paimon_table_new_write_builder(handle);
+ let wb = wb_result.write_builder;
+
+ let tw_result = paimon_write_builder_new_write(wb);
+ let tw = tw_result.write;
+ let pc_result = paimon_table_write_prepare_commit(tw);
+ assert!(pc_result.error.is_null());
+ assert!(!pc_result.messages.is_null());
+
+ let tc_result = paimon_write_builder_new_commit(wb);
+ let tc = tc_result.commit;
+ let err = paimon_table_commit_commit(tc, pc_result.messages);
+ assert!(err.is_null());
+ paimon_commit_messages_free(pc_result.messages);
+
+ let rows = read_rows_ffi(handle);
+ assert!(rows.is_empty());
+
+ paimon_table_commit_free(tc);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_write_overwrite_mode() {
+ let path = "memory:/test_write_overwrite";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ // First commit: write [1, 2]
+ {
+ let wb_result = paimon_table_new_write_builder(handle);
+ let wb = wb_result.write_builder;
+ let tw_result = paimon_write_builder_new_write(wb);
+ let tw = tw_result.write;
+
+ let batch = make_batch(vec![1, 2], vec!["a", "b"]);
+ let (ab, sb) = export_batch_to_ffi(batch);
+ paimon_table_write_write_arrow_batch(
+ tw,
+ (&**ab) as *const FFI_ArrowArray as *mut c_void,
+ (&**sb) as *const FFI_ArrowSchema as *mut c_void,
+ );
+
+ let pc = paimon_table_write_prepare_commit(tw);
+ let tc_result = paimon_write_builder_new_commit(wb);
+ paimon_table_commit_commit(tc_result.commit, pc.messages);
+ paimon_commit_messages_free(pc.messages);
+
+ paimon_table_commit_free(tc_result.commit);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ }
+
+ // Second commit with overwrite: write [3, 4]
+ {
+ let wb_result = paimon_table_new_write_builder(handle);
+ let wb = wb_result.write_builder;
+
+ let err = paimon_write_builder_with_overwrite(wb);
+ assert!(err.is_null());
+
+ let tw_result = paimon_write_builder_new_write(wb);
+ let tw = tw_result.write;
+
+ let batch = make_batch(vec![3, 4], vec!["c", "d"]);
+ let (ab, sb) = export_batch_to_ffi(batch);
+ paimon_table_write_write_arrow_batch(
+ tw,
+ (&**ab) as *const FFI_ArrowArray as *mut c_void,
+ (&**sb) as *const FFI_ArrowSchema as *mut c_void,
+ );
+
+ let pc = paimon_table_write_prepare_commit(tw);
+ let tc_result = paimon_write_builder_new_commit(wb);
+ paimon_table_commit_overwrite(tc_result.commit, pc.messages);
+ paimon_commit_messages_free(pc.messages);
+
+ paimon_table_commit_free(tc_result.commit);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ }
+
+ let rows = read_rows_ffi(handle);
+ assert_eq!(rows, vec![(3, "c".into()), (4, "d".into())]);
+
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_truncate_table() {
+ let path = "memory:/test_truncate";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ // Write some data first
+ {
+ let wb_result = paimon_table_new_write_builder(handle);
+ let wb = wb_result.write_builder;
+ let tw_result = paimon_write_builder_new_write(wb);
+ let tw = tw_result.write;
+ let batch = make_batch(vec![1, 2], vec!["a", "b"]);
+ let (ab, sb) = export_batch_to_ffi(batch);
+ paimon_table_write_write_arrow_batch(
+ tw,
+ (&**ab) as *const FFI_ArrowArray as *mut c_void,
+ (&**sb) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ let pc = paimon_table_write_prepare_commit(tw);
+ let tc_result = paimon_write_builder_new_commit(wb);
+ paimon_table_commit_commit(tc_result.commit, pc.messages);
+ paimon_commit_messages_free(pc.messages);
+ paimon_table_commit_free(tc_result.commit);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ }
+
+ // Truncate
+ {
+ let wb_result = paimon_table_new_write_builder(handle);
+ let wb = wb_result.write_builder;
+ let tc_result = paimon_write_builder_new_commit(wb);
+ let err = paimon_table_commit_truncate_table(tc_result.commit);
+ assert!(err.is_null());
+ paimon_table_commit_free(tc_result.commit);
+ paimon_write_builder_free(wb);
+ }
+
+ let rows = read_rows_ffi(handle);
+ assert!(rows.is_empty());
+
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_abort_commit() {
+ let path = "memory:/test_abort";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb_result = paimon_table_new_write_builder(handle);
+ let wb = wb_result.write_builder;
+
+ let tw_result = paimon_write_builder_new_write(wb);
+ let tw = tw_result.write;
+
+ let batch = make_batch(vec![1], vec!["a"]);
+ let (ab, sb) = export_batch_to_ffi(batch);
+ let err = paimon_table_write_write_arrow_batch(
+ tw,
+ (&**ab) as *const FFI_ArrowArray as *mut c_void,
+ (&**sb) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ assert!(err.is_null());
+
+ let pc_result = paimon_table_write_prepare_commit(tw);
+ assert!(pc_result.error.is_null());
+
+ let tc_result = paimon_write_builder_new_commit(wb);
+ let tc = tc_result.commit;
+ let err = paimon_table_commit_abort(tc, pc_result.messages);
+ assert!(err.is_null());
+ paimon_commit_messages_free(pc_result.messages);
+
+ let rows = read_rows_ffi(handle);
+ assert!(rows.is_empty());
+
+ paimon_table_commit_free(tc);
+ paimon_table_write_free(tw);
+ paimon_write_builder_free(wb);
+ unwrap_table(handle);
+ }
+}
+
+#[test]
+fn test_null_pointer_handling() {
+ unsafe {
+ let result = paimon_table_new_write_builder(ptr::null());
+ assert!(!result.error.is_null());
+ assert!(result.write_builder.is_null());
+ paimon_error_free(result.error);
+
+ let result = paimon_write_builder_new_write(ptr::null());
+ assert!(!result.error.is_null());
+ assert!(result.write.is_null());
+ paimon_error_free(result.error);
+
+ let result = paimon_write_builder_new_commit(ptr::null());
+ assert!(!result.error.is_null());
+ assert!(result.commit.is_null());
+ paimon_error_free(result.error);
+
+ let err =
+ paimon_table_write_write_arrow_batch(ptr::null_mut(),
ptr::null_mut(), ptr::null_mut());
+ assert!(!err.is_null());
+ paimon_error_free(err);
+
+ let result = paimon_table_write_prepare_commit(ptr::null_mut());
+ assert!(!result.error.is_null());
+ assert!(result.messages.is_null());
+ paimon_error_free(result.error);
+
+ let err = paimon_table_commit_commit(ptr::null(), ptr::null_mut());
+ assert!(!err.is_null());
+ paimon_error_free(err);
+
+ paimon_write_builder_free(ptr::null_mut());
+ paimon_table_write_free(ptr::null_mut());
+ paimon_table_commit_free(ptr::null_mut());
+ paimon_commit_messages_free(ptr::null_mut());
+ }
+}
+
+#[test]
+fn test_two_commits_same_builder() {
+ let path = "memory:/test_two_commits";
+ let file_io = memory_file_io();
+ setup_table_dirs(&file_io, path);
+ let table = Table::new(
+ file_io,
+ Identifier::new("default", "test"),
+ path.to_string(),
+ simple_table_schema(),
+ None,
+ );
+ let handle = unsafe { wrap_table(table) };
+
+ unsafe {
+ let wb_result = paimon_table_new_write_builder(handle);
+ let wb = wb_result.write_builder;
+
+ // First commit
+ {
+ let tw_result = paimon_write_builder_new_write(wb);
+ let tw = tw_result.write;
+ let batch = make_batch(vec![1], vec!["a"]);
+ let (ab, sb) = export_batch_to_ffi(batch);
+ paimon_table_write_write_arrow_batch(
+ tw,
+ (&**ab) as *const FFI_ArrowArray as *mut c_void,
+ (&**sb) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ let pc = paimon_table_write_prepare_commit(tw);
+ let tc_result = paimon_write_builder_new_commit(wb);
+ paimon_table_commit_commit(tc_result.commit, pc.messages);
+ paimon_commit_messages_free(pc.messages);
+ paimon_table_commit_free(tc_result.commit);
+ paimon_table_write_free(tw);
+ }
+
+ // Second commit with same builder
+ {
+ let tw_result = paimon_write_builder_new_write(wb);
+ let tw = tw_result.write;
+ let batch = make_batch(vec![2], vec!["b"]);
+ let (ab, sb) = export_batch_to_ffi(batch);
+ paimon_table_write_write_arrow_batch(
+ tw,
+ (&**ab) as *const FFI_ArrowArray as *mut c_void,
+ (&**sb) as *const FFI_ArrowSchema as *mut c_void,
+ );
+ let pc = paimon_table_write_prepare_commit(tw);
+ let tc_result = paimon_write_builder_new_commit(wb);
+ paimon_table_commit_commit(tc_result.commit, pc.messages);
+ paimon_commit_messages_free(pc.messages);
+ paimon_table_commit_free(tc_result.commit);
+ paimon_table_write_free(tw);
+ }
+
+ paimon_write_builder_free(wb);
+
+ let rows = read_rows_ffi(handle);
+ assert_eq!(rows, vec![(1, "a".into()), (2, "b".into())]);
+
+ unwrap_table(handle);
+ }
+}
diff --git a/bindings/c/src/types.rs b/bindings/c/src/types.rs
index c7b059a..8d3a466 100644
--- a/bindings/c/src/types.rs
+++ b/bindings/c/src/types.rs
@@ -16,9 +16,11 @@
// under the License.
use std::ffi::c_void;
+use std::sync::Arc;
+use arrow_schema::Schema as ArrowSchema;
use paimon::spec::{DataField, Predicate};
-use paimon::table::Table;
+use paimon::table::{CommitMessage, Table, TableCommit, TableWrite};
/// C-compatible key-value pair for options.
#[repr(C)]
@@ -155,6 +157,7 @@ pub struct paimon_predicate {
/// - `Timestamp`/`LocalZonedTimestamp` → `int_val` (millis) + `int_val2`
(nanos)
/// - `Decimal` → `int_val` + `int_val2` (unscaled i128) + `uint_val`
(precision) + `uint_val2` (scale)
#[repr(C)]
+#[derive(Default)]
pub struct paimon_datum {
pub tag: i32,
pub int_val: i64,
@@ -178,3 +181,52 @@ pub struct paimon_arrow_batch {
/// Pointer to a heap-allocated ArrowSchema.
pub schema: *mut c_void,
}
+
+// === Write/Commit opaque types ===
+
+/// Internal state for WriteBuilder that stores table, shared commit_user, and
overwrite flag.
+pub(crate) struct WriteBuilderState {
+ pub table: Table,
+ pub commit_user: String,
+ pub overwrite: bool,
+}
+
+pub(crate) struct TableWriteState {
+ pub write: TableWrite,
+ pub target_schema: Arc<ArrowSchema>,
+ pub table_location: String,
+ pub commit_user: String,
+}
+
+pub(crate) struct TableCommitState {
+ pub commit: TableCommit,
+ pub table_location: String,
+ pub commit_user: String,
+}
+
+pub(crate) struct CommitMessagesState {
+ pub messages: Vec<CommitMessage>,
+ pub table_location: String,
+ pub commit_user: String,
+}
+
+#[repr(C)]
+pub struct paimon_write_builder {
+ pub inner: *mut c_void,
+}
+
+#[repr(C)]
+pub struct paimon_table_write {
+ pub inner: *mut c_void,
+}
+
+#[repr(C)]
+pub struct paimon_table_commit {
+ pub inner: *mut c_void,
+}
+
+/// Opaque container for commit messages and their originating write context.
+#[repr(C)]
+pub struct paimon_commit_messages {
+ pub inner: *mut c_void,
+}
diff --git a/bindings/c/src/write.rs b/bindings/c/src/write.rs
new file mode 100644
index 0000000..7579234
--- /dev/null
+++ b/bindings/c/src/write.rs
@@ -0,0 +1,830 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::ffi::{c_char, c_void};
+use std::ptr;
+use std::sync::Arc;
+
+use arrow_array::ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema};
+use arrow_array::{Array, RecordBatch, RecordBatchOptions, StructArray};
+use arrow_schema::{DataType as ArrowDataType, Schema as ArrowSchema};
+use paimon::table::Table;
+
+use crate::error::{check_non_null, paimon_error, validate_cstr,
PaimonErrorCode};
+use crate::result::{
+ paimon_result_prepare_commit, paimon_result_table_commit,
paimon_result_table_write,
+ paimon_result_write_builder,
+};
+use crate::runtime;
+use crate::types::*;
+
+// ======================= WriteBuilder ===============================
+
+unsafe fn new_write_builder(
+ table: *const paimon_table,
+ commit_user: Option<String>,
+) -> paimon_result_write_builder {
+ if let Err(e) = check_non_null(table, "table") {
+ return paimon_result_write_builder {
+ write_builder: ptr::null_mut(),
+ error: e,
+ };
+ }
+ let table_ref = &*((*table).inner as *const Table);
+ let builder = table_ref.new_write_builder();
+ let commit_user = match commit_user {
+ Some(commit_user) => match builder.with_commit_user(commit_user) {
+ Ok(builder) => builder.commit_user().to_string(),
+ Err(e) => {
+ return paimon_result_write_builder {
+ write_builder: ptr::null_mut(),
+ error: paimon_error::from_paimon(e),
+ }
+ }
+ },
+ None => builder.commit_user().to_string(),
+ };
+ let state = WriteBuilderState {
+ table: table_ref.clone(),
+ commit_user,
+ overwrite: false,
+ };
+ let inner = Box::into_raw(Box::new(state)) as *mut c_void;
+ paimon_result_write_builder {
+ write_builder: Box::into_raw(Box::new(paimon_write_builder { inner })),
+ error: ptr::null_mut(),
+ }
+}
+
+/// Create a new WriteBuilder from a Table.
+///
+/// The returned WriteBuilder holds a shared `commit_user` (UUID) that will be
+/// used by both `new_write()` and `new_commit()` for duplicate-commit
detection.
+///
+/// # Safety
+/// `table` must be a valid pointer from `paimon_catalog_get_table`, or null
(returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_new_write_builder(
+ table: *const paimon_table,
+) -> paimon_result_write_builder {
+ new_write_builder(table, None)
+}
+
+/// Create a WriteBuilder with a caller-provided stable commit identity.
+///
+/// Writers whose messages are merged into one logical commit must use the
+/// same `commit_user`.
+///
+/// # Safety
+/// `table` must be a valid table pointer. `commit_user` must be a valid UTF-8
+/// C string and a safe file-name segment.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_new_write_builder_with_commit_user(
+ table: *const paimon_table,
+ commit_user: *const c_char,
+) -> paimon_result_write_builder {
+ let commit_user = match validate_cstr(commit_user, "commit_user") {
+ Ok(commit_user) => commit_user,
+ Err(error) => {
+ return paimon_result_write_builder {
+ write_builder: ptr::null_mut(),
+ error,
+ }
+ }
+ };
+ new_write_builder(table, Some(commit_user))
+}
+
+/// Free a paimon_write_builder.
+///
+/// # Safety
+/// Only call with a write_builder returned from
`paimon_table_new_write_builder`.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_write_builder_free(wb: *mut
paimon_write_builder) {
+ if !wb.is_null() {
+ let wrapper = Box::from_raw(wb);
+ if !wrapper.inner.is_null() {
+ drop(Box::from_raw(wrapper.inner as *mut WriteBuilderState));
+ }
+ }
+}
+
+/// Enable overwrite mode for the WriteBuilder.
+///
+/// In overwrite mode, a subsequent `paimon_table_commit_overwrite` will
replace
+/// the data in the written partitions rather than appending.
+///
+/// # Safety
+/// `wb` must be a valid pointer from `paimon_table_new_write_builder`, or
null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_write_builder_with_overwrite(
+ wb: *mut paimon_write_builder,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(wb, "wb") {
+ return e;
+ }
+ let state = &mut *((*wb).inner as *mut WriteBuilderState);
+ state.overwrite = true;
+ ptr::null_mut()
+}
+
+// ======================= TableWrite ===============================
+
+fn invalid_input(message: impl Into<String>) -> *mut paimon_error {
+ paimon_error::new(PaimonErrorCode::InvalidInput, message.into())
+}
+
+fn validate_batch_schema(
+ input: &RecordBatch,
+ target: &ArrowSchema,
+) -> Result<(), *mut paimon_error> {
+ let input_schema = input.schema();
+ if input_schema.fields().len() != target.fields().len() {
+ return Err(invalid_input(format!(
+ "Input schema is not consistent with the table schema. input:
{input_schema:?}, table: {target:?}"
+ )));
+ }
+ for (index, (input_field, target_field)) in input_schema
+ .fields()
+ .iter()
+ .zip(target.fields().iter())
+ .enumerate()
+ {
+ if input_field.name() != target_field.name()
+ || input_field.data_type() != target_field.data_type()
+ {
+ return Err(invalid_input(format!(
+ "Input schema is not consistent with the table schema. input:
{input_schema:?}, table: {target:?}"
+ )));
+ }
+ if !target_field.is_nullable() && input.column(index).null_count() !=
0 {
+ return Err(invalid_input(format!(
+ "Column '{}' is NOT NULL but the Arrow batch contains {} null
value(s)",
+ target_field.name(),
+ input.column(index).null_count()
+ )));
+ }
+ }
+ Ok(())
+}
+
+unsafe fn import_record_batch(
+ array: *mut c_void,
+ schema: *mut c_void,
+) -> Result<RecordBatch, *mut paimon_error> {
+ // Arrow's from_raw implements the C Data Interface move operation: it
+ // replaces the caller-owned struct with an empty/released value.
+ let ffi_array = FFI_ArrowArray::from_raw(array as *mut FFI_ArrowArray);
+ let ffi_schema = FFI_ArrowSchema::from_raw(schema as *mut FFI_ArrowSchema);
+ let data = match from_ffi(ffi_array, &ffi_schema) {
+ Ok(data) => data,
+ Err(e) => {
+ drop(ffi_schema);
+ return Err(invalid_input(format!(
+ "Failed to import Arrow record batch: {e}"
+ )));
+ }
+ };
+ drop(ffi_schema);
+
+ if !matches!(data.data_type(), ArrowDataType::Struct(_)) {
+ return Err(invalid_input(format!(
+ "Arrow record batch root must be Struct, got {:?}",
+ data.data_type()
+ )));
+ }
+
+ let struct_array = StructArray::from(data);
+ if struct_array.null_count() != 0 {
+ return Err(invalid_input(
+ "Arrow record batch root Struct must not contain nulls",
+ ));
+ }
+
+ let row_count = struct_array.len();
+ let (fields, columns, _) = struct_array.into_parts();
+ let schema = Arc::new(ArrowSchema::new(fields));
+ RecordBatch::try_new_with_options(
+ schema,
+ columns,
+ &RecordBatchOptions::new().with_row_count(Some(row_count)),
+ )
+ .map_err(|e| invalid_input(format!("Failed to construct Arrow record
batch: {e}")))
+}
+
+/// Create a new TableWrite from the WriteBuilder.
+///
+/// The returned TableWrite accumulates Arrow batches until `prepare_commit`
is called.
+///
+/// # Safety
+/// `wb` must be a valid pointer from `paimon_table_new_write_builder`, or
null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_write_builder_new_write(
+ wb: *const paimon_write_builder,
+) -> paimon_result_table_write {
+ if let Err(e) = check_non_null(wb, "wb") {
+ return paimon_result_table_write {
+ write: ptr::null_mut(),
+ error: e,
+ };
+ }
+ let state = &*((*wb).inner as *const WriteBuilderState);
+
+ let mut builder = match state
+ .table
+ .new_write_builder()
+ .with_commit_user(state.commit_user.clone())
+ {
+ Ok(b) => b,
+ Err(e) => {
+ return paimon_result_table_write {
+ write: ptr::null_mut(),
+ error: paimon_error::from_paimon(e),
+ }
+ }
+ };
+
+ if state.overwrite {
+ builder = builder.with_overwrite();
+ }
+
+ let tw = match builder.new_write() {
+ Ok(w) => w,
+ Err(e) => {
+ return paimon_result_table_write {
+ write: ptr::null_mut(),
+ error: paimon_error::from_paimon(e),
+ }
+ }
+ };
+
+ let target_schema =
+ match
paimon::arrow::build_target_arrow_schema(state.table.schema().fields()) {
+ Ok(schema) => schema,
+ Err(e) => {
+ return paimon_result_table_write {
+ write: ptr::null_mut(),
+ error: paimon_error::from_paimon(e),
+ }
+ }
+ };
+ let table_write = TableWriteState {
+ write: tw,
+ target_schema,
+ table_location: state.table.location().to_string(),
+ commit_user: state.commit_user.clone(),
+ };
+ let inner = Box::into_raw(Box::new(table_write)) as *mut c_void;
+ paimon_result_table_write {
+ write: Box::into_raw(Box::new(paimon_table_write { inner })),
+ error: ptr::null_mut(),
+ }
+}
+
+/// Free a paimon_table_write.
+///
+/// Dropping a TableWrite before calling `prepare_commit` discards any
+/// uncommitted data.
+///
+/// # Safety
+/// Only call with a write returned from `paimon_write_builder_new_write`.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_write_free(tw: *mut paimon_table_write) {
+ if !tw.is_null() {
+ let wrapper = Box::from_raw(tw);
+ if !wrapper.inner.is_null() {
+ drop(Box::from_raw(wrapper.inner as *mut TableWriteState));
+ }
+ }
+}
+
+/// Write a single Arrow record batch into the table's writers.
+///
+/// The Arrow data is imported via the Arrow C Data Interface. `array` and
+/// `schema` must point to valid `ArrowArray` and `ArrowSchema` structs
+/// filled by the caller. Ownership is transferred — the caller must not
+/// release the structs after this call.
+///
+/// # Safety
+/// `tw` must be a valid pointer from `paimon_write_builder_new_write`, or
null (returns error).
+/// `array` and `schema` must be valid pointers to initialized ArrowArray /
+/// ArrowSchema structs, or null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_write_write_arrow_batch(
+ tw: *mut paimon_table_write,
+ array: *mut c_void,
+ schema: *mut c_void,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(tw, "tw") {
+ return e;
+ }
+ if let Err(e) = check_non_null(array, "array") {
+ return e;
+ }
+ if let Err(e) = check_non_null(schema, "schema") {
+ return e;
+ }
+
+ let table_write = &mut *((*tw).inner as *mut TableWriteState);
+ let batch = match import_record_batch(array, schema) {
+ Ok(batch) => batch,
+ Err(error) => return error,
+ };
+ if let Err(error) = validate_batch_schema(&batch,
&table_write.target_schema) {
+ return error;
+ }
+
+ match runtime().block_on(table_write.write.write_arrow_batch(&batch)) {
+ Ok(()) => ptr::null_mut(),
+ Err(e) => paimon_error::from_paimon(e),
+ }
+}
+
+/// Close file writers and produce CommitMessages.
+///
+/// Consumes the open file writers (they are flushed and closed). After this
+/// call, the TableWrite can be reused — `write_arrow_batch` may be called
+/// again to start a new round of writes.
+///
+/// The returned `paimon_commit_messages` must be passed to a
+/// `paimon_table_commit_*` function and then freed with
+/// `paimon_commit_messages_free`.
+///
+/// # Safety
+/// `tw` must be a valid pointer from `paimon_write_builder_new_write`, or
null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_write_prepare_commit(
+ tw: *mut paimon_table_write,
+) -> paimon_result_prepare_commit {
+ if let Err(e) = check_non_null(tw, "tw") {
+ return paimon_result_prepare_commit {
+ messages: ptr::null_mut(),
+ error: e,
+ };
+ }
+ let table_write = &mut *((*tw).inner as *mut TableWriteState);
+
+ match runtime().block_on(table_write.write.prepare_commit()) {
+ Ok(messages) => {
+ let messages = CommitMessagesState {
+ messages,
+ table_location: table_write.table_location.clone(),
+ commit_user: table_write.commit_user.clone(),
+ };
+ let inner = Box::into_raw(Box::new(messages)) as *mut c_void;
+ paimon_result_prepare_commit {
+ messages: Box::into_raw(Box::new(paimon_commit_messages {
inner })),
+ error: ptr::null_mut(),
+ }
+ }
+ Err(e) => paimon_result_prepare_commit {
+ messages: ptr::null_mut(),
+ error: paimon_error::from_paimon(e),
+ },
+ }
+}
+
+// ======================= TableCommit ===============================
+
+/// Create a new TableCommit from the WriteBuilder.
+///
+/// The committer shares the same `commit_user` as the writer, which is
+/// required for duplicate-commit detection.
+///
+/// # Safety
+/// `wb` must be a valid pointer from `paimon_table_new_write_builder`, or
null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_write_builder_new_commit(
+ wb: *const paimon_write_builder,
+) -> paimon_result_table_commit {
+ if let Err(e) = check_non_null(wb, "wb") {
+ return paimon_result_table_commit {
+ commit: ptr::null_mut(),
+ error: e,
+ };
+ }
+ let state = &*((*wb).inner as *const WriteBuilderState);
+
+ let builder = match state
+ .table
+ .new_write_builder()
+ .with_commit_user(state.commit_user.clone())
+ {
+ Ok(b) => b,
+ Err(e) => {
+ return paimon_result_table_commit {
+ commit: ptr::null_mut(),
+ error: paimon_error::from_paimon(e),
+ }
+ }
+ };
+
+ let tc = match builder.try_new_commit() {
+ Ok(c) => c,
+ Err(e) => {
+ return paimon_result_table_commit {
+ commit: ptr::null_mut(),
+ error: paimon_error::from_paimon(e),
+ }
+ }
+ };
+
+ let table_commit = TableCommitState {
+ commit: tc,
+ table_location: state.table.location().to_string(),
+ commit_user: state.commit_user.clone(),
+ };
+ let inner = Box::into_raw(Box::new(table_commit)) as *mut c_void;
+ paimon_result_table_commit {
+ commit: Box::into_raw(Box::new(paimon_table_commit { inner })),
+ error: ptr::null_mut(),
+ }
+}
+
+/// Free a paimon_table_commit.
+///
+/// # Safety
+/// Only call with a commit returned from `paimon_write_builder_new_commit`.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_commit_free(tc: *mut
paimon_table_commit) {
+ if !tc.is_null() {
+ let wrapper = Box::from_raw(tc);
+ if !wrapper.inner.is_null() {
+ drop(Box::from_raw(wrapper.inner as *mut TableCommitState));
+ }
+ }
+}
+
+// ======================= CommitMessages ===============================
+
+/// Free a paimon_commit_messages.
+///
+/// # Safety
+/// Only call with messages returned from `paimon_table_write_prepare_commit`.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_commit_messages_free(msgs: *mut
paimon_commit_messages) {
+ if !msgs.is_null() {
+ let wrapper = Box::from_raw(msgs);
+ if !wrapper.inner.is_null() {
+ drop(Box::from_raw(wrapper.inner as *mut CommitMessagesState));
+ }
+ }
+}
+
+/// Merge `source` messages into `target` for one logical commit.
+///
+/// Both handles retain ownership and must be freed separately. They must have
+/// been prepared for the same table and `commit_user`.
+///
+/// # Safety
+/// `target` and `source` must be distinct valid commit-message handles.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_commit_messages_merge(
+ target: *mut paimon_commit_messages,
+ source: *const paimon_commit_messages,
+) -> *mut paimon_error {
+ if let Err(error) = check_non_null(target, "target") {
+ return error;
+ }
+ if let Err(error) = check_non_null(source, "source") {
+ return error;
+ }
+ if ptr::eq(target, source.cast_mut()) {
+ return invalid_input("target and source commit messages must be
distinct handles");
+ }
+
+ let target = &mut *((*target).inner as *mut CommitMessagesState);
+ let source = &*((*source).inner as *const CommitMessagesState);
+ if target.table_location != source.table_location || target.commit_user !=
source.commit_user {
+ return invalid_input(
+ "commit messages can only be merged when table and commit_user
both match",
+ );
+ }
+ target.messages.extend(source.messages.clone());
+ ptr::null_mut()
+}
+
+// ======================= Commit operations ===============================
+
+fn validate_commit_context(
+ commit: &TableCommitState,
+ messages: &CommitMessagesState,
+) -> Result<(), *mut paimon_error> {
+ if commit.table_location != messages.table_location {
+ return Err(invalid_input(format!(
+ "commit messages were prepared for a different table (message
table '{}', committer table '{}')",
+ messages.table_location, commit.table_location
+ )));
+ }
+ if commit.commit_user != messages.commit_user {
+ return Err(invalid_input(
+ "commit messages were prepared with a different commit_user",
+ ));
+ }
+ Ok(())
+}
+
+/// Commit the given messages in APPEND mode.
+///
+/// Empty messages is a no-op success.
+/// The caller retains ownership of `msgs`; it may retry after an error and
+/// must eventually release the handle with `paimon_commit_messages_free`.
+///
+/// # Safety
+/// `tc` must be a valid pointer from `paimon_write_builder_new_commit`, or
null (returns error).
+/// `msgs` must be a valid pointer from `paimon_table_write_prepare_commit`,
or null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_commit_commit(
+ tc: *const paimon_table_commit,
+ msgs: *mut paimon_commit_messages,
+) -> *mut paimon_error {
+ paimon_table_commit_commit_with_identifier(tc, msgs, i64::MAX)
+}
+
+/// Commit the given messages with a caller-provided identifier.
+///
+/// Identifiers must increase monotonically for a `commit_user`. All messages
+/// for one identifier must be merged and submitted in a single call.
+/// This operation does not filter previously committed identifiers. Use
+/// `paimon_table_commit_filter_and_commit_with_identifier` when retrying an
+/// uncertain result.
+/// The caller retains ownership of `msgs` and must free it explicitly.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_commit_commit_with_identifier(
+ tc: *const paimon_table_commit,
+ msgs: *mut paimon_commit_messages,
+ commit_identifier: i64,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(tc, "tc") {
+ return e;
+ }
+ if let Err(e) = check_non_null(msgs, "msgs") {
+ return e;
+ }
+
+ let table_commit = &*((*tc).inner as *const TableCommitState);
+ let messages = &*((*msgs).inner as *const CommitMessagesState);
+ if let Err(error) = validate_commit_context(table_commit, messages) {
+ return error;
+ }
+
+ match runtime().block_on(
+ table_commit
+ .commit
+ .commit_with_identifier(messages.messages.clone(),
commit_identifier),
+ ) {
+ Ok(()) => ptr::null_mut(),
+ Err(e) => paimon_error::from_paimon(e),
+ }
+}
+
+/// Filter a previously committed identifier, then commit if it is new.
+///
+/// Identifiers must increase monotonically for a `commit_user`. Use this only
+/// to retry the same uncertain result after all writer messages for the
+/// logical commit have been merged.
+/// The caller retains ownership of `msgs` and must free it explicitly.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_commit_filter_and_commit_with_identifier(
+ tc: *const paimon_table_commit,
+ msgs: *mut paimon_commit_messages,
+ commit_identifier: i64,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(tc, "tc") {
+ return e;
+ }
+ if let Err(e) = check_non_null(msgs, "msgs") {
+ return e;
+ }
+
+ let table_commit = &*((*tc).inner as *const TableCommitState);
+ let messages = &*((*msgs).inner as *const CommitMessagesState);
+ if let Err(error) = validate_commit_context(table_commit, messages) {
+ return error;
+ }
+
+ match runtime().block_on(
+ table_commit
+ .commit
+ .filter_and_commit_with_identifier(messages.messages.clone(),
commit_identifier),
+ ) {
+ Ok(()) => ptr::null_mut(),
+ Err(e) => paimon_error::from_paimon(e),
+ }
+}
+
+/// Commit in OVERWRITE mode, replacing data in the written partitions.
+///
+/// `static_partitions` is currently passed as `None` (overwrite all
+/// partitions that were written to).
+/// The caller retains ownership of `msgs`; it may retry after an error and
+/// must eventually release the handle with `paimon_commit_messages_free`.
+///
+/// # Safety
+/// `tc` must be a valid pointer from `paimon_write_builder_new_commit`, or
null (returns error).
+/// `msgs` must be a valid pointer from `paimon_table_write_prepare_commit`,
or null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_commit_overwrite(
+ tc: *const paimon_table_commit,
+ msgs: *mut paimon_commit_messages,
+) -> *mut paimon_error {
+ paimon_table_commit_overwrite_impl(tc, msgs, None)
+}
+
+/// Overwrite with a caller-provided stable commit identifier.
+///
+/// The caller retains ownership of `msgs` and must free it explicitly.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_commit_overwrite_with_identifier(
+ tc: *const paimon_table_commit,
+ msgs: *mut paimon_commit_messages,
+ commit_identifier: i64,
+) -> *mut paimon_error {
+ paimon_table_commit_overwrite_impl(tc, msgs, Some(commit_identifier))
+}
+
+unsafe fn paimon_table_commit_overwrite_impl(
+ tc: *const paimon_table_commit,
+ msgs: *mut paimon_commit_messages,
+ commit_identifier: Option<i64>,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(tc, "tc") {
+ return e;
+ }
+ if let Err(e) = check_non_null(msgs, "msgs") {
+ return e;
+ }
+
+ let table_commit = &*((*tc).inner as *const TableCommitState);
+ let messages = &*((*msgs).inner as *const CommitMessagesState);
+ if let Err(error) = validate_commit_context(table_commit, messages) {
+ return error;
+ }
+
+ let result = match commit_identifier {
+ Some(commit_identifier) => {
+ runtime().block_on(table_commit.commit.overwrite_with_identifier(
+ messages.messages.clone(),
+ None,
+ commit_identifier,
+ ))
+ }
+ None => runtime().block_on(
+ table_commit
+ .commit
+ .overwrite(messages.messages.clone(), None),
+ ),
+ };
+ match result {
+ Ok(()) => ptr::null_mut(),
+ Err(e) => paimon_error::from_paimon(e),
+ }
+}
+
+/// Truncate the entire table — removes all data.
+///
+/// This is an OVERWRITE with zero new files. The table's latest snapshot
+/// will have no data.
+///
+/// # Safety
+/// `tc` must be a valid pointer from `paimon_write_builder_new_commit`, or
null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_commit_truncate_table(
+ tc: *const paimon_table_commit,
+) -> *mut paimon_error {
+ paimon_table_commit_truncate_table_impl(tc, None)
+}
+
+/// Truncate the table with a caller-provided stable commit identifier.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_commit_truncate_table_with_identifier(
+ tc: *const paimon_table_commit,
+ commit_identifier: i64,
+) -> *mut paimon_error {
+ paimon_table_commit_truncate_table_impl(tc, Some(commit_identifier))
+}
+
+unsafe fn paimon_table_commit_truncate_table_impl(
+ tc: *const paimon_table_commit,
+ commit_identifier: Option<i64>,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(tc, "tc") {
+ return e;
+ }
+
+ let table_commit = &*((*tc).inner as *const TableCommitState);
+
+ let result = match commit_identifier {
+ Some(commit_identifier) => runtime().block_on(
+ table_commit
+ .commit
+ .truncate_table_with_identifier(commit_identifier),
+ ),
+ None => runtime().block_on(table_commit.commit.truncate_table()),
+ };
+ match result {
+ Ok(()) => ptr::null_mut(),
+ Err(e) => paimon_error::from_paimon(e),
+ }
+}
+
+/// Abort a prepared commit, cleaning up written data files.
+///
+/// This is a best-effort cleanup — it attempts to delete new data, changelog,
+/// and index files produced by the writer. Storage deletion errors are ignored
+/// so cleanup does not mask an earlier write or commit failure.
+/// The caller retains ownership of `msgs` and must free it explicitly.
+///
+/// # Safety
+/// `tc` must be a valid pointer from `paimon_write_builder_new_commit`, or
null (returns error).
+/// `msgs` must be a valid pointer from `paimon_table_write_prepare_commit`,
or null (returns error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_commit_abort(
+ tc: *const paimon_table_commit,
+ msgs: *mut paimon_commit_messages,
+) -> *mut paimon_error {
+ if let Err(e) = check_non_null(tc, "tc") {
+ return e;
+ }
+ if let Err(e) = check_non_null(msgs, "msgs") {
+ return e;
+ }
+
+ let table_commit = &*((*tc).inner as *const TableCommitState);
+ let messages = &*((*msgs).inner as *const CommitMessagesState);
+ if let Err(error) = validate_commit_context(table_commit, messages) {
+ return error;
+ }
+
+ match runtime().block_on(table_commit.commit.abort(&messages.messages)) {
+ Ok(()) => ptr::null_mut(),
+ Err(e) => paimon_error::from_paimon(e),
+ }
+}
+
+// --- C ABI signature guards -------------------------------------------------
+
+const _: unsafe extern "C" fn(*const paimon_table) ->
paimon_result_write_builder =
+ paimon_table_new_write_builder;
+const _: unsafe extern "C" fn(*const paimon_table, *const c_char) ->
paimon_result_write_builder =
+ paimon_table_new_write_builder_with_commit_user;
+const _: unsafe extern "C" fn(*const paimon_write_builder) ->
paimon_result_table_write =
+ paimon_write_builder_new_write;
+const _: unsafe extern "C" fn(*const paimon_write_builder) ->
paimon_result_table_commit =
+ paimon_write_builder_new_commit;
+const _: unsafe extern "C" fn(*mut paimon_table_write) ->
paimon_result_prepare_commit =
+ paimon_table_write_prepare_commit;
+const _: unsafe extern "C" fn(
+ *mut paimon_commit_messages,
+ *const paimon_commit_messages,
+) -> *mut paimon_error = paimon_commit_messages_merge;
+const _: unsafe extern "C" fn(
+ *mut paimon_table_write,
+ *mut c_void,
+ *mut c_void,
+) -> *mut paimon_error = paimon_table_write_write_arrow_batch;
+const _: unsafe extern "C" fn(
+ *const paimon_table_commit,
+ *mut paimon_commit_messages,
+) -> *mut paimon_error = paimon_table_commit_commit;
+const _: unsafe extern "C" fn(
+ *const paimon_table_commit,
+ *mut paimon_commit_messages,
+ i64,
+) -> *mut paimon_error = paimon_table_commit_commit_with_identifier;
+const _: unsafe extern "C" fn(
+ *const paimon_table_commit,
+ *mut paimon_commit_messages,
+ i64,
+) -> *mut paimon_error = paimon_table_commit_filter_and_commit_with_identifier;
+const _: unsafe extern "C" fn(
+ *const paimon_table_commit,
+ *mut paimon_commit_messages,
+) -> *mut paimon_error = paimon_table_commit_overwrite;
+const _: unsafe extern "C" fn(
+ *const paimon_table_commit,
+ *mut paimon_commit_messages,
+ i64,
+) -> *mut paimon_error = paimon_table_commit_overwrite_with_identifier;
+const _: unsafe extern "C" fn(*const paimon_table_commit) -> *mut paimon_error
=
+ paimon_table_commit_truncate_table;
+const _: unsafe extern "C" fn(*const paimon_table_commit, i64) -> *mut
paimon_error =
+ paimon_table_commit_truncate_table_with_identifier;
+const _: unsafe extern "C" fn(
+ *const paimon_table_commit,
+ *mut paimon_commit_messages,
+) -> *mut paimon_error = paimon_table_commit_abort;
diff --git a/crates/paimon/src/table/table_commit.rs
b/crates/paimon/src/table/table_commit.rs
index 2a81047..809e45d 100644
--- a/crates/paimon/src/table/table_commit.rs
+++ b/crates/paimon/src/table/table_commit.rs
@@ -121,12 +121,39 @@ impl TableCommit {
/// Commit new files with a caller-provided commit identifier.
///
- /// The identifier participates in retry idempotency, matching Python
- /// `FileStoreCommit.commit(commit_messages, commit_identifier)`.
+ /// Identifiers must increase monotonically for a given `commit_user`.
+ /// All messages for one identifier must be submitted in a single call.
+ /// This method does not filter previously committed identifiers. Use
+ /// [`Self::filter_and_commit_with_identifier`] when retrying an uncertain
+ /// commit result.
pub async fn commit_with_identifier(
&self,
commit_messages: Vec<CommitMessage>,
commit_identifier: i64,
+ ) -> Result<()> {
+ self.commit_with_identifier_impl(commit_messages, commit_identifier,
false)
+ .await
+ }
+
+ /// Filter a previously committed identifier, then commit if it is new.
+ ///
+ /// Identifiers must increase monotonically for a given `commit_user`. This
+ /// method is intended for retrying the same uncertain result; regular
+ /// commits should use [`Self::commit_with_identifier`].
+ pub async fn filter_and_commit_with_identifier(
+ &self,
+ commit_messages: Vec<CommitMessage>,
+ commit_identifier: i64,
+ ) -> Result<()> {
+ self.commit_with_identifier_impl(commit_messages, commit_identifier,
true)
+ .await
+ }
+
+ async fn commit_with_identifier_impl(
+ &self,
+ commit_messages: Vec<CommitMessage>,
+ commit_identifier: i64,
+ filter_committed: bool,
) -> Result<()> {
self.table.ensure_not_branch_reference_for_write()?;
@@ -147,6 +174,7 @@ impl TableCommit {
},
None,
commit_identifier,
+ filter_committed,
)
.await
}
@@ -189,6 +217,7 @@ impl TableCommit {
},
Some(expected_snapshot_id),
commit_identifier,
+ false,
)
.await
}
@@ -210,16 +239,35 @@ impl TableCommit {
commit_messages: Vec<CommitMessage>,
static_partitions: Option<HashMap<String, Option<Datum>>>,
) -> Result<()> {
- self.overwrite_with_identifier(commit_messages, static_partitions,
BATCH_COMMIT_IDENTIFIER)
- .await
+ self.overwrite_impl(
+ commit_messages,
+ static_partitions,
+ BATCH_COMMIT_IDENTIFIER,
+ false,
+ )
+ .await
}
/// Overwrite partitions with a caller-provided commit identifier.
+ ///
+ /// A previously committed identifier is filtered so retrying an uncertain
+ /// result cannot re-execute the destructive operation.
pub async fn overwrite_with_identifier(
&self,
commit_messages: Vec<CommitMessage>,
static_partitions: Option<HashMap<String, Option<Datum>>>,
commit_identifier: i64,
+ ) -> Result<()> {
+ self.overwrite_impl(commit_messages, static_partitions,
commit_identifier, true)
+ .await
+ }
+
+ async fn overwrite_impl(
+ &self,
+ commit_messages: Vec<CommitMessage>,
+ static_partitions: Option<HashMap<String, Option<Datum>>>,
+ commit_identifier: i64,
+ filter_committed: bool,
) -> Result<()> {
self.table.ensure_not_branch_reference_for_write()?;
@@ -262,6 +310,7 @@ impl TableCommit {
},
None,
commit_identifier,
+ filter_committed,
)
.await
}
@@ -446,15 +495,28 @@ impl TableCommit {
&self,
partitions: Vec<HashMap<String, Option<Datum>>>,
) -> Result<()> {
- self.truncate_partitions_with_identifier(partitions,
BATCH_COMMIT_IDENTIFIER)
+ self.truncate_partitions_impl(partitions, BATCH_COMMIT_IDENTIFIER,
false)
.await
}
/// Drop specific partitions with a caller-provided commit identifier.
+ ///
+ /// A previously committed identifier is filtered so retrying an uncertain
+ /// result cannot delete data committed in between.
pub async fn truncate_partitions_with_identifier(
&self,
partitions: Vec<HashMap<String, Option<Datum>>>,
commit_identifier: i64,
+ ) -> Result<()> {
+ self.truncate_partitions_impl(partitions, commit_identifier, true)
+ .await
+ }
+
+ async fn truncate_partitions_impl(
+ &self,
+ partitions: Vec<HashMap<String, Option<Datum>>>,
+ commit_identifier: i64,
+ filter_committed: bool,
) -> Result<()> {
self.table.ensure_not_branch_reference_for_write()?;
@@ -476,6 +538,7 @@ impl TableCommit {
},
None,
commit_identifier,
+ filter_committed,
)
.await
}
@@ -485,8 +548,13 @@ impl TableCommit {
&self,
partitions: Vec<HashMap<String, Option<Datum>>>,
) -> Result<()> {
- self.drop_partitions_with_identifier(partitions,
BATCH_COMMIT_IDENTIFIER)
- .await
+ if partitions.is_empty() {
+ return Err(crate::Error::DataInvalid {
+ message: "Partitions list cannot be empty.".to_string(),
+ source: None,
+ });
+ }
+ self.truncate_partitions(partitions).await
}
/// Python-compatible alias for dropping partitions with a caller-provided
@@ -511,12 +579,23 @@ impl TableCommit {
/// Truncate the entire table (OVERWRITE with no filter, only deletes).
pub async fn truncate_table(&self) -> Result<()> {
- self.truncate_table_with_identifier(BATCH_COMMIT_IDENTIFIER)
+ self.truncate_table_impl(BATCH_COMMIT_IDENTIFIER, false)
.await
}
/// Truncate the entire table with a caller-provided commit identifier.
+ ///
+ /// A previously committed identifier is filtered so retrying an uncertain
+ /// result cannot delete data committed in between.
pub async fn truncate_table_with_identifier(&self, commit_identifier: i64)
-> Result<()> {
+ self.truncate_table_impl(commit_identifier, true).await
+ }
+
+ async fn truncate_table_impl(
+ &self,
+ commit_identifier: i64,
+ filter_committed: bool,
+ ) -> Result<()> {
self.table.ensure_not_branch_reference_for_write()?;
self.try_commit(
@@ -531,11 +610,12 @@ impl TableCommit {
},
None,
commit_identifier,
+ filter_committed,
)
.await
}
- /// Abort a prepared commit by deleting newly written data and changelog
files.
+ /// Abort a prepared commit by deleting newly written data, changelog and
index files.
///
/// Deletion is best-effort and mirrors Python `FileStoreCommit.abort`:
missing
/// files or storage errors are ignored so abort cleanup never masks the
@@ -554,6 +634,11 @@ impl TableCommit {
let _ = self.table.file_io().delete_file(&path).await;
}
}
+ let index_dir = format!("{}/index",
self.table.location().trim_end_matches('/'));
+ for file in &message.new_index_files {
+ let path = format!("{index_dir}/{}", file.file_name);
+ let _ = self.table.file_io().delete_file(&path).await;
+ }
}
Ok(())
}
@@ -586,14 +671,30 @@ impl TableCommit {
mut plan: CommitEntriesPlan,
expected_snapshot_id: Option<i64>,
commit_identifier: i64,
+ filter_committed: bool,
) -> Result<()> {
let mut retry_count = 0u32;
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;
+ let mut filter_committed = filter_committed;
loop {
let latest_snapshot =
self.snapshot_manager.get_latest_snapshot().await?;
+ if filter_committed {
+ if self
+ .is_committed_identifier(&latest_snapshot,
commit_identifier)
+ .await?
+ {
+ break;
+ }
+ filter_committed = false;
+ }
if let Some(start_snapshot_id) = duplicate_check_start_snapshot_id
{
if self
.is_duplicate_commit(
@@ -602,7 +703,7 @@ impl TableCommit {
commit_identifier,
&plan.commit_kind_hint(),
)
- .await
+ .await?
{
break;
}
@@ -615,6 +716,7 @@ impl TableCommit {
if resolved.entries.is_empty()
&& resolved.changelog_entries.is_empty()
&& !resolved.index_manifest_changed
+ && !commit_empty_overwrite
{
break;
}
@@ -1065,26 +1167,52 @@ impl TableCommit {
}
/// Check if this commit was already completed (idempotency).
+ async fn is_committed_identifier(
+ &self,
+ latest_snapshot: &Option<Snapshot>,
+ commit_identifier: i64,
+ ) -> Result<bool> {
+ let Some(latest) = latest_snapshot else {
+ return Ok(false);
+ };
+ let earliest_snapshot_id = self
+ .snapshot_manager
+ .earliest_snapshot_id()
+ .await?
+ .unwrap_or(latest.id());
+ for snapshot_id in (earliest_snapshot_id..=latest.id()).rev() {
+ let snapshot = if snapshot_id == latest.id() {
+ latest.clone()
+ } else {
+ self.snapshot_manager.get_snapshot(snapshot_id).await?
+ };
+ if snapshot.commit_user() == self.commit_user {
+ return Ok(commit_identifier <= snapshot.commit_identifier());
+ }
+ }
+ Ok(false)
+ }
+
+ /// Check if this commit was already completed during an in-process retry.
async fn is_duplicate_commit(
&self,
start_snapshot_id: i64,
latest_snapshot: &Option<Snapshot>,
commit_identifier: i64,
commit_kind: &CommitKind,
- ) -> bool {
+ ) -> Result<bool> {
if let Some(latest) = latest_snapshot {
for snapshot_id in start_snapshot_id..=latest.id() {
- if let Ok(snap) =
self.snapshot_manager.get_snapshot(snapshot_id).await {
- if snap.commit_user() == self.commit_user
- && snap.commit_identifier() == commit_identifier
- && snap.commit_kind() == commit_kind
- {
- return true;
- }
+ let snap =
self.snapshot_manager.get_snapshot(snapshot_id).await?;
+ if snap.commit_user() == self.commit_user
+ && snap.commit_identifier() == commit_identifier
+ && snap.commit_kind() == commit_kind
+ {
+ return Ok(true);
}
}
}
- false
+ Ok(false)
}
/// Resolve commit entries and merge index entries based on the plan type.
@@ -3102,18 +3230,263 @@ mod tests {
let snap_manager = SnapshotManager::new(file_io.clone(),
table_path.to_string());
let latest = snap_manager.get_latest_snapshot().await.unwrap();
- assert!(
- commit
- .is_duplicate_commit(1, &latest, 7, &CommitKind::APPEND)
- .await
+ assert!(commit
+ .is_duplicate_commit(1, &latest, 7, &CommitKind::APPEND)
+ .await
+ .unwrap());
+ assert!(!commit
+ .is_duplicate_commit(1, &latest, 8, &CommitKind::APPEND)
+ .await
+ .unwrap());
+ }
+
+ #[tokio::test]
+ async fn test_filter_and_commit_with_same_identity_is_idempotent() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_repeated_commit_with_same_identity";
+ setup_dirs(&file_io, table_path).await;
+
+ let commit = setup_commit(&file_io, table_path);
+ let message = CommitMessage::new(vec![], 0,
vec![test_data_file("data-0.parquet", 100)]);
+
+ commit
+ .commit_with_identifier(vec![message.clone()], 7)
+ .await
+ .unwrap();
+ commit
+ .filter_and_commit_with_identifier(vec![message], 7)
+ .await
+ .unwrap();
+
+ let snapshot = latest_snapshot(&file_io, table_path).await.unwrap();
+ assert_eq!(
+ snapshot.id(),
+ 1,
+ "retrying the same commit identity must not create another
snapshot"
);
- assert!(
- !commit
- .is_duplicate_commit(1, &latest, 8, &CommitKind::APPEND)
- .await
+ }
+
+ #[tokio::test]
+ async fn test_filter_and_commit_rejects_expired_older_identifier() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_filter_expired_identifier";
+ setup_dirs(&file_io, table_path).await;
+
+ let commit = setup_commit(&file_io, table_path);
+ let first = CommitMessage::new(vec![], 0,
vec![test_data_file("data-0.parquet", 100)]);
+ let second = CommitMessage::new(vec![], 0,
vec![test_data_file("data-1.parquet", 100)]);
+ commit
+ .commit_with_identifier(vec![first.clone()], 1)
+ .await
+ .unwrap();
+ commit
+ .commit_with_identifier(vec![second], 2)
+ .await
+ .unwrap();
+
+ let snapshot_manager = SnapshotManager::new(file_io.clone(),
table_path.to_string());
+ snapshot_manager.delete_snapshot(1).await.unwrap();
+ commit
+ .filter_and_commit_with_identifier(vec![first], 1)
+ .await
+ .unwrap();
+
+ let snapshot = latest_snapshot(&file_io, table_path).await.unwrap();
+ assert_eq!(
+ snapshot.id(),
+ 2,
+ "an expired older identifier is still a retry"
);
}
+ #[tokio::test]
+ async fn test_overwrite_retry_preserves_intervening_commit() {
+ let file_io = test_file_io();
+ let table_path =
"memory:/test_overwrite_retry_preserves_intervening_commit";
+ setup_dirs(&file_io, table_path).await;
+
+ let commit = setup_commit(&file_io, table_path);
+ let other_commit = TableCommit::new(test_table(&file_io, table_path),
"other-user".into());
+ commit
+ .commit_with_identifier(
+ vec![CommitMessage::new(
+ vec![],
+ 0,
+ vec![test_data_file("initial.parquet", 100)],
+ )],
+ 1,
+ )
+ .await
+ .unwrap();
+
+ let overwrite =
+ CommitMessage::new(vec![], 0,
vec![test_data_file("overwrite.parquet", 100)]);
+ commit
+ .overwrite_with_identifier(vec![overwrite.clone()], None, 2)
+ .await
+ .unwrap();
+ other_commit
+ .commit_with_identifier(
+ vec![CommitMessage::new(
+ vec![],
+ 0,
+ vec![test_data_file("intervening.parquet", 100)],
+ )],
+ 1,
+ )
+ .await
+ .unwrap();
+
+ commit
+ .overwrite_with_identifier(vec![overwrite], None, 2)
+ .await
+ .unwrap();
+
+ let snapshot = latest_snapshot(&file_io, table_path).await.unwrap();
+ assert_eq!(snapshot.id(), 3, "the retry must not create a snapshot");
+ let mut file_names = active_entries(&file_io, table_path, &snapshot)
+ .await
+ .into_iter()
+ .map(|entry| entry.file().file_name.clone())
+ .collect::<Vec<_>>();
+ file_names.sort();
+ assert_eq!(
+ file_names,
+ vec!["intervening.parquet", "overwrite.parquet"],
+ "the retry must not remove data committed in between"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_truncate_retry_preserves_intervening_commit() {
+ let file_io = test_file_io();
+ let table_path =
"memory:/test_truncate_retry_preserves_intervening_commit";
+ setup_dirs(&file_io, table_path).await;
+
+ let commit = setup_commit(&file_io, table_path);
+ let other_commit = TableCommit::new(test_table(&file_io, table_path),
"other-user".into());
+ commit
+ .commit_with_identifier(
+ vec![CommitMessage::new(
+ vec![],
+ 0,
+ vec![test_data_file("initial.parquet", 100)],
+ )],
+ 1,
+ )
+ .await
+ .unwrap();
+ commit.truncate_table_with_identifier(2).await.unwrap();
+ other_commit
+ .commit_with_identifier(
+ vec![CommitMessage::new(
+ vec![],
+ 0,
+ vec![test_data_file("intervening.parquet", 100)],
+ )],
+ 1,
+ )
+ .await
+ .unwrap();
+
+ commit.truncate_table_with_identifier(2).await.unwrap();
+
+ let snapshot = latest_snapshot(&file_io, table_path).await.unwrap();
+ assert_eq!(snapshot.id(), 3, "the retry must not create a snapshot");
+ let entries = active_entries(&file_io, table_path, &snapshot).await;
+ assert_eq!(entries.len(), 1);
+ assert_eq!(entries[0].file().file_name, "intervening.parquet");
+ }
+
+ #[tokio::test]
+ async fn test_empty_truncate_retry_preserves_intervening_commit() {
+ let file_io = test_file_io();
+ let table_path =
"memory:/test_empty_truncate_retry_preserves_intervening_commit";
+ setup_dirs(&file_io, table_path).await;
+
+ let commit = setup_commit(&file_io, table_path);
+ let other_commit = TableCommit::new(test_table(&file_io, table_path),
"other-user".into());
+
+ commit.truncate_table_with_identifier(1).await.unwrap();
+ let marker = latest_snapshot(&file_io, table_path)
+ .await
+ .expect("identified no-op truncate must record a snapshot");
+ assert_eq!(marker.commit_user(), "test-user");
+ assert_eq!(marker.commit_identifier(), 1);
+ assert_eq!(marker.commit_kind(), &CommitKind::OVERWRITE);
+ other_commit
+ .commit_with_identifier(
+ vec![CommitMessage::new(
+ vec![],
+ 0,
+ vec![test_data_file("intervening.parquet", 100)],
+ )],
+ 1,
+ )
+ .await
+ .unwrap();
+
+ commit.truncate_table_with_identifier(1).await.unwrap();
+
+ let snapshot = latest_snapshot(&file_io, table_path).await.unwrap();
+ let entries = active_entries(&file_io, table_path, &snapshot).await;
+ assert_eq!(entries.len(), 1);
+ assert_eq!(
+ entries[0].file().file_name,
+ "intervening.parquet",
+ "retrying a no-op truncate must not delete data committed in
between"
+ );
+ }
+
+ #[tokio::test]
+ async fn test_identifier_filter_propagates_snapshot_read_error() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_identifier_filter_snapshot_read_error";
+ setup_dirs(&file_io, table_path).await;
+
+ let commit = setup_commit(&file_io, table_path);
+ for identifier in 1..=2 {
+ commit
+ .commit_with_identifier(
+ vec![CommitMessage::new(
+ vec![],
+ 0,
+
vec![test_data_file(&format!("data-{identifier}.parquet"), 100)],
+ )],
+ identifier,
+ )
+ .await
+ .unwrap();
+ }
+ let other_commit = TableCommit::new(test_table(&file_io, table_path),
"other-user".into());
+ other_commit
+ .commit_with_identifier(
+ vec![CommitMessage::new(
+ vec![],
+ 0,
+ vec![test_data_file("other.parquet", 100)],
+ )],
+ 1,
+ )
+ .await
+ .unwrap();
+
+ let snapshot_manager = SnapshotManager::new(file_io.clone(),
table_path.to_string());
+ file_io
+ .new_output(&snapshot_manager.snapshot_path(2))
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"not-json"))
+ .await
+ .unwrap();
+ let latest = snapshot_manager.get_latest_snapshot().await.unwrap();
+
+ let error = commit
+ .is_committed_identifier(&latest, 2)
+ .await
+ .expect_err("an unreadable active snapshot must fail identifier
filtering");
+ assert!(error.to_string().contains("snapshot JSON invalid"));
+ }
+
#[tokio::test]
async fn test_multiple_appends() {
let file_io = test_file_io();
@@ -4688,6 +5061,40 @@ mod tests {
.unwrap());
}
+ #[tokio::test]
+ async fn test_abort_deletes_new_index_files() {
+ let file_io = test_file_io();
+ let table_path = "memory:/test_abort_index_cleanup";
+ setup_dirs(&file_io, table_path).await;
+
+ let commit = setup_commit(&file_io, table_path);
+ let index_dir = format!("{table_path}/index");
+ let index_path = format!("{index_dir}/bucket-index");
+ file_io.mkdirs(&format!("{index_dir}/")).await.unwrap();
+ file_io
+ .new_output(&index_path)
+ .unwrap()
+ .write(bytes::Bytes::from_static(b"index"))
+ .await
+ .unwrap();
+
+ let mut message = CommitMessage::new(vec![], 0, vec![]);
+ message.new_index_files = vec![IndexFileMeta {
+ index_type: "HASH".to_string(),
+ file_name: "bucket-index".to_string(),
+ file_size: 5,
+ row_count: 1,
+ deletion_vectors_ranges: None,
+ global_index_meta: None,
+ }];
+ commit.abort(&[message]).await.unwrap();
+
+ assert!(
+ !file_io.exists(&index_path).await.unwrap(),
+ "abort must remove newly written index files"
+ );
+ }
+
#[tokio::test]
async fn test_delete_conflict_rejects_missing_file() {
let file_io = test_file_io();
