Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions bindings/c/src/table.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::<FFI_ArrowSchema>();
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,
Expand Down
18 changes: 17 additions & 1 deletion bindings/c/src/tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
Expand Down Expand Up @@ -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;
Expand Down
119 changes: 119 additions & 0 deletions bindings/c/src/tests/table_arrow_schema.rs
Original file line number Diff line number Diff line change
@@ -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);
}
}
26 changes: 26 additions & 0 deletions docs/src/c-binding.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading