From 1bb8e605941e5c29e6c2f04fe616edb7c39f77f1 Mon Sep 17 00:00:00 2001 From: "jianjian.xie" Date: Mon, 31 Aug 2026 12:53:53 -0700 Subject: [PATCH] refactor: centralize async FFI completion handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Summary: Intent: - Make future-to-callback translation reusable for additional async C APIs. - Detect accidental layout drift between the Rust ABI and the public C header. Changes: - Add a generic async bridge that maps success, Lance errors, and panics to exactly one completion. - Route asynchronous scanner materialization through the shared completion abstraction. - Compare public enum sizes and struct layouts across compiled C and Rust definitions. --- Generated by the 🪄 pr-create skill in devexp-agent-marketplace --- src/async_bridge.rs | 214 +++++++++++++++++++++++++++++++++ src/async_dispatcher.rs | 44 +++++++ src/lib.rs | 1 + src/scanner.rs | 170 ++++++-------------------- tests/compile_and_run_test.rs | 217 ++++++++++++++++++++++++++++++++++ tests/cpp/abi_layout.c | 122 +++++++++++++++++++ 6 files changed, 636 insertions(+), 132 deletions(-) create mode 100644 src/async_bridge.rs create mode 100644 tests/cpp/abi_layout.c diff --git a/src/async_bridge.rs b/src/async_bridge.rs new file mode 100644 index 0000000..3852e60 --- /dev/null +++ b/src/async_bridge.rs @@ -0,0 +1,214 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The Lance Authors + +//! Reusable bridge from Lance futures to C callback completions. + +use std::ffi::c_void; +use std::future::Future; + +use futures::FutureExt; + +use crate::async_dispatcher::Completion; +use crate::error::{LanceErrorCode, error_code_from_lance, panic_payload_message, swallow_unwind}; +use crate::runtime::RT; + +/// Spawn a Lance future and translate its terminal state into exactly one C +/// completion. +/// +/// `into_result` converts the successful Rust value into an operation-specific +/// C result pointer. `on_panic` lets stateful callers poison or invalidate +/// state before the panic completion is delivered. +pub(crate) fn spawn_lance_future( + completion: Completion, + future: F, + into_result: S, + on_panic: P, +) where + F: Future> + Send + 'static, + T: Send + 'static, + S: FnOnce(T) -> *mut c_void + Send + 'static, + P: FnOnce() + Send + 'static, +{ + RT.spawn(async move { + let completion_on_panic = completion; + let outcome = std::panic::AssertUnwindSafe(async move { + match future.await { + Ok(value) => completion.succeed(into_result(value)), + Err(err) => completion.fail(error_code_from_lance(&err), err.to_string()), + } + }) + .catch_unwind() + .await; + + if let Err(payload) = outcome { + swallow_unwind("async FFI panic hook", on_panic); + completion_on_panic.fail( + LanceErrorCode::Panic, + format!("panic in FFI call: {}", panic_payload_message(&*payload)), + ); + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::async_dispatcher::LanceCallback; + use crate::error::{lance_free_string, lance_last_error_code, lance_last_error_message}; + use std::ffi::CStr; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::{Arc, mpsc}; + use std::time::Duration; + + #[derive(Debug)] + struct Observation { + status: i32, + result: *mut c_void, + code: LanceErrorCode, + message: Option, + } + + unsafe impl Send for Observation {} + + unsafe extern "C" fn observe(ctx: *mut c_void, status: i32, result: *mut c_void) { + let tx = unsafe { &*(ctx as *const mpsc::Sender) }; + let code = lance_last_error_code(); + let message_ptr = lance_last_error_message(); + let message = if message_ptr.is_null() { + None + } else { + let message = unsafe { CStr::from_ptr(message_ptr) } + .to_string_lossy() + .into_owned(); + unsafe { lance_free_string(message_ptr) }; + Some(message) + }; + tx.send(Observation { + status, + result, + code, + message, + }) + .unwrap(); + } + + fn completion() -> ( + Completion, + mpsc::Receiver, + *mut mpsc::Sender, + ) { + let (tx, rx) = mpsc::channel(); + let ctx = Box::into_raw(Box::new(tx)); + let callback: LanceCallback = observe; + let completion = unsafe { Completion::new(callback, ctx.cast()) }; + (completion, rx, ctx) + } + + fn receive(rx: &mpsc::Receiver) -> Observation { + rx.recv_timeout(Duration::from_secs(5)) + .expect("async bridge must deliver completion") + } + + #[test] + fn success_is_converted_and_delivered() { + let (completion, rx, ctx) = completion(); + spawn_lance_future( + completion, + async { Ok::<_, lance_core::Error>(42_u64) }, + |value| Box::into_raw(Box::new(value)).cast(), + || panic!("success must not invoke on_panic"), + ); + + let observation = receive(&rx); + assert_eq!(observation.status, 0); + assert_eq!(observation.code, LanceErrorCode::Ok); + assert!(observation.message.is_none()); + assert_eq!( + unsafe { *Box::from_raw(observation.result.cast::()) }, + 42 + ); + unsafe { drop(Box::from_raw(ctx)) }; + } + + #[test] + fn lance_error_is_delivered_without_running_converter() { + let (completion, rx, ctx) = completion(); + spawn_lance_future::<_, (), _, _>( + completion, + async { + Err(lance_core::Error::invalid_input_source( + "invalid async input".into(), + )) + }, + |_| panic!("error must not run success converter"), + || panic!("ordinary error must not invoke on_panic"), + ); + + let observation = receive(&rx); + assert_eq!(observation.status, -1); + assert!(observation.result.is_null()); + assert_eq!(observation.code, LanceErrorCode::InvalidArgument); + assert!( + observation + .message + .as_deref() + .is_some_and(|message| message.contains("invalid async input")) + ); + unsafe { drop(Box::from_raw(ctx)) }; + } + + #[test] + fn panic_runs_hook_and_delivers_panic_error() { + let (completion, rx, ctx) = completion(); + let panicked = Arc::new(AtomicBool::new(false)); + let panicked_in_hook = Arc::clone(&panicked); + spawn_lance_future::<_, (), _, _>( + completion, + async { + panic!("async bridge panic"); + #[allow(unreachable_code)] + Ok(()) + }, + |_| std::ptr::null_mut(), + move || panicked_in_hook.store(true, Ordering::SeqCst), + ); + + let observation = receive(&rx); + assert!(panicked.load(Ordering::SeqCst)); + assert_eq!(observation.status, -1); + assert_eq!(observation.code, LanceErrorCode::Panic); + assert!( + observation + .message + .as_deref() + .is_some_and(|message| message.contains("async bridge panic")) + ); + unsafe { drop(Box::from_raw(ctx)) }; + } + + #[test] + fn panicking_hook_does_not_suppress_completion() { + let (completion, rx, ctx) = completion(); + spawn_lance_future::<_, (), _, _>( + completion, + async { + panic!("original async panic"); + #[allow(unreachable_code)] + Ok(()) + }, + |_| std::ptr::null_mut(), + || panic!("panic hook also panicked"), + ); + + let observation = receive(&rx); + assert_eq!(observation.status, -1); + assert_eq!(observation.code, LanceErrorCode::Panic); + assert!( + observation + .message + .as_deref() + .is_some_and(|message| message.contains("original async panic")) + ); + unsafe { drop(Box::from_raw(ctx)) }; + } +} diff --git a/src/async_dispatcher.rs b/src/async_dispatcher.rs index 0112ed5..1ddb819 100644 --- a/src/async_dispatcher.rs +++ b/src/async_dispatcher.rs @@ -19,6 +19,50 @@ use crate::error::{LanceErrorCode, clear_last_error, panic_payload_message, set_ /// - `result`: operation-specific result pointer (e.g., `*mut ArrowArrayStream`) pub type LanceCallback = unsafe extern "C" fn(ctx: *mut c_void, status: i32, result: *mut c_void); +/// A copyable async completion endpoint whose raw context is owned by the host. +/// +/// This centralizes the unsafe `void *` transport and status/error mapping used +/// by spawned FFI futures. The caller must keep `callback_ctx` valid until the +/// callback returns. +#[derive(Clone, Copy)] +pub(crate) struct Completion { + callback: LanceCallback, + callback_ctx: *mut c_void, +} + +// Safety: construction requires the caller to uphold the public callback +// contract: the callback is thread-safe and callback_ctx remains valid until +// completion delivery returns. +unsafe impl Send for Completion {} + +impl Completion { + /// Construct a completion endpoint from a validated callback and context. + /// + /// # Safety + /// `callback_ctx` must remain valid until `callback` returns, and + /// `callback` must be safe to invoke from any completion-delivery thread. + pub(crate) unsafe fn new(callback: LanceCallback, callback_ctx: *mut c_void) -> Self { + Self { + callback, + callback_ctx, + } + } + + pub(crate) fn succeed(self, result: *mut c_void) { + dispatch_callback(self.callback, self.callback_ctx, 0, result, None); + } + + pub(crate) fn fail(self, code: LanceErrorCode, message: impl Into) { + dispatch_callback( + self.callback, + self.callback_ctx, + -1, + std::ptr::null_mut(), + Some((code, message.into())), + ); + } +} + // Safety: LanceCallback is a C function pointer (Send by definition for FFI). // The ctx pointer is transferred to the dispatcher thread which calls the callback. unsafe impl Send for DispatcherMessage {} diff --git a/src/lib.rs b/src/lib.rs index 4d54641..ae0523a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ compile_error!( mod add_columns; mod alter_columns; +mod async_bridge; mod async_dispatcher; mod batch; mod compact; diff --git a/src/scanner.rs b/src/scanner.rs index 536d72e..3702009 100644 --- a/src/scanner.rs +++ b/src/scanner.rs @@ -13,7 +13,7 @@ use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker}; use arrow::ffi_stream::FFI_ArrowArrayStream; use arrow_schema::SchemaRef; use datafusion::physical_plan::ExecutionPlan; -use futures::{FutureExt, Stream, StreamExt}; +use futures::{Stream, StreamExt}; use lance::Dataset; use lance::dataset::scanner::{ DatasetRecordBatchStream, ExecutionStatsCallback, ExecutionSummaryCounts, @@ -25,7 +25,8 @@ use lance_io::stream::RecordBatchStream; use lance_table::format::IndexMetadata; use uuid::Uuid; -use crate::async_dispatcher::{self, LanceCallback}; +use crate::async_bridge::spawn_lance_future; +use crate::async_dispatcher::{Completion, LanceCallback}; use crate::batch::LanceBatch; use crate::dataset::LanceDataset; use crate::error::{ @@ -1085,9 +1086,10 @@ pub unsafe extern "C" fn lance_scanner_scan_async( set_last_error(LanceErrorCode::InvalidArgument, "callback must not be NULL"); return; }; + let completion = unsafe { Completion::new(callback, callback_ctx) }; unsafe { - scan_async_guarded(scanner, callback, callback_ctx, |s, cb, ctx| { - scan_async_setup(s, cb, ctx) + scan_async_guarded(scanner, completion, |scanner, completion| { + scan_async_setup(scanner, completion); }); } } @@ -1100,13 +1102,12 @@ pub unsafe extern "C" fn lance_scanner_scan_async( /// hook. /// /// # Safety -/// `scanner` must be NULL or a valid scanner handle; `callback` and -/// `callback_ctx` follow the same contract as `lance_scanner_scan_async`. +/// `scanner` must be NULL or a valid scanner handle; `completion` must follow +/// the callback and context contract of `lance_scanner_scan_async`. unsafe fn scan_async_guarded( scanner: *const LanceScanner, - callback: LanceCallback, - callback_ctx: *mut c_void, - setup: impl FnOnce(*const LanceScanner, LanceCallback, *mut c_void), + completion: Completion, + setup: impl FnOnce(*const LanceScanner, Completion), ) { // Capture the poison flag BEFORE setup runs: a panic during setup must // still be able to poison the handle it never finished configuring. @@ -1115,9 +1116,8 @@ unsafe fn scan_async_guarded( } else { Some(unsafe { &*scanner }.poison_flag()) }; - let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { - setup(scanner, callback, callback_ctx) - })); + let outcome = + std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| setup(scanner, completion))); if let Err(payload) = outcome { if let Some(flag) = poison_flag { flag.store(true, Ordering::SeqCst); @@ -1127,15 +1127,9 @@ unsafe fn scan_async_guarded( // entry point would abort the host, which is what this guard exists // to prevent. swallow_unwind("lance_scanner_scan_async panic report", || { - async_dispatcher::dispatch_callback( - callback, - callback_ctx, - -1, - ptr::null_mut(), - Some(( - LanceErrorCode::Panic, - format!("panic in FFI call: {}", panic_payload_message(&*payload)), - )), + completion.fail( + LanceErrorCode::Panic, + format!("panic in FFI call: {}", panic_payload_message(&*payload)), ); }); } @@ -1148,23 +1142,13 @@ unsafe fn scan_async_guarded( /// /// # Safety /// `scanner` must be NULL or a valid scanner handle (checked first). -unsafe fn scan_async_setup( - scanner: *const LanceScanner, - callback: LanceCallback, - callback_ctx: *mut c_void, -) { +unsafe fn scan_async_setup(scanner: *const LanceScanner, completion: Completion) { // Validation-time failures happen before scan_async returns, so they keep // setting the caller thread's TLS AND carry the error inside the dispatch // message for the callback thread to observe (issue #61). if scanner.is_null() { set_last_error(LanceErrorCode::InvalidArgument, "scanner is NULL"); - async_dispatcher::dispatch_callback( - callback, - callback_ctx, - -1, - ptr::null_mut(), - Some((LanceErrorCode::InvalidArgument, "scanner is NULL".into())), - ); + completion.fail(LanceErrorCode::InvalidArgument, "scanner is NULL"); return; } @@ -1176,15 +1160,9 @@ unsafe fn scan_async_setup( LanceErrorCode::Panic, "scanner is poisoned by an earlier panic", ); - async_dispatcher::dispatch_callback( - callback, - callback_ctx, - -1, - ptr::null_mut(), - Some(( - LanceErrorCode::Panic, - "scanner is poisoned by an earlier panic".into(), - )), + completion.fail( + LanceErrorCode::Panic, + "scanner is poisoned by an earlier panic", ); return; } @@ -1193,102 +1171,31 @@ unsafe fn scan_async_setup( Ok(sc) => sc, Err(err) => { set_lance_error(&err); - async_dispatcher::dispatch_callback( - callback, - callback_ctx, - -1, - ptr::null_mut(), - Some((error_code_from_lance(&err), err.to_string())), - ); + completion.fail(error_code_from_lance(&err), err.to_string()); return; } }; let handle = RT.handle().clone(); - // Wrap non-Send raw pointers for the async task. - // Safety: The C caller guarantees callback_ctx remains valid until callback fires. - #[derive(Clone, Copy)] - struct SendCallback { - callback: LanceCallback, - ctx: *mut c_void, - } - unsafe impl Send for SendCallback {} - - impl SendCallback { - fn dispatch( - &self, - status: i32, - result: *mut c_void, - error: Option<(LanceErrorCode, String)>, - ) { - async_dispatcher::dispatch_callback(self.callback, self.ctx, status, result, error); - } - } - - let send_cb = SendCallback { - callback, - ctx: callback_ctx, - }; - // Shared poison flag moved into the task: the GuardedReader below flips // it if a panic is caught during the consumer's later `get_next` calls. let poisoned = s.poison_flag(); - - RT.spawn(async move { - // Copies kept outside the inner future (which consumes the originals) - // so the panic arm below can still poison the handle and report. - let poisoned_on_panic = Arc::clone(&poisoned); - let send_cb_on_panic = send_cb; - // The whole task body runs under catch_unwind (issue #61): a panic - // here would otherwise die as an unobserved JoinError — the callback - // would never fire and the C caller would hang forever waiting for a - // completion that never arrives. - let outcome = std::panic::AssertUnwindSafe(async move { - let result = built_scanner.try_into_stream().await; - match result { - Ok(stream) => { - // Guard the exported stream at the reader level (issue - // #61): a mid-iteration panic — including a - // `Handle::block_on` panic on a consumer thread driving a - // Tokio runtime — becomes one terminal error item per - // the Arrow C stream error contract instead of unwinding - // out of arrow-rs's `get_next`, and cleanup panics on the - // `release` path are contained. - let schema = stream.schema(); - let reader = GuardedReader::new(stream, schema, handle, poisoned); - let ffi_stream = FFI_ArrowArrayStream::new(Box::new(reader)); - let ptr = Box::into_raw(Box::new(ffi_stream)); - send_cb.dispatch(0, ptr as *mut c_void, None); - } - Err(err) => { - // Runs on a Tokio worker AFTER scan_async returned: - // setting this thread's TLS would be invisible to - // everyone, so the error rides inside the dispatch - // message and the dispatcher installs it on the - // callback thread instead (issue #61). - send_cb.dispatch( - -1, - std::ptr::null_mut(), - Some((error_code_from_lance(&err), err.to_string())), - ); - } - } - }) - .catch_unwind() - .await; - if let Err(payload) = outcome { - poisoned_on_panic.store(true, Ordering::SeqCst); - send_cb_on_panic.dispatch( - -1, - std::ptr::null_mut(), - Some(( - LanceErrorCode::Panic, - format!("panic in FFI call: {}", panic_payload_message(&*payload)), - )), - ); - } - }); + let poisoned_on_panic = Arc::clone(&poisoned); + spawn_lance_future( + completion, + built_scanner.try_into_stream(), + move |stream| { + // Guard the exported stream at the reader level (issue #61): a + // mid-iteration panic becomes one terminal Arrow stream error, + // and cleanup panics on the release path are contained. + let schema = stream.schema(); + let reader = GuardedReader::new(stream, schema, handle, poisoned); + let ffi_stream = FFI_ArrowArrayStream::new(Box::new(reader)); + Box::into_raw(Box::new(ffi_stream)).cast() + }, + move || poisoned_on_panic.store(true, Ordering::SeqCst), + ); } /// Release the heap-allocated Arrow stream container returned through a @@ -2500,11 +2407,10 @@ mod tests { let (tx, rx) = std::sync::mpsc::channel::(); let ctx = Box::into_raw(Box::new(tx)) as *mut c_void; + let completion = unsafe { Completion::new(record_setup_panic, ctx) }; unsafe { - scan_async_guarded(scanner, record_setup_panic, ctx, |_, _, _| { - panic!("injected setup panic") - }); + scan_async_guarded(scanner, completion, |_, _| panic!("injected setup panic")); } assert!( diff --git a/tests/compile_and_run_test.rs b/tests/compile_and_run_test.rs index b419ac9..fe01b03 100644 --- a/tests/compile_and_run_test.rs +++ b/tests/compile_and_run_test.rs @@ -15,6 +15,7 @@ use std::path::{Path, PathBuf}; use std::process::Command; use std::sync::Arc; +use std::{collections::BTreeSet, mem}; use arrow_array::{FixedSizeListArray, Float32Array, Int32Array, RecordBatch, StringArray}; use arrow_schema::{DataType, Field, Schema}; @@ -174,6 +175,26 @@ fn compile_cpp_test(source: &Path, output: &Path, include_dir: &Path, lib_path: .success() } +/// Compile a standalone C source file that only inspects the public header. +fn compile_c_header_test(source: &Path, output: &Path, include_dir: &Path) -> bool { + let status = Command::new("clang") + .args([ + "-std=c11", + "-Wall", + "-Wextra", + "-Werror", + "-o", + output.to_str().unwrap(), + source.to_str().unwrap(), + &format!("-I{}", include_dir.display()), + ]) + .status(); + + status + .expect("C compiler is required for this ignored test") + .success() +} + /// Run a compiled test binary with the source dataset URI and a destination URI /// for the write test. The destination path must not pre-exist. fn run_test_binary(binary: &Path, dataset_uri: &str, write_uri: &str) { @@ -249,3 +270,199 @@ fn test_cpp_compilation_and_execution() { run_test_binary(&binary, &dataset_uri, &write_uri); } + +#[test] +#[ignore = "requires C compiler (clang); run with: cargo test --test compile_and_run_test -- --ignored"] +fn test_c_and_rust_abi_layouts_match() { + macro_rules! record_type { + ($records:ident, $type:ty) => { + $records.insert(format!( + "T|{}|{}|{}", + stringify!($type).rsplit("::").next().unwrap(), + mem::size_of::<$type>(), + mem::align_of::<$type>() + )); + }; + } + + macro_rules! record_field { + ($records:ident, $type:ty, $field:ident) => { + $records.insert(format!( + "F|{}.{}|{}", + stringify!($type).rsplit("::").next().unwrap(), + stringify!($field), + mem::offset_of!($type, $field) + )); + }; + } + + macro_rules! record_struct { + ($records:ident, $type:ty, [$($field:ident),+ $(,)?]) => { + record_type!($records, $type); + $(record_field!($records, $type, $field);)+ + }; + } + + let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + let include_dir = manifest_dir.join("include"); + let source = manifest_dir.join("tests").join("cpp").join("abi_layout.c"); + let build_dir = tempfile::tempdir().unwrap(); + let binary = build_dir.path().join("abi_layout"); + assert!( + compile_c_header_test(&source, &binary, &include_dir), + "C ABI layout test compilation failed" + ); + + let output = Command::new(&binary) + .output() + .expect("failed to run C ABI layout test"); + assert!( + output.status.success(), + "C ABI layout test failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + let c_records = String::from_utf8(output.stdout) + .expect("C ABI layout output must be UTF-8") + .lines() + .map(str::to_owned) + .collect::>(); + + let mut rust_records = BTreeSet::new(); + record_type!(rust_records, lance_c::LanceErrorCode); + record_type!(rust_records, lance_c::LanceVectorIndexType); + record_type!(rust_records, lance_c::LanceScalarIndexType); + record_type!(rust_records, lance_c::LanceMetricType); + record_type!(rust_records, lance_c::LanceDataType); + record_type!(rust_records, lance_c::LanceMergeWhenMatched); + record_type!(rust_records, lance_c::LanceMergeWhenNotMatched); + record_type!(rust_records, lance_c::LanceMergeWhenNotMatchedBySource); + record_type!(rust_records, lance_c::LanceColumnNullableMode); + record_type!(rust_records, lance_c::LanceScanMetricKind); + record_type!(rust_records, lance_c::LancePollStatus); + record_type!(rust_records, lance_c::LanceIndexSegmentBuildMode); + record_type!(rust_records, lance_c::LanceFtsCoverageMode); + record_type!(rust_records, lance_c::LanceWriteMode); + + record_struct!( + rust_records, + lance_c::LanceVectorIndexParams, + [ + index_type, + metric, + num_partitions, + num_sub_vectors, + num_bits, + max_iterations, + hnsw_m, + hnsw_ef_construction, + sample_rate, + ] + ); + record_struct!( + rust_records, + lance_c::LanceMergeInsertParams, + [ + when_matched, + when_matched_expr, + when_not_matched, + when_not_matched_by_source, + when_not_matched_by_source_expr, + ] + ); + record_struct!( + rust_records, + lance_c::LanceMergeInsertResult, + [num_inserted_rows, num_updated_rows, num_deleted_rows] + ); + record_struct!( + rust_records, + lance_c::LanceCompactionOptions, + [ + target_rows_per_fragment, + max_rows_per_group, + max_bytes_per_file, + num_threads, + batch_size, + ] + ); + record_struct!( + rust_records, + lance_c::LanceCompactionMetrics, + [ + fragments_removed, + fragments_added, + files_removed, + files_added, + ] + ); + record_struct!( + rust_records, + lance_c::LanceColumnAlteration, + [path, rename, nullable_mode, data_type] + ); + record_struct!(rust_records, lance_c::LanceSqlColumn, [name, expression]); + record_struct!( + rust_records, + lance_c::LanceScanMetric, + [name, name_len, kind, value] + ); + record_struct!( + rust_records, + lance_c::LanceScanStatistics, + [ + iops, + requests, + bytes_read, + indices_loaded, + index_partitions_loaded, + index_comparisons, + metrics, + metrics_len, + ] + ); + record_struct!( + rust_records, + lance_c::LanceIndexSegmentBuildOptions, + [ + fragment_ids, + fragment_count, + index_uuid, + ivf_centroids, + ivf_centroids_schema, + pq_codebook, + pq_codebook_schema, + mode, + ] + ); + record_struct!( + rust_records, + lance_c::LanceVectorIndexSegmentParams, + [ + index_type, + metric, + num_partitions, + num_sub_vectors, + num_bits, + max_iterations, + hnsw_m, + hnsw_ef_construction, + sample_rate, + ] + ); + record_struct!( + rust_records, + lance_c::LanceWriteParams, + [ + max_rows_per_file, + max_rows_per_group, + max_bytes_per_file, + data_storage_version, + enable_stable_row_ids, + ] + ); + + assert_eq!( + c_records, rust_records, + "public C and Rust ABI layouts diverged" + ); +} diff --git a/tests/cpp/abi_layout.c b/tests/cpp/abi_layout.c new file mode 100644 index 0000000..58fc284 --- /dev/null +++ b/tests/cpp/abi_layout.c @@ -0,0 +1,122 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The Lance Authors + +#include +#include + +#include "lance/lance.h" + +#define PRINT_TYPE(type) \ + printf("T|%s|%zu|%zu\n", #type, sizeof(type), _Alignof(type)) + +#define PRINT_FIELD(type, field) \ + printf("F|%s.%s|%zu\n", #type, #field, offsetof(type, field)) + +int main(void) { + PRINT_TYPE(LanceErrorCode); + PRINT_TYPE(LanceVectorIndexType); + PRINT_TYPE(LanceScalarIndexType); + PRINT_TYPE(LanceMetricType); + PRINT_TYPE(LanceDataType); + PRINT_TYPE(LanceMergeWhenMatched); + PRINT_TYPE(LanceMergeWhenNotMatched); + PRINT_TYPE(LanceMergeWhenNotMatchedBySource); + PRINT_TYPE(LanceColumnNullableMode); + PRINT_TYPE(LanceScanMetricKind); + PRINT_TYPE(LancePollStatus); + PRINT_TYPE(LanceIndexSegmentBuildMode); + PRINT_TYPE(LanceFtsCoverageMode); + PRINT_TYPE(LanceWriteMode); + + PRINT_TYPE(LanceVectorIndexParams); + PRINT_FIELD(LanceVectorIndexParams, index_type); + PRINT_FIELD(LanceVectorIndexParams, metric); + PRINT_FIELD(LanceVectorIndexParams, num_partitions); + PRINT_FIELD(LanceVectorIndexParams, num_sub_vectors); + PRINT_FIELD(LanceVectorIndexParams, num_bits); + PRINT_FIELD(LanceVectorIndexParams, max_iterations); + PRINT_FIELD(LanceVectorIndexParams, hnsw_m); + PRINT_FIELD(LanceVectorIndexParams, hnsw_ef_construction); + PRINT_FIELD(LanceVectorIndexParams, sample_rate); + + PRINT_TYPE(LanceMergeInsertParams); + PRINT_FIELD(LanceMergeInsertParams, when_matched); + PRINT_FIELD(LanceMergeInsertParams, when_matched_expr); + PRINT_FIELD(LanceMergeInsertParams, when_not_matched); + PRINT_FIELD(LanceMergeInsertParams, when_not_matched_by_source); + PRINT_FIELD(LanceMergeInsertParams, when_not_matched_by_source_expr); + + PRINT_TYPE(LanceMergeInsertResult); + PRINT_FIELD(LanceMergeInsertResult, num_inserted_rows); + PRINT_FIELD(LanceMergeInsertResult, num_updated_rows); + PRINT_FIELD(LanceMergeInsertResult, num_deleted_rows); + + PRINT_TYPE(LanceCompactionOptions); + PRINT_FIELD(LanceCompactionOptions, target_rows_per_fragment); + PRINT_FIELD(LanceCompactionOptions, max_rows_per_group); + PRINT_FIELD(LanceCompactionOptions, max_bytes_per_file); + PRINT_FIELD(LanceCompactionOptions, num_threads); + PRINT_FIELD(LanceCompactionOptions, batch_size); + + PRINT_TYPE(LanceCompactionMetrics); + PRINT_FIELD(LanceCompactionMetrics, fragments_removed); + PRINT_FIELD(LanceCompactionMetrics, fragments_added); + PRINT_FIELD(LanceCompactionMetrics, files_removed); + PRINT_FIELD(LanceCompactionMetrics, files_added); + + PRINT_TYPE(LanceColumnAlteration); + PRINT_FIELD(LanceColumnAlteration, path); + PRINT_FIELD(LanceColumnAlteration, rename); + PRINT_FIELD(LanceColumnAlteration, nullable_mode); + PRINT_FIELD(LanceColumnAlteration, data_type); + + PRINT_TYPE(LanceSqlColumn); + PRINT_FIELD(LanceSqlColumn, name); + PRINT_FIELD(LanceSqlColumn, expression); + + PRINT_TYPE(LanceScanMetric); + PRINT_FIELD(LanceScanMetric, name); + PRINT_FIELD(LanceScanMetric, name_len); + PRINT_FIELD(LanceScanMetric, kind); + PRINT_FIELD(LanceScanMetric, value); + + PRINT_TYPE(LanceScanStatistics); + PRINT_FIELD(LanceScanStatistics, iops); + PRINT_FIELD(LanceScanStatistics, requests); + PRINT_FIELD(LanceScanStatistics, bytes_read); + PRINT_FIELD(LanceScanStatistics, indices_loaded); + PRINT_FIELD(LanceScanStatistics, index_partitions_loaded); + PRINT_FIELD(LanceScanStatistics, index_comparisons); + PRINT_FIELD(LanceScanStatistics, metrics); + PRINT_FIELD(LanceScanStatistics, metrics_len); + + PRINT_TYPE(LanceIndexSegmentBuildOptions); + PRINT_FIELD(LanceIndexSegmentBuildOptions, fragment_ids); + PRINT_FIELD(LanceIndexSegmentBuildOptions, fragment_count); + PRINT_FIELD(LanceIndexSegmentBuildOptions, index_uuid); + PRINT_FIELD(LanceIndexSegmentBuildOptions, ivf_centroids); + PRINT_FIELD(LanceIndexSegmentBuildOptions, ivf_centroids_schema); + PRINT_FIELD(LanceIndexSegmentBuildOptions, pq_codebook); + PRINT_FIELD(LanceIndexSegmentBuildOptions, pq_codebook_schema); + PRINT_FIELD(LanceIndexSegmentBuildOptions, mode); + + PRINT_TYPE(LanceVectorIndexSegmentParams); + PRINT_FIELD(LanceVectorIndexSegmentParams, index_type); + PRINT_FIELD(LanceVectorIndexSegmentParams, metric); + PRINT_FIELD(LanceVectorIndexSegmentParams, num_partitions); + PRINT_FIELD(LanceVectorIndexSegmentParams, num_sub_vectors); + PRINT_FIELD(LanceVectorIndexSegmentParams, num_bits); + PRINT_FIELD(LanceVectorIndexSegmentParams, max_iterations); + PRINT_FIELD(LanceVectorIndexSegmentParams, hnsw_m); + PRINT_FIELD(LanceVectorIndexSegmentParams, hnsw_ef_construction); + PRINT_FIELD(LanceVectorIndexSegmentParams, sample_rate); + + PRINT_TYPE(LanceWriteParams); + PRINT_FIELD(LanceWriteParams, max_rows_per_file); + PRINT_FIELD(LanceWriteParams, max_rows_per_group); + PRINT_FIELD(LanceWriteParams, max_bytes_per_file); + PRINT_FIELD(LanceWriteParams, data_storage_version); + PRINT_FIELD(LanceWriteParams, enable_stable_row_ids); + + return 0; +}