spiral-sdk 0.1.10

//! Scandal over the lockstep native library, linked into the application.
#[cfg(feature = "source-runtime")]
use std::ffi::c_void;
use std::ffi::{CStr, CString, c_char};
use std::ops::Range;
use std::ptr;
use std::slice::from_raw_parts;
use std::sync::Arc;

use arrow_array::ffi_stream::{ArrowArrayStreamReader, FFI_ArrowArrayStream};
use arrow_array::{RecordBatch, RecordBatchReader};
use arrow_schema::{ArrowError, Schema, SchemaRef};
use async_trait::async_trait;
use bytes::Bytes;
use futures::stream::{self, BoxStream};
use futures::{StreamExt as _, TryStreamExt as _};
use serde::Serialize;
use serde::de::DeserializeOwned;
use spiral_contract::error::ErrorEnvelope;
use spiral_contract::safelong::SafeUnsignedLong;
use tokio::runtime::Handle;
use tokio::task::spawn_blocking;
use uuid::Uuid;

use self::raw::{
    ScandalBlobUpload, ScandalBuffer, ScandalByteStream, ScandalClient, ScandalError as OwnedError,
    ScandalReadCache, scandal_abort_blob_upload, scandal_append_rows, scandal_attach_columns,
    scandal_bind_projection, scandal_blob_upload_finish, scandal_blob_upload_release,
    scandal_blob_upload_write, scandal_byte_stream_next, scandal_byte_stream_release,
    scandal_client_create, scandal_client_create_with_read_cache,
    scandal_client_create_with_uri_resolver, scandal_client_release,
    scandal_client_resolves_resources, scandal_complete_blob_upload, scandal_delete_rows,
    scandal_error_json, scandal_error_message, scandal_error_release, scandal_evaluate,
    scandal_finalize_write, scandal_list_functions, scandal_pack_blobs, scandal_patch_rows,
    scandal_plan_changes, scandal_plan_scan, scandal_read_blob, scandal_read_cache_create,
    scandal_read_cache_release, scandal_read_cache_stats, scandal_read_legacy_table,
    scandal_read_scan, scandal_start_blob_upload, scandal_stat_blob, scandal_string_release,
    scandal_take_rows, scandal_upload_blob_part, scandal_write_blob,
};
use crate::scandal::api::{
    AbortBlobUploadRequest, AppendRowsRequest, AppendedFragments, AttachColumnsRequest,
    AttachedColumns, BindProjectionRequest, ByteRange, ChangePlan, CompleteBlobUploadRequest,
    DeleteRowsRequest, EvaluateRequest, FinalizeWriteRequest, FunctionCatalog, PackBlobsRequest,
    PatchRowsRequest, PlanChangesRequest, PlanScanRequest, ReadBlobRequest, ReadLegacyTableRequest,
    ReadScanRequest, StartBlobUploadRequest, StatBlobRequest, TakeRowsRequest,
    UploadBlobPartRequest, WriteBlobRequest,
};
use crate::scandal::blob::{Blob, BlobPart, BlobStat, BlobUpload};
use crate::scandal::error::ScandalError;
use crate::scandal::expression::NamedExpression;
use crate::scandal::functions::FunctionInfo;
use crate::scandal::read_cache::{ReadCacheOptions, ReadCacheStats};
use crate::scandal::types::{
    FragmentReceipt, PreparedWriteReceipt, Scan, ScanPlan, Split, SplitTake,
};
use crate::scandal::{
    ArrowSchema, ByteStream, Evaluation, RecordBatchStream, Scandal, ScandalConfig, SplitRead,
    StorageBinding, TableBinding, UriResolver, WriteBase, wire,
};

mod resolver;

mod raw {
    include!("bindings.rs");
}

/// An ABI operation with a JSON request and a JSON response.
type JsonCall =
    unsafe extern "C" fn(*const ScandalClient, *const c_char, *mut *mut c_char) -> *mut OwnedError;

/// Embedded scandal from the native library linked into this application: the prepared
/// archive from the same release, or the private runtime linked from source.
///
/// Every operation runs in the library's own runtime, called from a blocking worker of the
/// caller's Tokio runtime.
#[derive(Clone)]
pub struct NativeScandal(Arc<Library>);

impl NativeScandal {
    /// Take ownership of a client allocated by the source-linked runtime.
    ///
    /// # Safety
    ///
    /// `client` must be a non-null client handle created by the matching source-linked
    /// runtime. Ownership is transferred to this wrapper; no other owner may release it.
    #[cfg(feature = "source-runtime")]
    #[doc(hidden)]
    pub unsafe fn from_raw_client(client: *mut c_void) -> Self {
        assert!(!client.is_null(), "a source-linked client must be non-null");
        Self(Arc::new(Library {
            client: client.cast(),
        }))
    }

    /// Create scandal with the storage transports `config` allows.
    pub fn new(config: &ScandalConfig) -> Result<Self, ScandalError> {
        Self::create(None, config)
    }

    /// Create an embedded instance sharing the host's payload cache and I/O budgets.
    /// Each exact storage binding has a separate authorization namespace.
    pub fn with_read_cache(
        cache: &NativeReadCache,
        config: &ScandalConfig,
    ) -> Result<Self, ScandalError> {
        Self::create(Some(cache), config)
    }

    /// Create an embedded host with URI policy and credentials resolved by `resolver`.
    /// Call from the host's Tokio runtime; callbacks execute there through a blocking bridge.
    /// The native instance retains its transport clients and resolved-prefix caches.
    pub fn new_with_uri_resolver(
        config: &ScandalConfig,
        resolver: Arc<dyn UriResolver>,
    ) -> Result<Self, ScandalError> {
        let config = json_string(config)?;
        let resolver = resolver::callbacks(resolver)?;
        let mut client = ptr::null_mut();
        // SAFETY: The callback context transfers on entry, including error paths.
        let error = unsafe {
            scandal_client_create_with_uri_resolver(
                config.as_ptr(),
                ptr::null(),
                resolver,
                &raw mut client,
            )
        };
        check(error)?;
        Ok(Self(Arc::new(Library { client })))
    }

    fn create(
        cache: Option<&NativeReadCache>,
        config: &ScandalConfig,
    ) -> Result<Self, ScandalError> {
        let config = json_string(config)?;
        let mut client = ptr::null_mut();
        // SAFETY: Handles belong to the linked library; strings and output live through the call.
        let error = unsafe {
            match cache {
                Some(cache) => scandal_client_create_with_read_cache(
                    config.as_ptr(),
                    cache.0.cache,
                    &raw mut client,
                ),
                None => scandal_client_create(config.as_ptr(), &raw mut client),
            }
        };
        check(error)?;
        Ok(Self(Arc::new(Library { client })))
    }
}

/// An explicit host-owned encoded-page cache and global I/O budget.
/// Share this handle across native instances; sharing never authorizes a read.
#[derive(Clone)]
pub struct NativeReadCache(Arc<CacheLibrary>);

impl NativeReadCache {
    /// Create shared budgets in the linked runtime. Invalid budgets return an error.
    pub fn new(options: &ReadCacheOptions) -> Result<Self, ScandalError> {
        let options = json_string(options)?;
        let mut cache = ptr::null_mut();
        // SAFETY: Options are NUL-terminated and `cache` is writable for this call.
        let error = unsafe { scandal_read_cache_create(options.as_ptr(), &raw mut cache) };
        check(error)?;
        Ok(Self(Arc::new(CacheLibrary { cache })))
    }

    /// Return retained bytes and cumulative requested-I/O counters without reading storage.
    pub fn stats(&self) -> Result<ReadCacheStats, ScandalError> {
        let mut out = ptr::null_mut();
        // SAFETY: This Arc retains the cache, and `out` is writable.
        let error = unsafe { scandal_read_cache_stats(self.0.cache, &raw mut out) };
        check(error)?;
        json_result(out)
    }
}

struct CacheLibrary {
    cache: *mut ScandalReadCache,
}

// SAFETY: The cache ABI is thread-safe and the final Arc releases its handle once.
unsafe impl Send for CacheLibrary {}
unsafe impl Sync for CacheLibrary {}

impl Drop for CacheLibrary {
    fn drop(&mut self) {
        // SAFETY: No method can use this handle after its final owning Arc drops.
        unsafe { scandal_read_cache_release(self.cache) }
    }
}

/// The scandal client the linked library created.
struct Library {
    client: *mut ScandalClient,
}

// SAFETY: The ABI's calls are thread-safe; the client is released only after the final Arc.
unsafe impl Send for Library {}
unsafe impl Sync for Library {}

