From e482f51f771a42f4c711abe34f26e01f0e8141a7 Mon Sep 17 00:00:00 2001 From: Socrates Date: Tue, 14 Jul 2026 17:07:56 +0800 Subject: [PATCH 1/2] feat(c): add write/commit C FFI bindings 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 Tests (15 cases): - Read path: empty table read, data read, column projection - Predicate: equal, IS NOT NULL, IS IN, less_than filter, AND combinator - Write path: builder lifecycle, full roundtrip, multiple batches, empty commit, overwrite mode, truncate, abort - Error handling: null pointer checks - Integration: two commits with same builder Design follows Python bindings: in-process commit message transfer, shared commit_user between writer and committer, Arrow C Data Interface for zero-copy batch import. Closes #520 --- bindings/c/Cargo.toml | 4 +- bindings/c/src/lib.rs | 3 + bindings/c/src/result.rs | 26 + bindings/c/src/tests.rs | 996 +++++++++++++++++++++++++++++++++++++++ bindings/c/src/types.rs | 31 ++ bindings/c/src/write.rs | 506 ++++++++++++++++++++ 6 files changed, 1565 insertions(+), 1 deletion(-) create mode 100644 bindings/c/src/tests.rs create mode 100644 bindings/c/src/write.rs diff --git a/bindings/c/Cargo.toml b/bindings/c/Cargo.toml index a6e27cf6..6eb34c12 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/lib.rs b/bindings/c/src/lib.rs index 19942c6c..3e1367ea 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 19d523cb..216509f5 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 00000000..348245bf --- /dev/null +++ b/bindings/c/src/tests.rs @@ -0,0 +1,996 @@ +// 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::ptr; +use std::sync::Arc; + +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::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) +} + +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, 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 export_batch_to_ffi( + batch: RecordBatch, +) -> ( + Box>, + Box>, +) { + 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)), + ) +} + +/// 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::() + .unwrap(); + let name_arr = batch + .column_by_name("name") + .unwrap() + .as_any() + .downcast_ref::() + .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()); + + 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_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()); + + 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()); + + 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_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_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_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()); + + 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_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_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 c7b059ae..91bcc9e4 100644 --- a/bindings/c/src/types.rs +++ b/bindings/c/src/types.rs @@ -155,6 +155,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 +179,33 @@ 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, +} + +#[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 a Vec produced by prepare_commit. +#[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 00000000..23e83316 --- /dev/null +++ b/bindings/c/src/write.rs @@ -0,0 +1,506 @@ +// 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_void; +use std::ptr; + +use arrow_array::ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema}; +use arrow_array::{RecordBatch, StructArray}; +use paimon::table::{CommitMessage, Table, TableCommit, TableWrite}; + +use crate::error::{check_non_null, paimon_error, 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 =============================== + +/// 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 { + 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 wb = table_ref.new_write_builder(); + let commit_user = wb.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(), + } +} + +/// 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 =============================== + +/// 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 inner = Box::into_raw(Box::new(tw)) 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 TableWrite)); + } + } +} + +/// 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 TableWrite); + + let ffi_array = ptr::read(array as *const FFI_ArrowArray); + let ffi_schema = ptr::read(schema as *const FFI_ArrowSchema); + + let batch = match unsafe { from_ffi(ffi_array, &ffi_schema) } { + Ok(data) => { + // The ffi_array was consumed by from_ffi (moved by value). + // The ffi_schema was only borrowed; it will be dropped below, + // calling release on the schema resources. + drop(ffi_schema); + RecordBatch::from(StructArray::from(data)) + } + Err(e) => { + // from_ffi consumed ffi_array (by value). Its drop will call release. + // We let ffi_schema drop normally here too. + drop(ffi_schema); + return paimon_error::new( + PaimonErrorCode::InvalidInput, + format!("Failed to import Arrow record batch: {e}"), + ); + } + }; + + match runtime().block_on(table_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 TableWrite); + + match runtime().block_on(table_write.prepare_commit()) { + Ok(messages) => { + 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 inner = Box::into_raw(Box::new(tc)) 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 TableCommit)); + } + } +} + +// ======================= 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 Vec)); + } + } +} + +// ======================= Commit operations =============================== + +/// Commit the given messages in APPEND mode. +/// +/// Empty messages is a no-op success. +/// +/// # 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 { + 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 TableCommit); + let messages = Box::from_raw((*msgs).inner as *mut Vec); + // Release the outer wrapper to avoid double-free when caller frees msgs + drop(Box::from_raw(msgs)); + + if messages.is_empty() { + return ptr::null_mut(); + } + + match runtime().block_on(table_commit.commit(*messages)) { + 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). +/// +/// # 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 { + 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 TableCommit); + let messages = Box::from_raw((*msgs).inner as *mut Vec); + drop(Box::from_raw(msgs)); + + if messages.is_empty() { + return ptr::null_mut(); + } + + match runtime().block_on(table_commit.overwrite(*messages, None)) { + 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 { + if let Err(e) = check_non_null(tc, "tc") { + return e; + } + + let table_commit = &*((*tc).inner as *const TableCommit); + + match runtime().block_on(table_commit.truncate_table()) { + 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 and +/// changelog files produced by the writer. Errors during cleanup are +/// returned but do not roll back the cleanup of previously-deleted files. +/// +/// # 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 TableCommit); + let messages = Box::from_raw((*msgs).inner as *mut Vec); + drop(Box::from_raw(msgs)); + + if messages.is_empty() { + return ptr::null_mut(); + } + + match runtime().block_on(table_commit.abort(&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_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_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, +) -> *mut paimon_error = paimon_table_commit_overwrite; +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, + *mut paimon_commit_messages, +) -> *mut paimon_error = paimon_table_commit_abort; From 27a9d903e66bbcb104ccf5adf01f80c510932ce7 Mon Sep 17 00:00:00 2001 From: Socrates Date: Wed, 15 Jul 2026 10:55:20 +0800 Subject: [PATCH 2/2] fix(c): address write FFI review comments --- bindings/c/src/error.rs | 1 + bindings/c/src/tests.rs | 400 +++++++++++++++++++++++- bindings/c/src/types.rs | 25 +- bindings/c/src/write.rs | 345 +++++++++++++++----- crates/paimon/src/table/table_commit.rs | 75 ++++- 5 files changed, 769 insertions(+), 77 deletions(-) diff --git a/bindings/c/src/error.rs b/bindings/c/src/error.rs index e8d0757f..7b0a88fc 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/tests.rs b/bindings/c/src/tests.rs index 348245bf..1e95d95e 100644 --- a/bindings/c/src/tests.rs +++ b/bindings/c/src/tests.rs @@ -27,16 +27,18 @@ 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::Table; +use paimon::table::{SnapshotManager, Table}; use crate::error::*; use crate::table::*; @@ -87,6 +89,21 @@ fn make_batch(ids: Vec, names: Vec<&str>) -> RecordBatch { .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 export_batch_to_ffi( batch: RecordBatch, ) -> ( @@ -103,6 +120,32 @@ fn export_batch_to_ffi( ) } +fn export_array_to_ffi( + array: &dyn Array, +) -> ( + Box>, + Box>, +) { + 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 { @@ -614,6 +657,7 @@ fn test_write_commit_read_roundtrip() { 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!( @@ -628,6 +672,352 @@ fn test_write_commit_read_roundtrip() { } } +#[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_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_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_write_multiple_batches() { let path = "memory:/test_write_multi_batch"; @@ -667,6 +1057,7 @@ fn test_write_multiple_batches() { 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())]); @@ -706,6 +1097,7 @@ fn test_commit_empty_messages_noop() { 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()); @@ -750,6 +1142,7 @@ fn test_write_overwrite_mode() { 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); @@ -778,6 +1171,7 @@ fn test_write_overwrite_mode() { 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); @@ -822,6 +1216,7 @@ fn test_truncate_table() { 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); @@ -882,6 +1277,7 @@ fn test_abort_commit() { 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()); @@ -964,6 +1360,7 @@ fn test_two_commits_same_builder() { 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); } @@ -982,6 +1379,7 @@ fn test_two_commits_same_builder() { 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); } diff --git a/bindings/c/src/types.rs b/bindings/c/src/types.rs index 91bcc9e4..8d3a4666 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)] @@ -189,6 +191,25 @@ pub(crate) struct WriteBuilderState { pub overwrite: bool, } +pub(crate) struct TableWriteState { + pub write: TableWrite, + pub target_schema: Arc, + 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, + pub table_location: String, + pub commit_user: String, +} + #[repr(C)] pub struct paimon_write_builder { pub inner: *mut c_void, @@ -204,7 +225,7 @@ pub struct paimon_table_commit { pub inner: *mut c_void, } -/// Opaque container for a Vec produced by prepare_commit. +/// 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 index 23e83316..5a60de7a 100644 --- a/bindings/c/src/write.rs +++ b/bindings/c/src/write.rs @@ -15,14 +15,16 @@ // specific language governing permissions and limitations // under the License. -use std::ffi::c_void; +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::{RecordBatch, StructArray}; -use paimon::table::{CommitMessage, Table, TableCommit, TableWrite}; +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, PaimonErrorCode}; +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, @@ -32,16 +34,9 @@ use crate::types::*; // ======================= WriteBuilder =============================== -/// 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( +unsafe fn new_write_builder( table: *const paimon_table, + commit_user: Option, ) -> paimon_result_write_builder { if let Err(e) = check_non_null(table, "table") { return paimon_result_write_builder { @@ -50,8 +45,19 @@ pub unsafe extern "C" fn paimon_table_new_write_builder( }; } let table_ref = &*((*table).inner as *const Table); - let wb = table_ref.new_write_builder(); - let commit_user = wb.commit_user().to_string(); + 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, @@ -64,6 +70,44 @@ pub unsafe extern "C" fn paimon_table_new_write_builder( } } +/// 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. +/// +/// Distributed writers for one commit should 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 @@ -99,6 +143,75 @@ pub unsafe extern "C" fn paimon_write_builder_with_overwrite( // ======================= TableWrite =============================== +fn invalid_input(message: impl Into) -> *mut paimon_error { + paimon_error::new(PaimonErrorCode::InvalidInput, message.into()) +} + +fn validate_batch_schema( + input: &ArrowSchema, + target: &ArrowSchema, +) -> Result<(), *mut paimon_error> { + let matches = input.fields().len() == target.fields().len() + && input + .fields() + .iter() + .zip(target.fields().iter()) + .all(|(input, target)| { + input.name() == target.name() && input.data_type() == target.data_type() + }); + if matches { + Ok(()) + } else { + Err(invalid_input(format!( + "Input schema is not consistent with the table schema. input: {input:?}, table: {target:?}" + ))) + } +} + +unsafe fn import_record_batch( + array: *mut c_void, + schema: *mut c_void, +) -> Result { + // 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. @@ -145,7 +258,23 @@ pub unsafe extern "C" fn paimon_write_builder_new_write( } }; - let inner = Box::into_raw(Box::new(tw)) as *mut c_void; + 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(), @@ -164,7 +293,7 @@ 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 TableWrite)); + drop(Box::from_raw(wrapper.inner as *mut TableWriteState)); } } } @@ -196,31 +325,16 @@ pub unsafe extern "C" fn paimon_table_write_write_arrow_batch( return e; } - let table_write = &mut *((*tw).inner as *mut TableWrite); - - let ffi_array = ptr::read(array as *const FFI_ArrowArray); - let ffi_schema = ptr::read(schema as *const FFI_ArrowSchema); - - let batch = match unsafe { from_ffi(ffi_array, &ffi_schema) } { - Ok(data) => { - // The ffi_array was consumed by from_ffi (moved by value). - // The ffi_schema was only borrowed; it will be dropped below, - // calling release on the schema resources. - drop(ffi_schema); - RecordBatch::from(StructArray::from(data)) - } - Err(e) => { - // from_ffi consumed ffi_array (by value). Its drop will call release. - // We let ffi_schema drop normally here too. - drop(ffi_schema); - return paimon_error::new( - PaimonErrorCode::InvalidInput, - format!("Failed to import Arrow record batch: {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.schema(), &table_write.target_schema) { + return error; + } - match runtime().block_on(table_write.write_arrow_batch(&batch)) { + match runtime().block_on(table_write.write.write_arrow_batch(&batch)) { Ok(()) => ptr::null_mut(), Err(e) => paimon_error::from_paimon(e), } @@ -248,10 +362,15 @@ pub unsafe extern "C" fn paimon_table_write_prepare_commit( error: e, }; } - let table_write = &mut *((*tw).inner as *mut TableWrite); + let table_write = &mut *((*tw).inner as *mut TableWriteState); - match runtime().block_on(table_write.prepare_commit()) { + 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 })), @@ -310,7 +429,12 @@ pub unsafe extern "C" fn paimon_write_builder_new_commit( } }; - let inner = Box::into_raw(Box::new(tc)) as *mut c_void; + 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(), @@ -326,7 +450,7 @@ 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 TableCommit)); + drop(Box::from_raw(wrapper.inner as *mut TableCommitState)); } } } @@ -342,16 +466,36 @@ pub unsafe extern "C" fn paimon_commit_messages_free(msgs: *mut paimon_commit_me if !msgs.is_null() { let wrapper = Box::from_raw(msgs); if !wrapper.inner.is_null() { - drop(Box::from_raw(wrapper.inner as *mut Vec)); + drop(Box::from_raw(wrapper.inner as *mut CommitMessagesState)); } } } // ======================= 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). @@ -360,6 +504,20 @@ pub unsafe extern "C" fn paimon_commit_messages_free(msgs: *mut paimon_commit_me 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 stable identifier. +/// +/// Reusing the same non-batch identifier with the same `commit_user` is +/// idempotent, including across separate invocations after 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; @@ -368,16 +526,17 @@ pub unsafe extern "C" fn paimon_table_commit_commit( return e; } - let table_commit = &*((*tc).inner as *const TableCommit); - let messages = Box::from_raw((*msgs).inner as *mut Vec); - // Release the outer wrapper to avoid double-free when caller frees msgs - drop(Box::from_raw(msgs)); - - if messages.is_empty() { - return ptr::null_mut(); + 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(*messages)) { + 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), } @@ -387,6 +546,8 @@ pub unsafe extern "C" fn paimon_table_commit_commit( /// /// `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). @@ -395,6 +556,18 @@ pub unsafe extern "C" fn paimon_table_commit_commit( 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_with_identifier(tc, msgs, i64::MAX) +} + +/// 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 { if let Err(e) = check_non_null(tc, "tc") { return e; @@ -403,15 +576,17 @@ pub unsafe extern "C" fn paimon_table_commit_overwrite( return e; } - let table_commit = &*((*tc).inner as *const TableCommit); - let messages = Box::from_raw((*msgs).inner as *mut Vec); - drop(Box::from_raw(msgs)); - - if messages.is_empty() { - return ptr::null_mut(); + 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.overwrite(*messages, None)) { + match runtime().block_on(table_commit.commit.overwrite_with_identifier( + messages.messages.clone(), + None, + commit_identifier, + )) { Ok(()) => ptr::null_mut(), Err(e) => paimon_error::from_paimon(e), } @@ -427,14 +602,27 @@ pub unsafe extern "C" fn paimon_table_commit_overwrite( #[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_with_identifier(tc, i64::MAX) +} + +/// 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 { if let Err(e) = check_non_null(tc, "tc") { return e; } - let table_commit = &*((*tc).inner as *const TableCommit); + let table_commit = &*((*tc).inner as *const TableCommitState); - match runtime().block_on(table_commit.truncate_table()) { + match runtime().block_on( + table_commit + .commit + .truncate_table_with_identifier(commit_identifier), + ) { Ok(()) => ptr::null_mut(), Err(e) => paimon_error::from_paimon(e), } @@ -442,9 +630,10 @@ pub unsafe extern "C" fn paimon_table_commit_truncate_table( /// Abort a prepared commit, cleaning up written data files. /// -/// This is a best-effort cleanup — it attempts to delete new data and -/// changelog files produced by the writer. Errors during cleanup are -/// returned but do not roll back the cleanup of previously-deleted 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). @@ -461,15 +650,13 @@ pub unsafe extern "C" fn paimon_table_commit_abort( return e; } - let table_commit = &*((*tc).inner as *const TableCommit); - let messages = Box::from_raw((*msgs).inner as *mut Vec); - drop(Box::from_raw(msgs)); - - if messages.is_empty() { - return ptr::null_mut(); + 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.abort(&messages)) { + match runtime().block_on(table_commit.commit.abort(&messages.messages)) { Ok(()) => ptr::null_mut(), Err(e) => paimon_error::from_paimon(e), } @@ -479,6 +666,8 @@ pub unsafe extern "C" fn paimon_table_commit_abort( 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 = @@ -494,12 +683,24 @@ 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, ) -> *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, diff --git a/crates/paimon/src/table/table_commit.rs b/crates/paimon/src/table/table_commit.rs index 2a81047e..07f36d81 100644 --- a/crates/paimon/src/table/table_commit.rs +++ b/crates/paimon/src/table/table_commit.rs @@ -535,7 +535,7 @@ impl TableCommit { .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 +554,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(()) } @@ -594,7 +599,13 @@ impl TableCommit { loop { let latest_snapshot = self.snapshot_manager.get_latest_snapshot().await?; - if let Some(start_snapshot_id) = duplicate_check_start_snapshot_id { + // Explicit identifiers are stable commit identities and must also + // deduplicate a fresh invocation after an uncertain result. The + // batch identifier is intentionally excluded because callers may + // perform multiple independent batch commits with the same user. + let duplicate_check_start = duplicate_check_start_snapshot_id + .or((commit_identifier != BATCH_COMMIT_IDENTIFIER).then_some(1)); + if let Some(start_snapshot_id) = duplicate_check_start { if self .is_duplicate_commit( start_snapshot_id, @@ -3114,6 +3125,32 @@ mod tests { ); } + #[tokio::test] + async fn test_repeated_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 + .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" + ); + } + #[tokio::test] async fn test_multiple_appends() { let file_io = test_file_io(); @@ -4688,6 +4725,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();