[Draft] feat(bindings): add C++ streaming facade

.github/workflows/ci.yml

@@ -159,6 +159,34 @@ jobs:
       - name: Build
         run: cargo build --locked --features fulltext,vortex
 
+  cpp:
+    name: cpp (${{ matrix.os }})
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os:
+          - ubuntu-latest
+          - macos-latest
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Install cbindgen
+        uses: taiki-e/install-action@065d6a08a14e61e89fb0a4c10eecdbdef39c7d8e # v2.85.4
+        with:
+          tool: cbindgen@0.29.4
+
+      - name: Configure C++ facade
+        run: >
+          cmake -S bindings/cpp -B target/cpp-ci
+          -DPAIMON_CPP_BUILD_EXAMPLES=ON
+          -DPAIMON_CPP_BUILD_TESTS=ON
+
+      - name: Build C++ facade
+        run: cmake --build target/cpp-ci --parallel 4
+
+      - name: Test C++ facade
+        run: ctest --test-dir target/cpp-ci --output-on-failure
+
   unit:
     runs-on: ${{ matrix.os }}
     strategy:

.gitignore

@@ -16,6 +16,7 @@
 # under the License.
 
 /target
+/bindings/cpp/target/
 .idea
 .vscode
 **/.DS_Store

bindings/c/Cargo.toml

@@ -43,7 +43,9 @@ futures = "0.3"
 arrow = { workspace = true }
 arrow-array = { workspace = true }
 arrow-schema = { workspace = true }
-serde_json = "1.0.120"
+serde_json = { version = "1.0.120", features = ["raw_value"] }
+serde = { version = "1.0", features = ["derive"] }
+sha2 = "0.10"
 async-trait = "0.1.81"
 bytes = "1.7.1"
 

bindings/c/cbindgen.toml

@@ -0,0 +1,24 @@
+language = "C"
+header = """
+// 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.
+"""
+include_guard = "PAIMON_C_H"
+cpp_compat = true
+documentation = true
+usize_is_size_t = true
+style = "both"
+sort_by = "Name"

bindings/c/src/catalog.rs

@@ -15,17 +15,51 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::ffi::c_void;
+use std::ffi::{c_char, c_void};
+use std::panic::{catch_unwind, AssertUnwindSafe};
 use std::sync::Arc;
 
 use paimon::catalog::Identifier;
+use paimon::spec::Schema;
 use paimon::{Catalog, CatalogFactory, Options};
 
 use crate::error::{check_non_null, paimon_error, validate_cstr};
 use crate::result::{paimon_result_catalog_new, paimon_result_get_table};
 use crate::runtime;
 use crate::types::{paimon_catalog, paimon_option, paimon_table};
 
+fn catalog_panic_error(operation: &str) -> *mut paimon_error {
+    paimon_error::new(
+        crate::error::PaimonErrorCode::Unexpected,
+        format!("Rust panic while executing {operation}"),
+    )
+}
+
+fn validate_creation_schema_json(schema_json: &str) -> Result<Schema, *mut paimon_error> {
+    let parsed = serde_json::from_str::<Schema>(schema_json).map_err(|error| {
+        paimon_error::new(
+            crate::error::PaimonErrorCode::InvalidInput,
+            format!("Failed to parse creation schema JSON: {error}"),
+        )
+    })?;
+
+    let mut builder = Schema::builder();
+    for field in parsed.fields() {
+        builder = builder.column_with_description(
+            field.name(),
+            field.data_type().clone(),
+            field.description().map(str::to_string),
+        );
+    }
+    builder
+        .partition_keys(parsed.partition_keys().iter().cloned())
+        .primary_key(parsed.primary_keys().iter().cloned())
+        .options(parsed.options().clone())
+        .comment(parsed.comment().map(str::to_string))
+        .build()
+        .map_err(paimon_error::from_paimon)
+}
+
 /// Create a catalog using CatalogFactory with the given options.
 ///
 /// # Safety
@@ -136,3 +170,85 @@ pub unsafe extern "C" fn paimon_catalog_get_table(
         },
     }
 }