impl Drop for Library {
    fn drop(&mut self) {
        // SAFETY: The client is live and no call uses it after the final Arc.
        unsafe { scandal_client_release(self.client) }
    }
}

impl Library {
    /// Send a JSON request and decode the JSON response.
    fn call<T: DeserializeOwned>(
        &self,
        operation: JsonCall,
        request: &impl Serialize,
    ) -> Result<T, ScandalError> {
        let request = json_string(request)?;
        let mut out = ptr::null_mut();
        // SAFETY: The client is live, the request is NUL-terminated and `out` is writable.
        let error = unsafe { operation(self.client, request.as_ptr(), &raw mut out) };
        check(error)?;
        self.json(out)
    }

    fn json<T: DeserializeOwned>(&self, value: *mut c_char) -> Result<T, ScandalError> {
        json_result(value)
    }
}

fn json_result<T: DeserializeOwned>(value: *mut c_char) -> Result<T, ScandalError> {
    if value.is_null() {
        return Err(internal("native library returned no result"));
    }
    // SAFETY: A successful call allocated this string in the linked library.
    let result =
        serde_json::from_slice(unsafe { CStr::from_ptr(value) }.to_bytes()).map_err(internal);
    // SAFETY: The string is released exactly once after copying its result.
    unsafe { scandal_string_release(value) };
    result
}

/// Decode and release an owned library error.
fn check(error: *mut OwnedError) -> Result<(), ScandalError> {
    if error.is_null() {
        return Ok(());
    }
    // SAFETY: A non-null result owns one error; its strings live until it is released.
    let (json, message) = unsafe {
        (
            CStr::from_ptr(scandal_error_json(error))
                .to_bytes()
                .to_vec(),
            CStr::from_ptr(scandal_error_message(error))
                .to_string_lossy()
                .into_owned(),
        )
    };
    // SAFETY: The error is released exactly once, after its strings were copied.
    unsafe { scandal_error_release(error) };
    Err(serde_json::from_slice::<ErrorEnvelope>(&json)
        .map(|envelope| ScandalError::from_wire(&envelope))
        .unwrap_or(ScandalError::Internal { message }))
}

#[async_trait]
impl Scandal for NativeScandal {
    fn resolves_resources(&self) -> bool {
        // SAFETY: this Arc retains the client for the duration of the call.
        unsafe { scandal_client_resolves_resources(self.0.client) }
    }

