feat: add lance_dataset_write for create/append/overwrite from ArrowArrayStream (#16)

## Summary

- Adds `lance_dataset_write(uri, schema, stream, mode, storage_opts,
out_dataset)` — writes an `ArrowArrayStream` into a Lance dataset with a
committed manifest
- `LanceWriteMode` covers `CREATE` / `APPEND` / `OVERWRITE`
- Optional `out_dataset` hands back an open `LanceDataset*` at the new
version so callers don't need to reopen
- Matching `lance::Dataset::write(...)` static method in `lance.hpp`
with full RAII (`StreamGuard` + `SchemaGuard`)

## Motivation

Until now the C/C++ path only produced uncommitted fragment files (#5).
`lance_dataset_write` closes the primary write path and unblocks the
rest of Phase 3 (delete, update, merge-insert, schema evolution), which
all need a way to create a dataset first.

## FFI contract notes

- `mode` parameter is `int32_t` (not `LanceWriteMode`) on the wire —
defends against `-fshort-enums` ABI mismatch. Validated in Rust via
`LanceWriteMode::from_raw` before any unsafe enum construction.
- Stream is consumed via `ArrowArrayStreamReader::from_raw` **before**
uri/schema NULL checks, so the "consumed on every return path" contract
holds for every error branch — verified by
`test_dataset_write_releases_stream_on_every_error_path` (drop-counter
on the boxed reader).
- Schema is read by shared reference; the function does NOT call
`schema->release`. Caller (or C++ `SchemaGuard`) retains ownership.
Documented in the header.
- `*out_dataset` is written only on success; error paths leave it
untouched. Verified by sentinel-pointer test.
- C++ wrapper builds `SchemaGuard` BEFORE `get_schema` so a
non-conforming producer that partially populates the schema before
reporting failure still has its `release` fired on unwind. `StreamGuard`
covers `std::bad_alloc` during `kv` construction; `disarm()`s right
before the C call.

## Test plan

- `cargo test` — 87 integration tests, 13 covering the writer
(CREATE/APPEND/OVERWRITE happy paths, OVERWRITE on a missing path,
CREATE on an existing path, declared/append schema mismatches, empty
stream, NULL args, invalid mode, `out_dataset` propagation,
stream-release-on-every-error-path, out_dataset-untouched-on-error)
- `cargo clippy --all-targets -- -D warnings` clean
- `cargo fmt --check` clean
- `cargo test --test compile_and_run_test -- --ignored` — C and C++
scan→write round-trips pass

Closes #14.

Author: LuciferYang

include/lance.h

@@ -481,6 +481,62 @@ int32_t lance_scanner_full_text_search(
     uint32_t max_fuzzy_distance
 );
 
+/* ─── Dataset writer ─── */
+
+/**
+ * Write mode for lance_dataset_write. Values are ABI-stable.
+ *
+ * The `mode` parameter on the FFI call is a fixed-width int32_t — not this
+ * enum type — so callers built with `-fshort-enums` or non-default enum
+ * sizing cannot mismatch the Rust ABI. 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. This
+ *                     function does NOT call schema->release; the caller
+ *                     retains ownership and must release the schema after the
+ *                     call returns (success or failure).
+ * @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.
+ *                     On entry `*out_dataset` should be NULL or a pointer
+ *                     whose previous value is no longer needed; this function
+ *                     overwrites the slot on success without releasing any
+ *                     prior handle.
+ * @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,
+    int32_t 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,118 @@ 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");
+        }
+
+        // RAII guard for the stream. Until `lance_dataset_write` is called,
+        // any exception (failed `get_schema`, `std::bad_alloc` while building
+        // `kv`, etc.) must release the stream. After that call Rust owns it,
+        // so we `disarm()` immediately before invoking the C API.
+        struct StreamGuard {
+            ArrowArrayStream* s;
+            bool armed = true;
+            // Explicit constructor: `= delete`d copy/move ctors disqualify
+            // this from being an aggregate under C++20, so brace-init like
+            // `StreamGuard{stream}` would otherwise fail to compile there.
+            explicit StreamGuard(ArrowArrayStream* p) noexcept : s(p) {}
+            ~StreamGuard() noexcept {
+                if (armed && s && s->release) s->release(s);
+            }
+            void disarm() noexcept { armed = false; }
+            StreamGuard(const StreamGuard&) = delete;
+            StreamGuard& operator=(const StreamGuard&) = delete;
+            StreamGuard(StreamGuard&&) = delete;
+            StreamGuard& operator=(StreamGuard&&) = delete;
+        } stream_guard{stream};
+
+        // Defensive: a non-conforming or already-released producer may have a
+        // null `get_schema`. Without this guard a bad caller would crash with
+        // a null function-pointer dereference on the next line.
+        if (stream->get_schema == nullptr) {
+            throw Error(LANCE_ERR_INVALID_ARGUMENT,
+                        "stream get_schema callback is null");
+        }
+
+        // Arm SchemaGuard before calling `get_schema` so a non-conforming
+        // producer that partially populates the schema before returning an
+        // error still has its `release` fired on unwind. The zero-init keeps
+        // the destructor a no-op on the clean-error path (release == null).
+        struct SchemaGuard {
+            ArrowSchema* s;
+            // Explicit constructor for the same C++20 aggregate-init reason
+            // documented on StreamGuard above.
+            explicit SchemaGuard(ArrowSchema* p) noexcept : s(p) {}
+            ~SchemaGuard() noexcept {
+                if (s && s->release) s->release(s);
+            }
+            SchemaGuard(const SchemaGuard&) = delete;
+            SchemaGuard& operator=(const SchemaGuard&) = delete;
+            SchemaGuard(SchemaGuard&&) = delete;
+            SchemaGuard& operator=(SchemaGuard&&) = delete;
+        };
+        ArrowSchema schema = {};
+        SchemaGuard schema_guard{&schema};
+
+        // On failure, StreamGuard releases the stream and SchemaGuard
+        // releases any partial schema state — preserving the "consumed on
+        // return or throw" contract for both resources.
+        if (stream->get_schema(stream, &schema) != 0) {
+            const char* err = stream->get_last_error
+                ? stream->get_last_error(stream)
+                : nullptr;
+            std::string msg = std::string("failed to read stream schema: ") +
+                              (err ? err : "unknown");
+            throw Error(LANCE_ERR_INVALID_ARGUMENT, msg);
+        }
+
+        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();
+
+        // The C API consumes the stream on every return path, so disarm the
+        // guard before calling. After this point the stream pointer is logically
+        // owned by Rust and any C++-side exception must not re-release it.
+        stream_guard.disarm();
+
+        LanceDataset* out = nullptr;
+        int32_t rc = lance_dataset_write(
+            uri.c_str(),
+            &schema,
+            stream,
+            static_cast<int32_t>(mode),
+            opts_ptr,
+            &out);
+        if (rc != 0) check_error();
+        // Defensive null guard: a conforming Rust impl never returns rc == 0
+        // with `out == nullptr`, but constructing a Dataset around a null
+        // handle would silently crash on the first method call. Throw
+        // explicitly rather than going through `check_error()` because the
+        // thread-local code is `LANCE_OK` on this path (rc == 0).
+        if (!out) {
+            throw Error(LANCE_ERR_INTERNAL,
+                        "lance_dataset_write returned success with null out_dataset");
+        }
+        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

@@ -25,6 +25,7 @@ mod index;
 pub mod runtime;
 mod scanner;
 mod versions;
+mod writer;
 
 // Re-export all extern "C" symbols so they appear in the cdylib.
 pub use batch::*;
@@ -36,3 +37,4 @@ pub use fragment_writer::*;
 pub use index::*;
 pub use scanner::*;
 pub use versions::*;
+pub use writer::*;