review fixes

Author: myrrc

vortex-ffi/build.rs

@@ -8,6 +8,7 @@ use std::process::Command;
 use std::process::exit;
 
 fn main() {
+    println!("cargo::rustc-check-cfg=cfg(vortex_nightly)");
     println!("cargo:rerun-if-changed=src");
     println!("cargo:rerun-if-changed=cbindgen.toml");
     println!("cargo:rerun-if-changed=Cargo.toml");

vortex-ffi/cbindgen.toml

@@ -51,8 +51,8 @@ struct ArrowArrayStream {
     void (*release)(struct ArrowArrayStream*);
     void* private_data;
 };
-typedef ArrowSchema FFI_ArrowSchema;
-typedef ArrowArrayStream FFI_ArrowArrayStream;
+typedef struct ArrowSchema FFI_ArrowSchema;
+typedef struct ArrowArrayStream FFI_ArrowArrayStream;
 #endif
 """
 

vortex-ffi/src/scan.rs

@@ -6,7 +6,6 @@ use std::ffi::c_int;
 use std::ops::Range;
 use std::ptr;
 use std::sync::Arc;
-use std::sync::Mutex;
 
 use arrow_array::RecordBatch;
 use arrow_array::cast::AsArray;
@@ -43,12 +42,11 @@ use crate::error::vx_error;
 use crate::expression::vx_expression;
 use crate::session::vx_session;
 
-pub enum VxScanState {
+pub enum VxScan {
     Pending(Box<dyn DataSourceScan>),
     Started(PartitionStream),
     Finished,
 }
-pub type VxScan = Mutex<VxScanState>;
 crate::box_wrapper!(VxScan, vx_scan);
 
 pub enum VxPartitionScan {
@@ -233,7 +231,7 @@ pub unsafe extern "C-unwind" fn vx_data_source_scan(
                     unsafe { &mut *estimate },
                 );
             }
-            Ok(vx_scan::new(Mutex::new(VxScanState::Pending(scan))))
+            Ok(vx_scan::new(VxScan::Pending(scan)))
         })
     })
 }
@@ -249,8 +247,8 @@ pub unsafe extern "C-unwind" fn vx_scan_dtype(
     err: *mut *mut vx_error,
 ) -> *const vx_dtype {
     try_or(err, ptr::null(), || {
-        let scan = vx_scan::as_ref(scan).lock().expect("failed to lock mutex");
-        let VxScanState::Pending(ref scan) = *scan else {
+        let scan = vx_scan::as_ref(scan);
+        let VxScan::Pending(ref scan) = *scan else {
             vortex_bail!("dtype unavailable: scan already started");
         };
         Ok(vx_dtype::new_ref(scan.dtype()))
@@ -265,22 +263,21 @@ pub unsafe extern "C-unwind" fn vx_scan_dtype(
 /// "err".
 /// On error returns NULL and sets "err".
 ///
-/// This function is thread-safe: callers running a multi-threaded pipeline
-/// should call it concurrently and dispatch each partition to a dedicated
-/// worker thread.
+/// This function is thread-unsafe. Callers running a multi-threaded pipeline
+/// should synchronise on calls to this function and dispatch each produced
+/// partition to a dedicated worker thread.
 #[unsafe(no_mangle)]
 pub unsafe extern "C-unwind" fn vx_scan_next_partition(
     scan: *mut vx_scan,
     err: *mut *mut vx_error,
 ) -> *mut vx_partition {
     let scan = vx_scan::as_mut(scan);
-    let mut scan = scan.lock().expect("failed to lock mutex");
     let scan = &mut *scan;
     unsafe {
-        let ptr = scan as *mut VxScanState;
+        let ptr = scan as *mut VxScan;
 
         let on_finish = || -> VortexResult<*mut vx_partition> {
-            ptr::write(ptr, VxScanState::Finished);
+            ptr::write(ptr, VxScan::Finished);
             Ok(ptr::null_mut())
         };
 
@@ -289,7 +286,7 @@ pub unsafe extern "C-unwind" fn vx_scan_next_partition(
                 Some(partition) => {
                     let partition = VxPartitionScan::Pending(partition?);
                     let partition = vx_partition::new(partition);
-                    ptr::write(ptr, VxScanState::Started(stream));
+                    ptr::write(ptr, VxScan::Started(stream));
                     Ok(partition)
                 }
                 None => on_finish(),
@@ -298,9 +295,9 @@ pub unsafe extern "C-unwind" fn vx_scan_next_partition(
 
         let owned = ptr::read(ptr);
         try_or_default(err, || match owned {
-            VxScanState::Pending(scan) => on_stream(scan.partitions()),
-            VxScanState::Started(stream) => on_stream(stream),
-            VxScanState::Finished => on_finish(),
+            VxScan::Pending(scan) => on_stream(scan.partitions()),
+            VxScan::Started(stream) => on_stream(stream),
+            VxScan::Finished => on_finish(),
         })
     }
 }

vortex-ffi/test/common.h

@@ -38,4 +38,4 @@ struct Defer {
 };
 #define CONCAT(x, y)  x##y
 #define CONCAT2(x, y) CONCAT(x, y)
-#define defer Defer CONCAT2(defer_, __LINE__) = [&]
+#define defer         Defer CONCAT2(defer_, __LINE__) = [&]

vortex-ffi/test/scan.cpp

@@ -28,12 +28,14 @@ using nanoarrow::UniqueSchema;
 
 struct TempPath : fs::path {
     TempPath() = default;
-    explicit TempPath(fs::path p) : fs::path(std::move(p)) {}
+    explicit TempPath(fs::path p) : fs::path(std::move(p)) {
+    }
 
     TempPath(const TempPath &) = delete;
     TempPath &operator=(const TempPath &) = delete;
 
-    TempPath(TempPath &&other) noexcept : fs::path(std::move(other)) {}
+    TempPath(TempPath &&other) noexcept : fs::path(std::move(other)) {
+    }
     TempPath &operator=(TempPath &&other) noexcept {
         if (this != &other) {
             fs::remove(*this);
@@ -179,7 +181,7 @@ UniqueArrayStream sample_array_stream() {
     require_no_error(error);
 
     INFO("Written vortex file "s + path.generic_string());
-    return TempPath{path};
+    return TempPath {path};
 }
 
 TEST_CASE("Creating datasources", "[datasource]") {
@@ -216,7 +218,7 @@ TEST_CASE("Creating datasources", "[datasource]") {
     vx_error_free(error);
 
     TempPath file = write_sample(session);
-    for (const char *files : { file.c_str(), "/tmp/*.vortex"}) {
+    for (const char *files : {file.c_str(), "/tmp/*.vortex"}) {
         INFO("reading "s + files);
         opts.paths = files;
         ds = vx_data_source_new(session, &opts, &error);
@@ -479,11 +481,18 @@ TEST_CASE("Multithreaded scan", "[datasource]") {
     REQUIRE(estimate.type == VX_ESTIMATE_INEXACT);
     REQUIRE(estimate.estimate == NUM_FILES);
 
+    std::mutex mutex;
     std::vector<std::jthread> threads;
     for (size_t i = 0; i < NUM_FILES; ++i) {
         threads.emplace_back([&] {
-            vx_partition *partition = vx_scan_next_partition(scan, &error);
-            require_no_error(error);
+            vx_partition *partition = nullptr;
+            {
+                std::lock_guard _(mutex);
+                partition = vx_scan_next_partition(scan, &error);
+                require_no_error(error);
+                REQUIRE(partition != nullptr);
+            }
+
             defer {
                 vx_partition_free(partition);
             };