    async fn read_legacy_table(
        &self,
        request: &ReadLegacyTableRequest,
    ) -> Result<ByteStream, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(request)?;
        let bytes = blocking(move || {
            let mut pointer = ptr::null_mut();
            // SAFETY: The client is live, the request is NUL-terminated and the output is null.
            let error = unsafe {
                scandal_read_legacy_table(library.client, request.as_ptr(), &raw mut pointer)
            };
            check(error)?;
            Ok(NativeBytes {
                pointer,
                _library: library,
            })
        })
        .await?;
        Ok(bytes.into_stream())
    }

    async fn list_functions(&self) -> Result<Vec<FunctionInfo>, ScandalError> {
        let library = Arc::clone(&self.0);
        blocking(move || {
            let mut out = ptr::null_mut();
            // SAFETY: The client is live and `out` is writable.
            let error = unsafe { scandal_list_functions(library.client, &raw mut out) };
            check(error)?;
            let catalog: FunctionCatalog = library.json(out)?;
            Ok(catalog.functions)
        })
        .await
    }

    async fn plan_scan(
        &self,
        scan: &Scan,
        storage: &StorageBinding,
    ) -> Result<ScanPlan, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = PlanScanRequest {
            scan: scan.clone(),
            storage: Some(storage.clone()),
        };
        blocking(move || library.call(scandal_plan_scan, &request)).await
    }

    async fn read_scan(
        &self,
        scan: &Scan,
        split: &Split,
        storage: &StorageBinding,
    ) -> Result<SplitRead, ScandalError> {
        let request = json_string(&ReadScanRequest {
            scan: scan.clone(),
            split: split.clone(),
            storage: Some(storage.clone()),
        })?;
        self.split_read(scandal_read_scan, request).await
    }

    async fn take_rows(
        &self,
        scan: &Scan,
        takes: &[SplitTake],
        storage: &StorageBinding,
    ) -> Result<SplitRead, ScandalError> {
        let request = json_string(&TakeRowsRequest {
            scan: scan.clone(),
            takes: takes.to_vec(),
            storage: Some(storage.clone()),
        })?;
        self.split_read(scandal_take_rows, request).await
    }

    async fn read_blob(
        &self,
        blob: &Blob,
        range: Option<Range<u64>>,
        storage: &StorageBinding,
    ) -> Result<ByteStream, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(&ReadBlobRequest {
            blob: blob.clone(),
            range: range.map(|range| ByteRange {
                start: SafeUnsignedLong(range.start),
                end: SafeUnsignedLong(range.end),
            }),
            storage: Some(storage.clone()),
        })?;
        let bytes = blocking(move || {
            let mut pointer = ptr::null_mut();
            // SAFETY: The client is live, the request is NUL-terminated and `pointer` is null.
            let error =
                unsafe { scandal_read_blob(library.client, request.as_ptr(), &raw mut pointer) };
            check(error)?;
            Ok(NativeBytes {
                pointer,
                _library: library,
            })
        })
        .await?;
        Ok(bytes.into_stream())
    }

    async fn stat_blob(
        &self,
        blob: &Blob,
        storage: &StorageBinding,
    ) -> Result<BlobStat, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = StatBlobRequest {
            blob: blob.clone(),
            storage: Some(storage.clone()),
        };
        blocking(move || library.call(scandal_stat_blob, &request)).await
    }

    async fn write_blob(
        &self,
        table: &TableBinding,
        media_type: &str,
        max_bytes: u64,
        mut input: ByteStream,
    ) -> Result<Blob, ScandalError> {
        let request = json_string(&WriteBlobRequest {
            table: table.clone(),
            media_type: media_type.to_owned(),
            max_bytes: SafeUnsignedLong(max_bytes),
        })?;
        let upload = Arc::new(NativeUpload::start(Arc::clone(&self.0), &request)?);
        while let Some(part) = input.try_next().await? {
            let upload = Arc::clone(&upload);
            blocking(move || upload.write(&part)).await?;
        }
        blocking(move || upload.finish()).await
    }

    async fn pack_blobs(
        &self,
        table: &TableBinding,
        attempt: Option<Uuid>,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> Result<Vec<RecordBatch>, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(&PackBlobsRequest {
            table: table.clone(),
            attempt,
        })?;
        let mut input = host_stream(schema, input)?;
        let reader = blocking(move || {
            let mut out = FFI_ArrowArrayStream::empty();
            // SAFETY: The client is live, the request is NUL-terminated, the input stream is
            // owned and `out` is empty. The library takes the input.
            let error = unsafe {
                scandal_pack_blobs(
                    library.client,
                    request.as_ptr(),
                    &raw mut input,
                    &raw mut out,
                )
            };
            check(error)?;
            NativeReader::new(out, library)
        })
        .await?;
        reader.into_stream().try_collect().await
    }

    async fn start_blob_upload(
        &self,
        table: &TableBinding,
        attempt: Uuid,
        media_type: &str,
        part_bytes: Option<u64>,
    ) -> Result<BlobUpload, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = StartBlobUploadRequest {
            table: table.clone(),
            attempt,
            media_type: media_type.to_owned(),
            part_bytes: part_bytes.map(SafeUnsignedLong),
        };
        blocking(move || library.call(scandal_start_blob_upload, &request)).await
    }

    async fn upload_blob_part(
        &self,
        upload: &BlobUpload,
        part: u32,
        input: ByteStream,
        storage: &StorageBinding,
    ) -> Result<BlobPart, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(&UploadBlobPartRequest {
            upload: upload.clone(),
            storage: Some(storage.clone()),
            part,
        })?;
        // The ABI borrows one buffer: the part, held whole as the library holds it.
        let bytes = input
            .try_fold(Vec::new(), |mut bytes, chunk| async move {
                bytes.extend_from_slice(&chunk);
                Ok(bytes)
            })
            .await?;
        blocking(move || {
            let mut out = ptr::null_mut();
            // SAFETY: The client is live, the request is NUL-terminated, the part's bytes stay
            // borrowed for the call and `out` is writable.
            let error = unsafe {
                scandal_upload_blob_part(
                    library.client,
                    request.as_ptr(),
                    bytes.as_ptr(),
                    bytes.len(),
                    &raw mut out,
                )
            };
            check(error)?;
            library.json(out)
        })
        .await
    }

    async fn complete_blob_upload(
        &self,
        upload: &BlobUpload,
        parts: &[BlobPart],
        storage: &StorageBinding,
    ) -> Result<Blob, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = CompleteBlobUploadRequest {
            upload: upload.clone(),
            storage: Some(storage.clone()),
            parts: parts.to_vec(),
        };
        blocking(move || library.call(scandal_complete_blob_upload, &request)).await
    }

    async fn abort_blob_upload(
        &self,
        upload: &BlobUpload,
        storage: &StorageBinding,
    ) -> Result<(), ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(&AbortBlobUploadRequest {
            upload: upload.clone(),
            storage: Some(storage.clone()),
        })?;
        blocking(move || {
            // SAFETY: The client is live and the request is NUL-terminated.
            check(unsafe { scandal_abort_blob_upload(library.client, request.as_ptr()) })
        })
        .await
    }

    async fn append_rows(
        &self,
        table: &TableBinding,
        base: Option<Uuid>,
        attempt: Option<Uuid>,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
        self.append_partition(
            &AppendRowsRequest {
                table: table.clone(),
                base,
                attempt,
                properties: None,
                schema_changes: None,
            },
            schema,
            input,
        )
        .await
    }

    async fn append_partition(
        &self,
        request: &AppendRowsRequest,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(request)?;
        let mut input = host_stream(schema, input)?;
        blocking(move || {
            let mut out = ptr::null_mut();
            // SAFETY: The client is live, the request is NUL-terminated, the input stream is
            // owned and `out` is writable. The library takes the input.
            let error = unsafe {
                scandal_append_rows(
                    library.client,
                    request.as_ptr(),
                    &raw mut input,
                    &raw mut out,
                )
            };
            check(error)?;
            let appended: AppendedFragments = library.json(out)?;
            Ok(appended.receipts)
        })
        .await
    }

    async fn patch_rows(
        &self,
        request: &PatchRowsRequest,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(request)?;
        let mut input = host_stream(schema, input)?;
        blocking(move || {
            let mut out = ptr::null_mut();
            // SAFETY: The client is live, the request is NUL-terminated, the input stream is
            // owned and `out` is writable. The library takes the input.
            let error = unsafe {
                scandal_patch_rows(
                    library.client,
                    request.as_ptr(),
                    &raw mut input,
                    &raw mut out,
                )
            };
            check(error)?;
            let written: AppendedFragments = library.json(out)?;
            Ok(written.receipts)
        })
        .await
    }

    async fn delete_rows(
        &self,
        request: &DeleteRowsRequest,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> Result<Vec<FragmentReceipt>, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(request)?;
        let mut input = host_stream(schema, input)?;
        blocking(move || {
            let mut out = ptr::null_mut();
            // SAFETY: The client is live, the request is NUL-terminated, the input stream is
            // owned and `out` is writable. The library takes the input.
            let error = unsafe {
                scandal_delete_rows(
                    library.client,
                    request.as_ptr(),
                    &raw mut input,
                    &raw mut out,
                )
            };
            check(error)?;
            let written: AppendedFragments = library.json(out)?;
            Ok(written.receipts)
        })
        .await
    }

    async fn attach_columns(
        &self,
        request: &AttachColumnsRequest,
        schema: SchemaRef,
        input: RecordBatchStream,
    ) -> Result<AttachedColumns, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(request)?;
        let mut input = host_stream(schema, input)?;
        blocking(move || {
            let mut out = ptr::null_mut();
            // SAFETY: The client is live, the request is NUL-terminated, the input stream is
            // owned and `out` is writable. The library takes the input.
            let error = unsafe {
                scandal_attach_columns(
                    library.client,
                    request.as_ptr(),
                    &raw mut input,
                    &raw mut out,
                )
            };
            check(error)?;
            library.json(out)
        })
        .await
    }

    async fn plan_changes(&self, request: &PlanChangesRequest) -> Result<ChangePlan, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = request.clone();
        blocking(move || library.call(scandal_plan_changes, &request)).await
    }

    async fn finalize_write(
        &self,
        table: &TableBinding,
        base: &WriteBase,
        receipts: &[FragmentReceipt],
    ) -> Result<PreparedWriteReceipt, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = FinalizeWriteRequest {
            table: table.clone(),
            base: base.clone(),
            receipts: receipts.to_vec(),
        };
        blocking(move || library.call(scandal_finalize_write, &request)).await
    }

    async fn bind_projection(
        &self,
        input: &Schema,
        projection: &[NamedExpression],
    ) -> Result<SchemaRef, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = BindProjectionRequest {
            input: ArrowSchema::from_schema(input)?,
            projection: projection.to_vec(),
        };
        let schema: ArrowSchema =
            blocking(move || library.call(scandal_bind_projection, &request)).await?;
        Ok(Arc::new(schema.to_schema()?))
    }

    async fn evaluate(
        &self,
        projection: &[NamedExpression],
        input_schema: SchemaRef,
        input: RecordBatchStream,
        tables: &[TableBinding],
    ) -> Result<Evaluation, ScandalError> {
        let library = Arc::clone(&self.0);
        let request = json_string(&EvaluateRequest {
            projection: projection.to_vec(),
            tables: tables.to_vec(),
        })?;
        let mut input = host_stream(input_schema, input)?;
        let reader = blocking(move || {
            let mut out = FFI_ArrowArrayStream::empty();
            // SAFETY: The client is live, the request is NUL-terminated, the input stream is
            // owned and `out` is empty. The library takes the input.
            let error = unsafe {
                scandal_evaluate(
                    library.client,
                    request.as_ptr(),
                    &raw mut input,
                    &raw mut out,
                )
            };
            check(error)?;
            NativeReader::new(out, library)
        })
        .await?;
        Ok(Evaluation {
            schema: reader.schema(),
            batches: reader.into_stream(),
        })
    }
}

