QuakeWang commented on code in PR #522: URL: https://github.com/apache/paimon-rust/pull/522#discussion_r3584942967
########## bindings/c/src/write.rs: ########## @@ -0,0 +1,707 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::ffi::{c_char, c_void}; +use std::ptr; +use std::sync::Arc; + +use arrow_array::ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema}; +use arrow_array::{Array, RecordBatch, RecordBatchOptions, StructArray}; +use arrow_schema::{DataType as ArrowDataType, Schema as ArrowSchema}; +use paimon::table::Table; + +use crate::error::{check_non_null, paimon_error, validate_cstr, PaimonErrorCode}; +use crate::result::{ + paimon_result_prepare_commit, paimon_result_table_commit, paimon_result_table_write, + paimon_result_write_builder, +}; +use crate::runtime; +use crate::types::*; + +// ======================= WriteBuilder =============================== + +unsafe fn new_write_builder( + table: *const paimon_table, + commit_user: Option<String>, +) -> paimon_result_write_builder { + if let Err(e) = check_non_null(table, "table") { + return paimon_result_write_builder { + write_builder: ptr::null_mut(), + error: e, + }; + } + let table_ref = &*((*table).inner as *const Table); + let builder = table_ref.new_write_builder(); + let commit_user = match commit_user { + Some(commit_user) => match builder.with_commit_user(commit_user) { + Ok(builder) => builder.commit_user().to_string(), + Err(e) => { + return paimon_result_write_builder { + write_builder: ptr::null_mut(), + error: paimon_error::from_paimon(e), + } + } + }, + None => builder.commit_user().to_string(), + }; + let state = WriteBuilderState { + table: table_ref.clone(), + commit_user, + overwrite: false, + }; + let inner = Box::into_raw(Box::new(state)) as *mut c_void; + paimon_result_write_builder { + write_builder: Box::into_raw(Box::new(paimon_write_builder { inner })), + error: ptr::null_mut(), + } +} + +/// Create a new WriteBuilder from a Table. +/// +/// The returned WriteBuilder holds a shared `commit_user` (UUID) that will be +/// used by both `new_write()` and `new_commit()` for duplicate-commit detection. +/// +/// # Safety +/// `table` must be a valid pointer from `paimon_catalog_get_table`, or null (returns error). +#[no_mangle] +pub unsafe extern "C" fn paimon_table_new_write_builder( + table: *const paimon_table, +) -> paimon_result_write_builder { + new_write_builder(table, None) +} + +/// Create a WriteBuilder with a caller-provided stable commit identity. +/// +/// Distributed writers for one commit should use the same `commit_user`. Review Comment: This states that distributed writers can share a `commit_user`, but opaque commit messages cannot be merged. If two writers commit separately with the same identifier, the second call succeeds without adding its files to the snapshot, silently dropping data. ########## crates/paimon/src/table/table_commit.rs: ########## @@ -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)); Review Comment: Every commit with an explicit identifier scans from snapshot 1 to the latest snapshot, resulting in O(N²) snapshot reads over successive commits. If an older snapshot has expired, retrying its identifier may also commit the data again, breaking idempotency. ########## bindings/c/src/write.rs: ########## @@ -0,0 +1,707 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::ffi::{c_char, c_void}; +use std::ptr; +use std::sync::Arc; + +use arrow_array::ffi::{from_ffi, FFI_ArrowArray, FFI_ArrowSchema}; +use arrow_array::{Array, RecordBatch, RecordBatchOptions, StructArray}; +use arrow_schema::{DataType as ArrowDataType, Schema as ArrowSchema}; +use paimon::table::Table; + +use crate::error::{check_non_null, paimon_error, validate_cstr, PaimonErrorCode}; +use crate::result::{ + paimon_result_prepare_commit, paimon_result_table_commit, paimon_result_table_write, + paimon_result_write_builder, +}; +use crate::runtime; +use crate::types::*; + +// ======================= WriteBuilder =============================== + +unsafe fn new_write_builder( + table: *const paimon_table, + commit_user: Option<String>, +) -> paimon_result_write_builder { + if let Err(e) = check_non_null(table, "table") { + return paimon_result_write_builder { + write_builder: ptr::null_mut(), + error: e, + }; + } + let table_ref = &*((*table).inner as *const Table); + let builder = table_ref.new_write_builder(); + let commit_user = match commit_user { + Some(commit_user) => match builder.with_commit_user(commit_user) { + Ok(builder) => builder.commit_user().to_string(), + Err(e) => { + return paimon_result_write_builder { + write_builder: ptr::null_mut(), + error: paimon_error::from_paimon(e), + } + } + }, + None => builder.commit_user().to_string(), + }; + let state = WriteBuilderState { + table: table_ref.clone(), + commit_user, + overwrite: false, + }; + let inner = Box::into_raw(Box::new(state)) as *mut c_void; + paimon_result_write_builder { + write_builder: Box::into_raw(Box::new(paimon_write_builder { inner })), + error: ptr::null_mut(), + } +} + +/// Create a new WriteBuilder from a Table. +/// +/// The returned WriteBuilder holds a shared `commit_user` (UUID) that will be +/// used by both `new_write()` and `new_commit()` for duplicate-commit detection. +/// +/// # Safety +/// `table` must be a valid pointer from `paimon_catalog_get_table`, or null (returns error). +#[no_mangle] +pub unsafe extern "C" fn paimon_table_new_write_builder( + table: *const paimon_table, +) -> paimon_result_write_builder { + new_write_builder(table, None) +} + +/// Create a WriteBuilder with a caller-provided stable commit identity. +/// +/// 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 +/// Only call with a write_builder returned from `paimon_table_new_write_builder`. +#[no_mangle] +pub unsafe extern "C" fn paimon_write_builder_free(wb: *mut paimon_write_builder) { + if !wb.is_null() { + let wrapper = Box::from_raw(wb); + if !wrapper.inner.is_null() { + drop(Box::from_raw(wrapper.inner as *mut WriteBuilderState)); + } + } +} + +/// Enable overwrite mode for the WriteBuilder. +/// +/// In overwrite mode, a subsequent `paimon_table_commit_overwrite` will replace +/// the data in the written partitions rather than appending. +/// +/// # Safety +/// `wb` must be a valid pointer from `paimon_table_new_write_builder`, or null (returns error). +#[no_mangle] +pub unsafe extern "C" fn paimon_write_builder_with_overwrite( + wb: *mut paimon_write_builder, +) -> *mut paimon_error { + if let Err(e) = check_non_null(wb, "wb") { + return e; + } + let state = &mut *((*wb).inner as *mut WriteBuilderState); + state.overwrite = true; + ptr::null_mut() +} + +// ======================= TableWrite =============================== + +fn invalid_input(message: impl Into<String>) -> *mut paimon_error { + paimon_error::new(PaimonErrorCode::InvalidInput, message.into()) +} + +fn validate_batch_schema( + input: &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() Review Comment: This ignores the target field's nullability. A nullable Arrow column containing nulls is accepted for a Paimon `NOT NULL` field, allowing data that violates the table schema to be written. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
