From 2fd97d625e509c7c15c84e813124b23a588817a4 Mon Sep 17 00:00:00 2001 From: Socrates Date: Thu, 5 Nov 2026 16:15:12 +0800 Subject: [PATCH] feat(c): export a pinned table's Arrow schema --- bindings/c/src/table.rs | 48 +++++++++ bindings/c/src/tests.rs | 18 +++- bindings/c/src/tests/table_arrow_schema.rs | 119 +++++++++++++++++++++ docs/src/c-binding.md | 26 +++++ 4 files changed, 210 insertions(+), 1 deletion(-) create mode 100644 bindings/c/src/tests/table_arrow_schema.rs diff --git a/bindings/c/src/table.rs b/bindings/c/src/table.rs index 0feba23be..873581ddd 100644 --- a/bindings/c/src/table.rs +++ b/bindings/c/src/table.rs @@ -62,6 +62,54 @@ unsafe fn box_table_read_state(state: TableReadState) -> *mut paimon_table_read // ======================= Table =============================== +/// Export the pinned table's write schema through the Arrow C Data Interface. +/// Uses the core Paimon-to-Arrow mapping without querying the catalog. +/// +/// The table is borrowed. The output remains valid after the table is freed. +/// Release it with its ArrowSchema release callback, not paimon_arrow_batch_free. +/// Errors leave the output unchanged. A live output must be released before reuse. +/// +/// # Safety +/// table must be a live table handle. schema must point to an initialized, +/// writable, empty ArrowSchema (release is null). Null pointers return an error. +#[no_mangle] +pub unsafe extern "C" fn paimon_table_export_arrow_schema( + table: *const paimon_table, + schema: *mut c_void, +) -> *mut paimon_error { + if let Err(error) = check_non_null(table, "table") { + return error; + } + if let Err(error) = check_non_null((*table).inner, "table.inner") { + return error; + } + if let Err(error) = check_non_null(schema, "schema") { + return error; + } + let output = schema.cast::(); + if (*output).release.is_some() { + return paimon_error::new( + PaimonErrorCode::InvalidInput, + "ArrowSchema output must be empty".into(), + ); + } + let table = &*((*table).inner as *const Table); + let schema = match paimon::arrow::build_target_arrow_schema(table.schema().fields()) { + Ok(schema) => schema, + Err(error) => return paimon_error::from_paimon(error), + }; + match FFI_ArrowSchema::try_from(schema.as_ref()) { + Ok(schema) => { + std::ptr::write(output, schema); + std::ptr::null_mut() + } + Err(error) => paimon_error::from_paimon(paimon::Error::UnexpectedError { + message: format!("Failed to export Arrow schema: {error}"), + source: Some(Box::new(error)), + }), + } +} + fn resolved_table_result( file_io: FileIO, table_path: String, diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs index 91a0f8e69..814cd2742 100644 --- a/bindings/c/src/tests.rs +++ b/bindings/c/src/tests.rs @@ -27,6 +27,7 @@ //! runtime only for Rust-API setup (write_data_rust, setup_table_dirs). mod commit_messages; +mod table_arrow_schema; use std::collections::HashMap; use std::ffi::{c_void, CString}; @@ -1920,7 +1921,22 @@ fn test_write_commit_read_roundtrip() { assert!(tw_result.error.is_null()); let tw = tw_result.write; - let batch = make_batch(vec![1, 2, 3], vec!["a", "b", "c"]); + let mut exported = FFI_ArrowSchema::empty(); + assert!(paimon_table_export_arrow_schema( + handle, + (&mut exported as *mut FFI_ArrowSchema).cast(), + ) + .is_null()); + let schema = ArrowSchema::try_from(&exported).unwrap(); + exported.release.unwrap()(&mut exported); + let batch = RecordBatch::try_new( + Arc::new(schema), + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(StringArray::from(vec!["a", "b", "c"])), + ], + ) + .unwrap(); 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; diff --git a/bindings/c/src/tests/table_arrow_schema.rs b/bindings/c/src/tests/table_arrow_schema.rs new file mode 100644 index 000000000..336302062 --- /dev/null +++ b/bindings/c/src/tests/table_arrow_schema.rs @@ -0,0 +1,119 @@ +// 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 super::*; +use paimon::spec::{DecimalType, TimestampType, VarBinaryType}; + +unsafe fn export(table: *const paimon_table, schema: &mut FFI_ArrowSchema) -> *mut paimon_error { + paimon_table_export_arrow_schema(table, (schema as *mut FFI_ArrowSchema).cast()) +} + +unsafe fn invalid(error: *mut paimon_error) { + assert!(!error.is_null()); + let code = (*error).code; + paimon_error_free(error); + assert_eq!(code, PaimonErrorCode::InvalidInput as i32); +} + +#[test] +fn exported_table_arrow_schema_preserves_types_and_outlives_table() { + let schema = Schema::builder() + .column("id", DataType::Int(IntType::with_nullable(false))) + .column("name", DataType::VarChar(VarCharType::string_type())) + .column( + "amount", + DataType::Decimal(DecimalType::new(12, 2).unwrap()), + ) + .column("time", DataType::Timestamp(TimestampType::new(6).unwrap())) + .column( + "bytes", + DataType::VarBinary(VarBinaryType::new(100).unwrap()), + ) + .build() + .unwrap(); + unsafe { + let table = wrap_table(Table::new( + memory_file_io(), + Identifier::new("default", "pinned"), + "memory:/no-catalog-or-schema-files".into(), + TableSchema::new(17, &schema), + None, + )); + let mut output = FFI_ArrowSchema::empty(); + assert!(export(table, &mut output).is_null()); + // Reusing live storage is rejected without replacing its contents. + invalid(export(table, &mut output)); + unwrap_table(table); + let actual = ArrowSchema::try_from(&output).unwrap(); + let expected = [ + ("id", ArrowDataType::Int32, false), + ("name", ArrowDataType::Utf8, true), + ("amount", ArrowDataType::Decimal128(12, 2), true), + ( + "time", + ArrowDataType::Timestamp(arrow_schema::TimeUnit::Microsecond, None), + true, + ), + ("bytes", ArrowDataType::Binary, true), + ]; + assert_eq!(actual.fields().len(), expected.len()); + for (id, (field, (name, data_type, nullable))) in + actual.fields().iter().zip(expected).enumerate() + { + assert_eq!(field.name(), name); + assert_eq!(field.data_type(), &data_type); + assert_eq!(field.is_nullable(), nullable); + assert_eq!( + field.metadata().get("PARQUET:field_id"), + Some(&id.to_string()) + ); + } + output.release.unwrap()(&mut output); + assert!(output.release.is_none()); + } +} + +#[test] +fn exported_table_arrow_schema_validates_pointers_and_allows_released_storage() { + unsafe { + let table = wrap_table(Table::new( + memory_file_io(), + Identifier::new("default", "schema"), + "memory:/schema".into(), + simple_table_schema(), + None, + )); + let mut output = FFI_ArrowSchema::empty(); + invalid(export(ptr::null(), &mut output)); + assert!(output.release.is_none()); + invalid(paimon_table_export_arrow_schema(table, ptr::null_mut())); + invalid(export( + &paimon_table { + inner: ptr::null_mut(), + }, + &mut output, + )); + assert!(output.release.is_none()); + for _ in 0..2 { + assert!(export(table, &mut output).is_null()); + assert_eq!(ArrowSchema::try_from(&output).unwrap().fields().len(), 2); + output.release.unwrap()(&mut output); + assert!(output.release.is_none()); + } + unwrap_table(table); + } +} diff --git a/docs/src/c-binding.md b/docs/src/c-binding.md index b32a739ea..83dbc7f76 100644 --- a/docs/src/c-binding.md +++ b/docs/src/c-binding.md @@ -707,6 +707,32 @@ Writing uses a **write-then-commit** flow: 5. Pass the messages to `paimon_table_commit_commit`. 6. Free the messages, writer, committer, and builder. +### Exporting the Table's Write Schema + +Use `paimon_table_export_arrow_schema` to construct batches with the pinned +table's field order, names, nullability and core Arrow types. The function does +not query the catalog or refresh the table schema. Include your Arrow +implementation's C Data Interface declarations for `struct ArrowSchema`: + +```c +struct ArrowSchema schema = {0}; +paimon_error *error = paimon_table_export_arrow_schema(table, &schema); +if (error != NULL) { + paimon_error_free(error); + return -1; +} +/* Import schema into the Arrow implementation and construct matching batches. + An importing consumer may move ownership and set schema.release to NULL. */ +if (schema.release != NULL) schema.release(&schema); +``` + +The table handle is borrowed; the exported schema remains valid after it is +freed. Release the schema through its callback, rather than +`paimon_arrow_batch_free` (which frees a library-owned batch allocation). +The output must be initialized with a null release callback. Reusing a live +schema returns an error without replacing it; release it before reuse. +The core type mapping remains authoritative, including for unsupported types. + ### Cross-process Commit Messages Workers can prepare files and export their ordinary commit messages for an