/// An ABI operation with a JSON request and an Arrow stream of a read's batches.
type ReadCall = unsafe extern "C" fn(
    *const ScandalClient,
    *const c_char,
    *mut FFI_ArrowArrayStream,
) -> *mut OwnedError;

impl NativeScandal {
    /// Run a read and separate its batches' rows from their progress columns.
    async fn split_read(
        &self,
        operation: ReadCall,
        request: CString,
    ) -> Result<SplitRead, ScandalError> {
        let library = Arc::clone(&self.0);
        let reader = blocking(move || {
            let mut out = FFI_ArrowArrayStream::empty();
            // SAFETY: The client is live, the request is NUL-terminated and `out` is empty.
            let error = unsafe { operation(library.client, request.as_ptr(), &raw mut out) };
            check(error)?;
            NativeReader::new(out, library)
        })
        .await?;
        Ok(SplitRead {
            schema: wire::read_scan_rows(&reader.schema())?,
            batches: reader
                .into_stream()
                .map(|batch| batch.and_then(wire::detach_progress))
                .boxed(),
        })
    }
}

/// An Arrow stream the library returned, kept alive with its library.
struct NativeReader {
    reader: ArrowArrayStreamReader,
    _library: Arc<Library>,
}

impl NativeReader {
    fn new(stream: FFI_ArrowArrayStream, library: Arc<Library>) -> Result<Self, ScandalError> {
        Ok(Self {
            reader: ArrowArrayStreamReader::try_new(stream).map_err(internal)?,
            _library: library,
        })
    }

