feat: add lance_dataset_write for create/append/overwrite from ArrowArrayStream

Writes an ArrowArrayStream into a Lance dataset with a committed manifest.
A mode enum (CREATE / APPEND / OVERWRITE) and an optional out_dataset that
returns the open dataset at the new version (so callers don't need to
reopen). Structure follows fragment_writer.rs: schema fail-fast, storage
options pass-through, thread-local errors. C++ gets a scoped WriteMode
enum and a lance::Dataset::write() static method that reads the stream's
schema automatically.

Two FFI details:
  - mode is received as i32 and validated via LanceWriteMode::from_raw;
    accepting the enum directly would be UB for out-of-range values.
  - The stream is consumed via ArrowArrayStreamReader::from_raw right
    after the NULL check so the "consumed on any return" contract holds.

Tests: 11 new Rust unit tests (CREATE/APPEND/OVERWRITE happy paths,
OVERWRITE-on-missing, CREATE-on-existing, schema mismatches, empty
stream, NULL args, invalid mode, out_dataset propagation). The ignored
C and C++ integration tests now do a scan->write round-trip.

Closes #14.

Author: LuciferYang

include/lance.h

@@ -340,6 +340,51 @@ int32_t lance_write_fragments(
     const char* const* storage_opts
 );
 
+/* ─── Dataset writer ─── */
+
+/**
+ * Write mode for lance_dataset_write. Values are ABI-stable.
+ * The Rust implementation validates the received integer and rejects any
+ * out-of-range value with LANCE_ERR_INVALID_ARGUMENT.
+ */
+typedef enum {
+    LANCE_WRITE_CREATE    = 0,  /* Create new dataset; fail if path exists. */
+    LANCE_WRITE_APPEND    = 1,  /* Append; fail if the new schema is incompatible. */
+    LANCE_WRITE_OVERWRITE = 2,  /* Overwrite existing, or create if missing. */
+} LanceWriteMode;
+
+/**
+ * Write an Arrow record batch stream to a Lance dataset at `uri`, committing
+ * a manifest.
+ *
+ * @param uri          Dataset URI (file://, s3://, memory://, etc.). Must not
+ *                     be NULL or an empty string.
+ * @param schema       Required Arrow schema. The stream schema must match or
+ *                     the call fails with LANCE_ERR_INVALID_ARGUMENT.
+ * @param stream       Arrow C Data Interface stream consumed by this call.
+ *                     Do not use the stream after returning, regardless of
+ *                     the return code.
+ * @param mode         CREATE / APPEND / OVERWRITE (see LanceWriteMode).
+ * @param storage_opts NULL-terminated key-value pairs ["k","v",NULL], or NULL.
+ * @param out_dataset  If non-NULL, on success receives an open LanceDataset*
+ *                     at the newly-committed version (caller must
+ *                     lance_dataset_close it). Pass NULL to discard. On error
+ *                     *out_dataset is left unchanged — do not read or free it.
+ * @return 0 on success, -1 on error. Possible error codes include
+ *         LANCE_ERR_DATASET_ALREADY_EXISTS (CREATE on an existing path),
+ *         LANCE_ERR_INVALID_ARGUMENT (NULL/empty args, invalid mode,
+ *         schema mismatch),
+ *         LANCE_ERR_COMMIT_CONFLICT (concurrent writer).
+ */
+int32_t lance_dataset_write(
+    const char* uri,
+    const struct ArrowSchema* schema,
+    struct ArrowArrayStream* stream,
+    LanceWriteMode mode,
+    const char* const* storage_opts,
+    LanceDataset** out_dataset
+);
+
 #ifdef __cplusplus
 } /* extern "C" */
 #endif

include/lance.hpp

@@ -17,6 +17,7 @@
 
 #include "lance.h"
 
+#include <cstdint>
 #include <memory>
 #include <stdexcept>
 #include <string>
@@ -94,6 +95,14 @@ struct VersionInfo {
     int64_t  timestamp_ms;
 };
 
