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/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/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..72ba5829 --- /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, 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>, 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>, + 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)), + ) +} + +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 { + 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()); + 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 c7b059ae..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)] @@ -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, + 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, +} + +#[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 00000000..500c1cd0 --- /dev/null +++ b/bindings/c/src/write.rs @@ -0,0 +1,801 @@ +// 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, +) -> 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) -> *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 { + // 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_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; + } + 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.overwrite_with_identifier( + messages.messages.clone(), + None, + commit_identifier, + )) { + 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_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 TableCommitState); + + match runtime().block_on( + table_commit + .commit + .truncate_table_with_identifier(commit_identifier), + ) { + 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 2a81047e..fd345a41 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, 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, + 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, + 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 } @@ -262,6 +291,7 @@ impl TableCommit { }, None, commit_identifier, + false, ) .await } @@ -476,6 +506,7 @@ impl TableCommit { }, None, commit_identifier, + false, ) .await } @@ -531,11 +562,12 @@ impl TableCommit { }, None, commit_identifier, + false, ) .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 +586,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 +623,25 @@ impl TableCommit { mut plan: CommitEntriesPlan, expected_snapshot_id: Option, commit_identifier: i64, + filter_committed: bool, ) -> Result<()> { let mut retry_count = 0u32; let mut duplicate_check_start_snapshot_id: Option = None; let mut retry_state: Option> = None; let start_time_ms = current_time_millis(); + 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( @@ -1065,6 +1113,35 @@ impl TableCommit { } /// Check if this commit was already completed (idempotency). + async fn is_committed_identifier( + &self, + latest_snapshot: &Option, + commit_identifier: i64, + ) -> Result { + 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() { + Some(latest.clone()) + } else { + self.snapshot_manager.get_snapshot(snapshot_id).await.ok() + }; + if let Some(snapshot) = snapshot { + 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, @@ -3114,6 +3191,65 @@ mod tests { ); } + #[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" + ); + } + + #[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_multiple_appends() { let file_io = test_file_io(); @@ -4688,6 +4824,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();