    fn schema(&self) -> SchemaRef {
        self.reader.schema()
    }

    /// Pull batches on blocking workers, one at a time.
    fn into_stream(self) -> BoxStream<'static, Result<RecordBatch, ScandalError>> {
        stream::try_unfold(self, |mut reader| async move {
            let (batch, reader) = blocking(move || {
                let batch = reader.reader.next().transpose().map_err(internal);
                Ok((batch, reader))
            })
            .await?;
            Ok(batch?.map(|batch| (batch, reader)))
        })
        .boxed()
    }
}

/// A blob's byte stream in the library.
struct NativeBytes {
    pointer: *mut ScandalByteStream,
    _library: Arc<Library>,
}

// SAFETY: The stream is used by one call at a time and released once, on drop.
unsafe impl Send for NativeBytes {}

impl Drop for NativeBytes {
    fn drop(&mut self) {
        // SAFETY: The stream is live and released exactly once.
        unsafe { scandal_byte_stream_release(self.pointer) }
    }
}

impl NativeBytes {
    /// The next chunk, or `None` at the end.
    fn next(&mut self) -> Result<Option<Bytes>, ScandalError> {
        let mut buffer = ScandalBuffer {
            data: ptr::null(),
            len: 0,
        };
        // SAFETY: The stream is live and `buffer` is writable.
        let error = unsafe { scandal_byte_stream_next(self.pointer, &raw mut buffer) };
        check(error)?;
        if buffer.len == 0 {
            return Ok(None);
        }
        // SAFETY: The library lends `len` bytes until the next call on this stream.
        let chunk = unsafe { from_raw_parts(buffer.data, buffer.len) };
        Ok(Some(Bytes::copy_from_slice(chunk)))
    }