+
+/// Create a table from a logical Paimon `Schema` JSON document.
+///
+/// The input is normalized and validated through `SchemaBuilder` before it is
+/// sent to the catalog. Field IDs in the JSON are therefore treated as input
+/// ordering hints and reassigned canonically from zero.
+///
+/// # Safety
+/// `catalog` and `identifier` must be valid Paimon handles. `schema_json` must
+/// point to a valid null-terminated UTF-8 string.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_catalog_create_table_from_schema_json(
+    catalog: *const paimon_catalog,
+    identifier: *const crate::types::paimon_identifier,
+    schema_json: *const c_char,
+    ignore_if_exists: bool,
+) -> *mut paimon_error {
+    let outcome = catch_unwind(AssertUnwindSafe(|| {
+        if let Err(error) = check_non_null(catalog, "catalog") {
+            return error;
+        }
+        if let Err(error) = check_non_null(identifier, "identifier") {
+            return error;
+        }
+        let schema_json = match validate_cstr(schema_json, "schema_json") {
+            Ok(value) => value,
+            Err(error) => return error,
+        };
+        let schema = match validate_creation_schema_json(&schema_json) {
+            Ok(value) => value,
+            Err(error) => return error,
+        };
+        let catalog_ref = &*((*catalog).inner as *const Arc<dyn Catalog>);
+        let identifier_ref = &*((*identifier).inner as *const Identifier);
+        match runtime().block_on(catalog_ref.create_table(identifier_ref, schema, ignore_if_exists))
+        {
+            Ok(()) => std::ptr::null_mut(),
+            Err(error) => paimon_error::from_paimon(error),
+        }
+    }));
+    outcome.unwrap_or_else(|_| catalog_panic_error("paimon_catalog_create_table_from_schema_json"))
+}
+
+/// Drop a table from the catalog.
+///
+/// # Safety
+/// `catalog` and `identifier` must be valid Paimon handles, or null (returns an
+/// error).
+#[no_mangle]
+pub unsafe extern "C" fn paimon_catalog_drop_table(
+    catalog: *const paimon_catalog,
+    identifier: *const crate::types::paimon_identifier,
+    ignore_if_not_exists: bool,
+) -> *mut paimon_error {
+    let outcome = catch_unwind(AssertUnwindSafe(|| {
+        if let Err(error) = check_non_null(catalog, "catalog") {
+            return error;
+        }
+        if let Err(error) = check_non_null(identifier, "identifier") {
+            return error;
+        }
+        let catalog_ref = &*((*catalog).inner as *const Arc<dyn Catalog>);
+        let identifier_ref = &*((*identifier).inner as *const Identifier);
+        match runtime().block_on(catalog_ref.drop_table(identifier_ref, ignore_if_not_exists)) {
+            Ok(()) => std::ptr::null_mut(),
+            Err(error) => paimon_error::from_paimon(error),
+        }
+    }));
+    outcome.unwrap_or_else(|_| catalog_panic_error("paimon_catalog_drop_table"))
+}
+
+const _: unsafe extern "C" fn(
+    *const paimon_catalog,
+    *const crate::types::paimon_identifier,
+    *const c_char,
+    bool,
+) -> *mut paimon_error = paimon_catalog_create_table_from_schema_json;
+const _: unsafe extern "C" fn(
+    *const paimon_catalog,
+    *const crate::types::paimon_identifier,
+    bool,
+) -> *mut paimon_error = paimon_catalog_drop_table;

bindings/c/src/error.rs

@@ -19,15 +19,25 @@ use std::ffi::{c_char, CStr};
 
 use crate::types::paimon_bytes;
 
+pub const PAIMON_ERROR_UNEXPECTED: i32 = 0;
+pub const PAIMON_ERROR_UNSUPPORTED: i32 = 1;
+pub const PAIMON_ERROR_NOT_FOUND: i32 = 2;
+pub const PAIMON_ERROR_ALREADY_EXISTS: i32 = 3;
+pub const PAIMON_ERROR_INVALID_INPUT: i32 = 4;
+pub const PAIMON_ERROR_IO: i32 = 5;
+pub const PAIMON_ERROR_OUT_OF_RANGE: i32 = 6;
+
 /// Error codes for paimon C API.
 #[repr(i32)]
 pub enum PaimonErrorCode {
-    Unexpected = 0,
-    Unsupported = 1,
-    NotFound = 2,
-    AlreadyExists = 3,
-    InvalidInput = 4,
-    IoError = 5,
+    Unexpected = PAIMON_ERROR_UNEXPECTED,
+    Unsupported = PAIMON_ERROR_UNSUPPORTED,
+    NotFound = PAIMON_ERROR_NOT_FOUND,
+    AlreadyExists = PAIMON_ERROR_ALREADY_EXISTS,
+    InvalidInput = PAIMON_ERROR_INVALID_INPUT,
+    IoError = PAIMON_ERROR_IO,
+    /// A requested streaming checkpoint or snapshot is no longer readable.
+    OutOfRange = PAIMON_ERROR_OUT_OF_RANGE,
 }
 
 /// C-compatible error type.
@@ -53,9 +63,18 @@ impl paimon_error {
             paimon::Error::TableNotExist { .. }
             | paimon::Error::DatabaseNotExist { .. }
             | paimon::Error::ColumnNotExist { .. } => PaimonErrorCode::NotFound,
+            paimon::Error::SnapshotNotExist { .. } => PaimonErrorCode::OutOfRange,
             paimon::Error::TableAlreadyExist { .. }
             | paimon::Error::DatabaseAlreadyExist { .. }
             | paimon::Error::ColumnAlreadyExist { .. } => PaimonErrorCode::AlreadyExists,
+            paimon::Error::DataInvalid { message, .. }
+                if message.contains("snapshot")
+                    && (message.contains("expired")
+                        || message.contains("out of range")
+                        || message.contains("too large")) =>
+            {
+                PaimonErrorCode::OutOfRange
+            }
             paimon::Error::ConfigInvalid { .. }
             | paimon::Error::DataTypeInvalid { .. }
             | paimon::Error::DataInvalid { .. }

bindings/c/src/lib.rs

@@ -25,6 +25,7 @@ mod error;
 mod file_io;
 mod identifier;
 mod result;
+mod stream;
 mod table;
 #[cfg(test)]
 mod tests;

bindings/c/src/result.rs

@@ -146,6 +146,18 @@ pub struct paimon_result_prepare_commit {
     pub error: *mut paimon_error,
 }
 
+#[repr(C)]
+pub struct paimon_result_prepared_commit {
+    pub prepared: *mut paimon_prepared_commit,
+    pub error: *mut paimon_error,
+}
+
+#[repr(C)]
+pub struct paimon_result_bytes {
+    pub bytes: paimon_bytes,
+    pub error: *mut paimon_error,
+}
+
 #[repr(C)]
 pub struct paimon_result_postpone_fixed_bucket_write_builder {
     pub write_builder: *mut paimon_postpone_fixed_bucket_write_builder,