+// ─── Write mode ──────────────────────────────────────────────────────────────
+
+enum class WriteMode : int32_t {
+    Create    = LANCE_WRITE_CREATE,
+    Append    = LANCE_WRITE_APPEND,
+    Overwrite = LANCE_WRITE_OVERWRITE,
+};
+
 // ─── Dataset ─────────────────────────────────────────────────────────────────
 
 class Dataset {
@@ -122,6 +131,59 @@ class Dataset {
         return Dataset(ds);
     }
 
+    /// Write an Arrow record batch stream to a Lance dataset and return the
+    /// open dataset at the committed version.
+    ///
+    /// The stream must be self-describing; its own schema is used. Treat the
+    /// stream as consumed once this call returns or throws — do not reuse it.
+    /// Throws lance::Error on failure (including if `stream` is null).
+    static Dataset write(
+        const std::string& uri,
+        ArrowArrayStream* stream,
+        WriteMode mode,
+        const std::vector<std::pair<std::string, std::string>>& storage_opts = {}) {
+
+        if (stream == nullptr) {
+            throw Error(LANCE_ERR_INVALID_ARGUMENT, "stream must not be null");
+        }
+
+        // Pull the stream's schema so we can pass it to the C API.
+        ArrowSchema schema = {};
+        if (stream->get_schema(stream, &schema) != 0) {
+            const char* err = stream->get_last_error
+                ? stream->get_last_error(stream)
+                : nullptr;
+            throw Error(
+                LANCE_ERR_INVALID_ARGUMENT,
+                std::string("failed to read stream schema: ") +
+                    (err ? err : "unknown"));
+        }
+        struct SchemaGuard {
+            ArrowSchema* s;
+            ~SchemaGuard() { if (s && s->release) s->release(s); }
+        } guard{&schema};
+
+        std::vector<const char*> kv;
+        for (auto& [k, v] : storage_opts) {
+            kv.push_back(k.c_str());
+            kv.push_back(v.c_str());
+        }
+        kv.push_back(nullptr);
+        const char* const* opts_ptr =
+            storage_opts.empty() ? nullptr : kv.data();
+
+        LanceDataset* out = nullptr;
+        int32_t rc = lance_dataset_write(
+            uri.c_str(),
+            &schema,
+            stream,
+            static_cast<LanceWriteMode>(mode),
+            opts_ptr,
+            &out);
+        if (rc != 0) check_error();
+        return Dataset(out);
+    }
+
     /// Number of rows in the dataset.
     uint64_t count_rows() const {
         uint64_t n = lance_dataset_count_rows(handle_.get());

src/lib.rs

@@ -24,6 +24,7 @@ mod helpers;
 pub mod runtime;
 mod scanner;
 mod versions;
+mod writer;
 
 // Re-export all extern "C" symbols so they appear in the cdylib.
 pub use batch::*;
@@ -34,3 +35,4 @@ pub use error::{
 pub use fragment_writer::*;
 pub use scanner::*;
 pub use versions::*;
+pub use writer::*;

src/writer.rs

@@ -0,0 +1,186 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+//! Dataset write C API: create, append, or overwrite a Lance dataset from an
+//! Arrow C Data Interface stream, committing a manifest.
+//!
+//! Mirrors the structure of `src/fragment_writer.rs` but produces a full
+//! dataset with a committed manifest rather than uncommitted fragment files.
+
+use std::ffi::c_char;
+use std::sync::Arc;
+
+use arrow::ffi::FFI_ArrowSchema;
+use arrow::ffi_stream::{ArrowArrayStreamReader, FFI_ArrowArrayStream};
+use arrow::record_batch::RecordBatchReader;
+use arrow_schema::Schema as ArrowSchema;
+use lance::Dataset;
+use lance::dataset::{WriteMode as LanceWriteModeUpstream, WriteParams};
+use lance_core::Result;
+use lance_io::object_store::{ObjectStoreParams, StorageOptionsAccessor};
+
+use crate::dataset::LanceDataset;
+use crate::error::ffi_try;
+use crate::helpers;
+use crate::runtime::block_on;
+
+/// Write mode for `lance_dataset_write`.
+///
+/// Discriminants are pinned for ABI stability. The FFI accepts this as
+/// `int32_t` and validates via [`LanceWriteMode::from_raw`] — storing an
+/// out-of-range tag as an enum would be UB.
+#[repr(C)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum LanceWriteMode {
+    /// Create a new dataset. Fails with `LANCE_ERR_DATASET_ALREADY_EXISTS` if
+    /// the path already exists.
+    Create = 0,
+    /// Append to an existing dataset. Fails with `LANCE_ERR_INVALID_ARGUMENT`
+    /// if the stream schema is incompatible with the existing dataset schema.
+    Append = 1,
+    /// Overwrite an existing dataset (or create one if the path does not exist).
+    Overwrite = 2,
+}
+
+impl LanceWriteMode {
+    /// Validate a raw FFI integer into a `LanceWriteMode`. Out-of-range
+    /// values become `InvalidInput`.
+    fn from_raw(raw: i32) -> Result<Self> {
+        match raw {
+            0 => Ok(Self::Create),
+            1 => Ok(Self::Append),
+            2 => Ok(Self::Overwrite),
+            other => Err(lance_core::Error::InvalidInput {
+                source: format!(
+                    "invalid write mode {other}; expected 0 (create), 1 (append), or 2 (overwrite)"
+                )
+                .into(),
+                location: snafu::location!(),
+            }),
+        }
+    }
+}
+
+impl From<LanceWriteMode> for LanceWriteModeUpstream {
+    fn from(mode: LanceWriteMode) -> Self {
+        match mode {
+            LanceWriteMode::Create => LanceWriteModeUpstream::Create,
+            LanceWriteMode::Append => LanceWriteModeUpstream::Append,
+            LanceWriteMode::Overwrite => LanceWriteModeUpstream::Overwrite,
+        }
+    }
+}
+
+/// Write an Arrow record batch stream to a Lance dataset at `uri`, committing a manifest.
+///
+/// - `uri`: Dataset URI (`file://`, `s3://`, `memory://`, ...). Must not be NULL or empty.
+/// - `schema`: Caller-provided Arrow schema. The stream's schema must match;
+///   mismatch returns `LANCE_ERR_INVALID_ARGUMENT`.
+/// - `stream`: Arrow C Data Interface stream. Consumed by this call — the
+///   caller must not use it again on any return path.
+/// - `mode`: `LANCE_WRITE_CREATE` (0), `LANCE_WRITE_APPEND` (1), or
+///   `LANCE_WRITE_OVERWRITE` (2). Any other value → `LANCE_ERR_INVALID_ARGUMENT`.
+/// - `storage_opts`: NULL-terminated key-value pairs `["k","v",NULL]`, or NULL.
+/// - `out_dataset`: If non-NULL, receives an open `LanceDataset*` at the new
+///   version on success (caller closes). Pass NULL to discard. On error
+///   `*out_dataset` is untouched — do not read or free it.
+///
+/// Returns 0 on success, -1 on error.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn lance_dataset_write(
+    uri: *const c_char,
+    schema: *const FFI_ArrowSchema,
+    stream: *mut FFI_ArrowArrayStream,
+    mode: i32,
+    storage_opts: *const *const c_char,
+    out_dataset: *mut *mut LanceDataset,
+) -> i32 {
+    ffi_try!(
+        unsafe { write_dataset_inner(uri, schema, stream, mode, storage_opts, out_dataset) },
+        neg
+    )
+}
+
+unsafe fn write_dataset_inner(
+    uri: *const c_char,
+    schema: *const FFI_ArrowSchema,
+    stream: *mut FFI_ArrowArrayStream,
+    mode: i32,
+    storage_opts: *const *const c_char,
+    out_dataset: *mut *mut LanceDataset,
+) -> Result<i32> {
+    if uri.is_null() || schema.is_null() || stream.is_null() {
+        return Err(lance_core::Error::InvalidInput {
+            source: "uri, schema, and stream must not be NULL".into(),
+            location: snafu::location!(),
+        });
+    }
+
+    // Consume the stream before any other fallible validation. `from_raw`
+    // swaps the caller's stream into a Rust-owned reader unconditionally, so
+    // the stream's resources are released on every return path.
+    let reader = unsafe { ArrowArrayStreamReader::from_raw(stream) }.map_err(|e| {
+        lance_core::Error::InvalidInput {
+            source: e.to_string().into(),
+            location: snafu::location!(),
+        }
+    })?;
+
+    // Validate the mode at the boundary — storing an out-of-range tag as a
+    // `LanceWriteMode` would be UB.
+    let mode = LanceWriteMode::from_raw(mode)?;
+
+    let uri_str = unsafe { helpers::parse_c_string(uri)? }.ok_or_else(|| {
+        lance_core::Error::InvalidInput {
+            source: "uri must not be empty".into(),
+            location: snafu::location!(),
+        }
+    })?;
+
+    let expected_schema = ArrowSchema::try_from(unsafe { &*schema }).map_err(|e| {
+        lance_core::Error::InvalidInput {
+            source: format!("invalid schema: {e}").into(),
+            location: snafu::location!(),
+        }
+    })?;
+
+    let opts = unsafe { helpers::parse_storage_options(storage_opts)? };
+
+    // Fail fast: compare the stream schema against the caller-provided schema.
+    let stream_schema = reader.schema();
+    if stream_schema.fields() != expected_schema.fields() {
+        return Err(lance_core::Error::InvalidInput {
+            source: format!(
+                "stream schema does not match the provided schema.\n  expected: {expected_schema}\n  got:      {stream_schema}"
+            )
+            .into(),
+            location: snafu::location!(),
+        });
+    }
+
+    let mut params = WriteParams {
+        mode: mode.into(),
+        ..WriteParams::default()
+    };
+    if !opts.is_empty() {
+        params.store_params = Some(ObjectStoreParams {
+            storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options(
+                opts,
+            ))),
+            ..ObjectStoreParams::default()
+        });
+    }
+
+    let dataset = block_on(Dataset::write(reader, uri_str, Some(params)))?;
+
+    if !out_dataset.is_null() {
+        let handle = LanceDataset {
+            inner: Arc::new(dataset),
+        };
+        unsafe {
+            *out_dataset = Box::into_raw(Box::new(handle));
+        }
+    }
+
+    Ok(0)
+}