    fn into_stream(self) -> ByteStream {
        stream::try_unfold(self, |mut bytes| async move {
            let (chunk, bytes) = blocking(move || Ok((bytes.next()?, bytes))).await?;
            Ok(chunk.map(|chunk| (chunk, bytes)))
        })
        .boxed()
    }
}

/// A blob upload in the library.
struct NativeUpload {
    pointer: *mut ScandalBlobUpload,
    library: Arc<Library>,
}

// SAFETY: The caller writes and finishes the upload in sequence; release runs once, on drop.
unsafe impl Send for NativeUpload {}
unsafe impl Sync for NativeUpload {}

impl Drop for NativeUpload {
    fn drop(&mut self) {
        // SAFETY: The upload is live and released exactly once; an unfinished one is cancelled.
        unsafe { scandal_blob_upload_release(self.pointer) }
    }
}

impl NativeUpload {
    fn start(library: Arc<Library>, request: &CStr) -> Result<Self, ScandalError> {
        let mut pointer = ptr::null_mut();
        // SAFETY: The client is live, the request is NUL-terminated and `pointer` is null.
        let error =
            unsafe { scandal_write_blob(library.client, request.as_ptr(), &raw mut pointer) };
        check(error)?;
        Ok(Self { pointer, library })
    }

    fn write(&self, part: &[u8]) -> Result<(), ScandalError> {
        // SAFETY: The upload is live and the part's bytes are readable for this call.
        let error = unsafe { scandal_blob_upload_write(self.pointer, part.as_ptr(), part.len()) };
        check(error)
    }

    fn finish(&self) -> Result<Blob, ScandalError> {
        let mut out = ptr::null_mut();
        // SAFETY: The upload is live and `out` is writable.
        let error = unsafe { scandal_blob_upload_finish(self.pointer, &raw mut out) };
        check(error)?;
        self.library.json(out)
    }
}

/// Lend the caller's rows to the library as an Arrow C stream.
///
/// The library reads it on its own threads, which wait on the caller's runtime for each batch.
fn host_stream(
    schema: SchemaRef,
    input: RecordBatchStream,
) -> Result<FFI_ArrowArrayStream, ScandalError> {
    let handle = Handle::try_current()
        .map_err(|_| internal("native scandal streams rows from a Tokio runtime"))?;
    Ok(FFI_ArrowArrayStream::new(Box::new(HostReader {
        schema,
        input,
        handle,
    })))
}

struct HostReader {
    schema: SchemaRef,
    input: RecordBatchStream,
    handle: Handle,
}

impl Iterator for HostReader {
    type Item = Result<RecordBatch, ArrowError>;

    fn next(&mut self) -> Option<Self::Item> {
        self.handle
            .block_on(self.input.next())
            .map(|batch| batch.map_err(|error| ArrowError::ExternalError(Box::new(error))))
    }
}

impl RecordBatchReader for HostReader {
    fn schema(&self) -> SchemaRef {
        Arc::clone(&self.schema)
    }
}

/// Run a blocking library call on a blocking worker of the caller's runtime.
#[expect(
    clippy::disallowed_methods,
    reason = "The open crate drives a synchronous C ABI without linking a Vortex runtime"
)]
async fn blocking<T: Send + 'static>(
    work: impl FnOnce() -> Result<T, ScandalError> + Send + 'static,
) -> Result<T, ScandalError> {
    spawn_blocking(work)
        .await
        .map_err(|_| internal("native scandal call stopped"))?
}

fn json_string(value: &impl Serialize) -> Result<CString, ScandalError> {
    CString::new(serde_json::to_vec(value).map_err(internal)?).map_err(internal)
}

fn internal(error: impl ToString) -> ScandalError {
    ScandalError::Internal {
        message: error.to_string(),
    }
}