diff --git a/include/lance/lance.h b/include/lance/lance.h index e09eb09..e9a8a84 100644 --- a/include/lance/lance.h +++ b/include/lance/lance.h @@ -194,6 +194,7 @@ typedef struct LanceDataStatistics LanceDataStatistics; typedef struct LanceIndexSegmentBuilder LanceIndexSegmentBuilder; typedef struct LanceIndexSegmentMetadata LanceIndexSegmentMetadata; typedef struct LanceFtsQueryContext LanceFtsQueryContext; +typedef struct LanceBlobFile LanceBlobFile; /* ─── Shared session ─── */ @@ -923,6 +924,180 @@ int32_t lance_dataset_take_rows( struct ArrowArrayStream* out ); +/* ─── Blob v2 random access ─── */ + +/* + * A LanceBlobFile is a file-like handle over one value of a Blob v2 column, + * returned by lance_dataset_take_blobs() / lance_dataset_take_blobs_by_indices() + * and released with lance_blob_file_close(). It owns what it needs to read, + * so it stays valid after the dataset is closed. Not thread-safe per handle; + * distinct handles are independent. + * + * Reads are cursor-based: the cursor starts at 0, lance_blob_file_read() and + * lance_blob_file_read_up_to() advance it, lance_blob_file_read_range() does + * not, lance_blob_file_seek() sets it. + */ + +/** + * Take blob handles by dataset row ID. + * + * Row IDs are values from the `_rowid` scanner column, not zero-based row + * offsets. They must belong to the same dataset snapshot used for this read. + * + * On success `out[i]` holds the handle for `row_ids[i]`, or NULL when that + * blob value is null (an empty blob is a handle of size 0). The caller closes + * every non-NULL handle exactly once. On failure `out` is left untouched; a + * row ID that does not resolve fails the whole call. + * + * @param dataset Open dataset snapshot. + * @param row_ids Array of dataset row IDs. May be NULL only when + * `num_row_ids` is zero. + * @param num_row_ids Length of `row_ids`. Zero is a no-op that succeeds + * without writing to `out`. + * @param column Name of a Blob v2 column. Must not be NULL. A missing + * column, or a column that is not a blob column, is an + * error. + * @param out Caller-allocated array of at least `num_row_ids` + * handle pointers. Must not be NULL. + * @return 0 on success, -1 on error + */ +int32_t lance_dataset_take_blobs( + const LanceDataset* dataset, + const uint64_t* row_ids, + size_t num_row_ids, + const char* column, + LanceBlobFile** out +); + +/** + * Take blob handles by row index. + * + * Row indices are 0-based offsets in the dataset, as used by + * lance_dataset_take(). Ownership, ordering, NULL slots, and failure + * behavior are identical to lance_dataset_take_blobs(). + * + * @param dataset Open dataset snapshot. + * @param indices Array of 0-based row offsets. May be NULL only when + * `num_indices` is zero. + * @param num_indices Length of `indices`. Zero is a no-op that succeeds + * without writing to `out`. + * @param column Name of a Blob v2 column. Must not be NULL. + * @param out Caller-allocated array of at least `num_indices` + * handle pointers. Must not be NULL. + * @return 0 on success, -1 on error + */ +int32_t lance_dataset_take_blobs_by_indices( + const LanceDataset* dataset, + const uint64_t* indices, + size_t num_indices, + const char* column, + LanceBlobFile** out +); + +/** + * Return the size of the blob in bytes. + * + * Metadata carried by the handle: no storage access, independent of the + * cursor, still available after lance_dataset_close(). + * + * @param blob Blob handle. NULL is an error. + * @return The blob size, or 0 on error. A return of 0 may be an empty blob + * or an error; check lance_last_error_code() to tell them apart. + */ +uint64_t lance_blob_file_size(const LanceBlobFile* blob); + +/** + * Read from the current cursor to the end of the blob. + * + * With the cursor at 0 that is the whole blob. The cursor ends up at the end. + * `dst` must hold every remaining byte; a smaller buffer is an error and + * reads nothing. At or past the end this writes nothing and succeeds. + * + * @param blob Blob handle. NULL is an error. + * @param dst Destination buffer. May be NULL only when no bytes remain + * from the current cursor. + * @param dst_len Capacity of `dst` in bytes. Must be at least the number of + * bytes remaining from the cursor, or 0 if the cursor is at + * or past the end. + * @return 0 on success, -1 on error + */ +int32_t lance_blob_file_read(LanceBlobFile* blob, uint8_t* dst, size_t dst_len); + +/** + * Read at most `len` bytes from the current cursor. + * + * Reads `min(len, size - cursor)` bytes and advances the cursor past them, + * so repeated calls walk the blob. At or past the end this writes no bytes, + * stores 0 in `*bytes_read`, and succeeds. + * + * @param blob Blob handle. NULL is an error. + * @param dst Destination buffer. May be NULL only when `len` is zero. + * @param len Maximum number of bytes to read. + * @param bytes_read Receives the number of bytes actually written to `dst`, + * never more than `len`. Must not be NULL. Written only on + * success. + * @return 0 on success, -1 on error + */ +int32_t lance_blob_file_read_up_to( + LanceBlobFile* blob, + uint8_t* dst, + size_t len, + size_t* bytes_read +); + +/** + * Read exactly `len` bytes starting at `offset`, without moving the cursor. + * + * `offset` is blob-relative. A non-empty range that ends past the blob size, + * or an `offset` plus `len` that overflows 64 bits, is an error; `len` 0 + * succeeds without checking `offset`. Nothing is written to `dst` on error. + * + * @param blob Blob handle. NULL is an error. + * @param offset Byte offset from the start of the blob. + * @param dst Destination buffer of at least `len` bytes. May be NULL only + * when `len` is zero. + * @param len Number of bytes to read. Zero is a no-op that succeeds. + * @return 0 on success, -1 on error + */ +int32_t lance_blob_file_read_range( + const LanceBlobFile* blob, + uint64_t offset, + uint8_t* dst, + size_t len +); + +/** + * Move the cursor to `pos`. + * + * Seeking past the end of the blob is allowed, mirroring the underlying Lance + * API; a subsequent read then returns zero bytes. + * + * @param blob Blob handle. NULL is an error. + * @param pos New cursor position, in bytes from the start of the blob. + * @return 0 on success, -1 on error + */ +int32_t lance_blob_file_seek(LanceBlobFile* blob, uint64_t pos); + +/** + * Report the current cursor position. + * + * @param blob Blob handle. NULL is an error. + * @param pos Receives the cursor position in bytes from the start of the + * blob. Must not be NULL. Written only on success. + * @return 0 on success, -1 on error + */ +int32_t lance_blob_file_tell(const LanceBlobFile* blob, uint64_t* pos); + +/** + * Close a blob handle and free it. + * + * Call exactly once per non-NULL handle; the handle is invalid afterwards. + * NULL is a no-op. Never fails and leaves the pending error untouched. + * + * @param blob Blob handle, or NULL. + */ +void lance_blob_file_close(LanceBlobFile* blob); + /* ─── Scanner builder ─── */ /** diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp index b3a44f4..fcb1197 100644 --- a/include/lance/lance.hpp +++ b/include/lance/lance.hpp @@ -213,6 +213,73 @@ class FtsQueryContext { const LanceFtsQueryContext* c_handle() const { return handle_.get(); } }; +// ─── Blob file ─────────────────────────────────────────────────────────────── + +/// RAII handle over one value of a Blob v2 column, from `Dataset::take_blobs()` +/// or `take_blobs_by_indices()`. Stays usable after the Dataset is destroyed. +/// `read()` and `read_up_to()` advance the cursor, `read_range()` does not. +/// Not thread-safe per handle. +class BlobFile { + Handle handle_; + +public: + /// Adopt a handle from the C API; closed on destruction. + explicit BlobFile(LanceBlobFile* blob) : handle_(blob) {} + + /// Size of the blob in bytes. Independent of the cursor. + uint64_t size() const { + uint64_t n = lance_blob_file_size(handle_.get()); + if (lance_last_error_code() != LANCE_OK) check_error(); + return n; + } + + /// Read from the cursor to the end (the whole blob when the cursor is 0). + std::vector read() { + uint64_t blob_size = size(); + uint64_t cursor = tell(); + uint64_t remaining = cursor >= blob_size ? 0 : blob_size - cursor; + std::vector out(static_cast(remaining)); + if (lance_blob_file_read(handle_.get(), out.data(), out.size()) != 0) + check_error(); + return out; + } + + /// Read at most `len` bytes from the current cursor, advancing it past + /// them. The result is shorter than `len` at the end of the blob. + std::vector read_up_to(size_t len) { + std::vector out(len); + size_t bytes_read = 0; + if (lance_blob_file_read_up_to( + handle_.get(), out.data(), len, &bytes_read) != 0) + check_error(); + out.resize(bytes_read); + return out; + } + + /// Read exactly `len` bytes at `offset` without moving the cursor. The + /// range must lie within the blob. + std::vector read_range(uint64_t offset, size_t len) const { + std::vector out(len); + if (lance_blob_file_read_range( + handle_.get(), offset, out.data(), len) != 0) + check_error(); + return out; + } + + /// Move the cursor. Seeking past the end is allowed; reads then return + /// no bytes. + void seek(uint64_t pos) { + if (lance_blob_file_seek(handle_.get(), pos) != 0) check_error(); + } + + /// Current cursor position, in bytes from the start of the blob. + uint64_t tell() const { + uint64_t pos = 0; + if (lance_blob_file_tell(handle_.get(), &pos) != 0) check_error(); + return pos; + } +}; + // ─── Dataset ───────────────────────────────────────────────────────────────── class Dataset { @@ -229,6 +296,24 @@ class Dataset { return kv; } + /// Move raw handles into RAII owners. `blobs` must be reserved up front so + /// nothing can throw while handles are still unowned. + /// + /// Leak-freedom also rests on the C side: `lance_dataset_take_blobs*` fill + /// `out` all-or-nothing and leave it untouched on error. A partial fill + /// before an error would leak, because `check_error()` throws before this + /// runs and `raw` owns nothing. + static void adopt_blobs(const std::vector& raw, + std::vector>& blobs) { + for (auto* blob : raw) { + if (blob) { + blobs.emplace_back(BlobFile(blob)); + } else { + blobs.emplace_back(std::nullopt); + } + } + } + public: /// Open a dataset at the given URI. Pass `version` = 0 (the default) for /// the latest, or a specific version id from `versions()` to check out @@ -766,6 +851,43 @@ class Dataset { } } + /// Take blob handles by dataset row ID; element `i` is for `row_ids[i]`, + /// `std::nullopt` for a null blob value. + std::vector> take_blobs( + const uint64_t* row_ids, size_t num_row_ids, + const std::string& column) const { + std::vector raw(num_row_ids, nullptr); + std::vector> blobs; + blobs.reserve(num_row_ids); + // An empty vector's data() may be null, which the C side rejects. + if (num_row_ids > 0 && + lance_dataset_take_blobs(handle_.get(), row_ids, num_row_ids, + column.c_str(), raw.data()) != 0) { + check_error(); + } + adopt_blobs(raw, blobs); + return blobs; + } + + /// Take blob handles by 0-based row index, with the same ownership and + /// null-slot semantics as the overload above. + std::vector> take_blobs_by_indices( + const uint64_t* indices, size_t num_indices, + const std::string& column) const { + std::vector raw(num_indices, nullptr); + std::vector> blobs; + blobs.reserve(num_indices); + // Same empty-request shortcut as take_blobs(). + if (num_indices > 0 && + lance_dataset_take_blobs_by_indices( + handle_.get(), indices, num_indices, + column.c_str(), raw.data()) != 0) { + check_error(); + } + adopt_blobs(raw, blobs); + return blobs; + } + /// Create a Scanner builder for this dataset. Scanner scan() const; diff --git a/src/blob.rs b/src/blob.rs new file mode 100644 index 0000000..54c7e56 --- /dev/null +++ b/src/blob.rs @@ -0,0 +1,344 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The Lance Authors + +//! Blob v2 C API: take per-row blob handles and read their bytes. +//! +//! A [`LanceBlobFile`] wraps an upstream `BlobFile`, so it stays usable after +//! the dataset handle is closed. [`lance_blob_file_read`] and +//! [`lance_blob_file_read_up_to`] advance the cursor, +//! [`lance_blob_file_read_range`] does not. + +use std::ffi::c_char; +use std::ptr; + +use lance::dataset::BlobFile; +use lance_core::Result; + +use crate::dataset::LanceDataset; +use crate::error::{ffi_try, swallow_unwind}; +use crate::helpers; +use crate::runtime::block_on; + +/// Opaque handle to one blob value; independent of the dataset handle. +pub struct LanceBlobFile { + inner: BlobFile, +} + +/// Row addressing used by a take entry point. +#[derive(Clone, Copy)] +enum TakeBy { + /// `_rowid` values. + RowIds, + /// Zero-based row offsets. + Indices, +} + +impl TakeBy { + /// C names of the identifier array and its count, for error messages. + fn param_names(self) -> (&'static str, &'static str) { + match self { + Self::RowIds => ("row_ids", "num_row_ids"), + Self::Indices => ("indices", "num_indices"), + } + } +} + +// --------------------------------------------------------------------------- +// Taking blob handles +// --------------------------------------------------------------------------- + +/// Take blob handles by dataset row ID. See `lance.h` for the full contract. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_dataset_take_blobs( + dataset: *const LanceDataset, + row_ids: *const u64, + num_row_ids: usize, + column: *const c_char, + out: *mut *mut LanceBlobFile, +) -> i32 { + ffi_try!( + unsafe { + dataset_take_blobs_inner(dataset, row_ids, num_row_ids, column, out, TakeBy::RowIds) + }, + neg + ) +} + +/// Take blob handles by row index (offset). See `lance.h` for the full contract. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_dataset_take_blobs_by_indices( + dataset: *const LanceDataset, + indices: *const u64, + num_indices: usize, + column: *const c_char, + out: *mut *mut LanceBlobFile, +) -> i32 { + ffi_try!( + unsafe { + dataset_take_blobs_inner(dataset, indices, num_indices, column, out, TakeBy::Indices) + }, + neg + ) +} + +unsafe fn dataset_take_blobs_inner( + dataset: *const LanceDataset, + ids: *const u64, + num_ids: usize, + column: *const c_char, + out: *mut *mut LanceBlobFile, + take_by: TakeBy, +) -> Result { + if dataset.is_null() { + return Err(lance_core::Error::invalid_input("dataset must not be NULL")); + } + if out.is_null() { + return Err(lance_core::Error::invalid_input("out must not be NULL")); + } + if column.is_null() { + return Err(lance_core::Error::invalid_input("column must not be NULL")); + } + if num_ids > 0 && ids.is_null() { + let (ids_param, count_param) = take_by.param_names(); + return Err(lance_core::Error::invalid_input(format!( + "{ids_param} must not be NULL when {count_param} = {num_ids}" + ))); + } + let column = + unsafe { helpers::parse_c_string(column)? }.expect("column was checked for NULL above"); + + // Nothing to take; `out` stays untouched. + if num_ids == 0 { + return Ok(0); + } + + let ds = unsafe { &*dataset }; + let id_slice = unsafe { std::slice::from_raw_parts(ids, num_ids) }; + + let snap = ds.snapshot(); + // Upstream reports an unknown column as FieldNotFound, which reaches C as + // LANCE_ERR_INTERNAL; make it an invalid argument like the other take + // entry points do. A non-blob column is already invalid input upstream. + if snap.schema().field(column).is_none() { + return Err(lance_core::Error::invalid_input(format!( + "column '{column}' does not exist in the dataset schema" + ))); + } + let blobs = match take_by { + TakeBy::RowIds => block_on(snap.take_blobs(id_slice, column))?, + TakeBy::Indices => block_on(snap.take_blobs_by_indices(id_slice, column))?, + }; + + // Never report success with part of `out` unwritten. + if blobs.len() != num_ids { + return Err(lance_core::Error::internal(format!( + "expected {num_ids} blob handles, got {}", + blobs.len() + ))); + } + + // Every failure above returns before `out` is touched. + for (i, blob) in blobs.into_iter().enumerate() { + let handle = match blob { + Some(inner) => Box::into_raw(Box::new(LanceBlobFile { inner })), + None => ptr::null_mut(), + }; + unsafe { ptr::write_unaligned(out.add(i), handle) }; + } + Ok(0) +} + +// --------------------------------------------------------------------------- +// Blob handle accessors +// --------------------------------------------------------------------------- + +/// Borrow a handle, rejecting NULL with a message naming the parameter. +unsafe fn blob_ref<'a>(blob: *const LanceBlobFile) -> Result<&'a LanceBlobFile> { + if blob.is_null() { + return Err(lance_core::Error::invalid_input("blob must not be NULL")); + } + Ok(unsafe { &*blob }) +} + +/// Return the blob size in bytes. See `lance.h` for the full contract. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_blob_file_size(blob: *const LanceBlobFile) -> u64 { + ffi_try!(unsafe { blob_file_size_inner(blob) }, 0) +} + +unsafe fn blob_file_size_inner(blob: *const LanceBlobFile) -> Result { + Ok(unsafe { blob_ref(blob)? }.inner.size()) +} + +/// Read from the cursor to the end of the blob. See `lance.h` for the full +/// contract. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_blob_file_read( + blob: *mut LanceBlobFile, + dst: *mut u8, + dst_len: usize, +) -> i32 { + ffi_try!(unsafe { blob_file_read_inner(blob, dst, dst_len) }, neg) +} + +unsafe fn blob_file_read_inner( + blob: *mut LanceBlobFile, + dst: *mut u8, + dst_len: usize, +) -> Result { + let handle = unsafe { blob_ref(blob)? }; + let size = handle.inner.size(); + let cursor = block_on(handle.inner.tell())?; + let remaining = size.saturating_sub(cursor); + + if (dst_len as u64) < remaining { + return Err(lance_core::Error::invalid_input(format!( + "dst_len {dst_len} is smaller than the {remaining} bytes remaining from cursor {cursor} (blob size {size})" + ))); + } + if dst.is_null() && remaining > 0 { + return Err(lance_core::Error::invalid_input(format!( + "dst must not be NULL when {remaining} bytes remain from cursor {cursor} (blob size {size})" + ))); + } + + let bytes = block_on(handle.inner.read())?; + if !bytes.is_empty() { + // The read stops at the blob end, so `bytes` fits in the `dst_len` + // checked above; size the slice to what is actually written. + let dst = unsafe { std::slice::from_raw_parts_mut(dst, bytes.len()) }; + dst.copy_from_slice(&bytes); + } + Ok(0) +} + +/// Read at most `len` bytes from the cursor. See `lance.h` for the full +/// contract. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_blob_file_read_up_to( + blob: *mut LanceBlobFile, + dst: *mut u8, + len: usize, + bytes_read: *mut usize, +) -> i32 { + ffi_try!( + unsafe { blob_file_read_up_to_inner(blob, dst, len, bytes_read) }, + neg + ) +} + +unsafe fn blob_file_read_up_to_inner( + blob: *mut LanceBlobFile, + dst: *mut u8, + len: usize, + bytes_read: *mut usize, +) -> Result { + let handle = unsafe { blob_ref(blob)? }; + if bytes_read.is_null() { + return Err(lance_core::Error::invalid_input( + "bytes_read must not be NULL", + )); + } + if dst.is_null() && len > 0 { + return Err(lance_core::Error::invalid_input(format!( + "dst must not be NULL when len = {len}" + ))); + } + + let bytes = block_on(handle.inner.read_up_to(len))?; + if !bytes.is_empty() { + // Upstream caps the read at `len`; size the slice to what is actually + // written. + let dst = unsafe { std::slice::from_raw_parts_mut(dst, bytes.len()) }; + dst.copy_from_slice(&bytes); + } + unsafe { ptr::write_unaligned(bytes_read, bytes.len()) }; + Ok(0) +} + +/// Read `len` bytes at `offset`, leaving the cursor alone. See `lance.h` for +/// the full contract. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_blob_file_read_range( + blob: *const LanceBlobFile, + offset: u64, + dst: *mut u8, + len: usize, +) -> i32 { + ffi_try!( + unsafe { blob_file_read_range_inner(blob, offset, dst, len) }, + neg + ) +} + +unsafe fn blob_file_read_range_inner( + blob: *const LanceBlobFile, + offset: u64, + dst: *mut u8, + len: usize, +) -> Result { + let handle = unsafe { blob_ref(blob)? }; + let end = offset.checked_add(len as u64).ok_or_else(|| { + lance_core::Error::invalid_input(format!( + "offset {offset} plus len {len} overflows a 64-bit byte range" + )) + })?; + + if len == 0 { + return Ok(0); + } + if dst.is_null() { + return Err(lance_core::Error::invalid_input(format!( + "dst must not be NULL when len = {len}" + ))); + } + + // Bounds are checked upstream against the blob size. + let bytes = block_on(handle.inner.read_range(offset..end))?; + unsafe { std::slice::from_raw_parts_mut(dst, len) }.copy_from_slice(&bytes); + Ok(0) +} + +/// Move the cursor. See `lance.h` for the full contract. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_blob_file_seek(blob: *mut LanceBlobFile, pos: u64) -> i32 { + ffi_try!(unsafe { blob_file_seek_inner(blob, pos) }, neg) +} + +unsafe fn blob_file_seek_inner(blob: *mut LanceBlobFile, pos: u64) -> Result { + let handle = unsafe { blob_ref(blob)? }; + // Seeking past the end is allowed, as upstream; reads then return nothing. + block_on(handle.inner.seek(pos))?; + Ok(0) +} + +/// Report the cursor. See `lance.h` for the full contract. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_blob_file_tell(blob: *const LanceBlobFile, pos: *mut u64) -> i32 { + ffi_try!(unsafe { blob_file_tell_inner(blob, pos) }, neg) +} + +unsafe fn blob_file_tell_inner(blob: *const LanceBlobFile, pos: *mut u64) -> Result { + let handle = unsafe { blob_ref(blob)? }; + if pos.is_null() { + return Err(lance_core::Error::invalid_input("pos must not be NULL")); + } + let cursor = block_on(handle.inner.tell())?; + unsafe { ptr::write_unaligned(pos, cursor) }; + Ok(0) +} + +/// Close a blob handle. See `lance.h` for the full contract. +/// +/// `swallow_unwind` rather than `ffi_try!`, so a pending error survives the +/// close. Dropping the `BlobFile` releases its resources; upstream `close()` +/// only sets a flag. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_blob_file_close(blob: *mut LanceBlobFile) { + if blob.is_null() { + return; + } + swallow_unwind("lance_blob_file_close", || unsafe { + drop(Box::from_raw(blob)); + }); +} diff --git a/src/lib.rs b/src/lib.rs index 18b355b..3481760 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -24,6 +24,7 @@ mod add_columns; mod alter_columns; mod async_dispatcher; mod batch; +mod blob; mod compact; mod data_statistics; mod dataset; @@ -52,6 +53,7 @@ mod writer; pub use add_columns::*; pub use alter_columns::*; pub use batch::*; +pub use blob::*; pub use compact::*; pub use data_statistics::*; pub use dataset::*; diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs index 03e176c..7f4ea31 100644 --- a/tests/c_api_test.rs +++ b/tests/c_api_test.rs @@ -14482,7 +14482,10 @@ fn blob_batch(schema: &Arc, base_id: u32) -> RecordBatch { /// Two-fragment v2.2 dataset with a blob column, a plain binary column and an /// id column; one [`blob_batch`] per entry of [`BLOB_FRAGMENT_BASE_IDS`]. -fn create_blob_v2_dataset() -> (tempfile::TempDir, String) { +/// +/// With `enable_stable_row_ids` a `_rowid` goes through the row id index +/// instead of being the row address. +fn create_blob_v2_dataset(enable_stable_row_ids: bool) -> (tempfile::TempDir, String) { let tmp = tempfile::tempdir().unwrap(); let uri = tmp.path().join("blob_ds").to_str().unwrap().to_string(); @@ -14509,6 +14512,7 @@ fn create_blob_v2_dataset() -> (tempfile::TempDir, String) { }, // Blob v2 is a 2.2 storage feature. data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2), + enable_stable_row_ids, ..Default::default() }; Dataset::write( @@ -14669,7 +14673,7 @@ fn assert_blob_description_field(schema: &Schema, name: &str) { #[test] fn test_scanner_blob_handling_all_binary_materializes_bytes() { - let (_tmp, uri) = create_blob_v2_dataset(); + let (_tmp, uri) = create_blob_v2_dataset(false); let c_uri = c_str(&uri); let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; assert!(!ds.is_null()); @@ -14712,7 +14716,7 @@ fn test_scanner_blob_handling_all_binary_materializes_bytes() { #[test] fn test_scanner_blob_handling_defaults_to_descriptions() { - let (_tmp, uri) = create_blob_v2_dataset(); + let (_tmp, uri) = create_blob_v2_dataset(false); let c_uri = c_str(&uri); let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; assert!(!ds.is_null()); @@ -14757,7 +14761,7 @@ fn test_scanner_blob_handling_defaults_to_descriptions() { #[test] fn test_scanner_blob_handling_all_descriptions() { - let (_tmp, uri) = create_blob_v2_dataset(); + let (_tmp, uri) = create_blob_v2_dataset(false); let c_uri = c_str(&uri); let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; assert!(!ds.is_null()); @@ -14797,7 +14801,7 @@ fn test_scanner_blob_handling_all_descriptions() { #[test] fn test_scanner_blob_handling_rejected_after_scan_started() { - let (_tmp, uri) = create_blob_v2_dataset(); + let (_tmp, uri) = create_blob_v2_dataset(false); let c_uri = c_str(&uri); let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; assert!(!ds.is_null()); @@ -14830,7 +14834,7 @@ fn test_scanner_blob_handling_rejected_after_scan_started() { #[test] fn test_scanner_blob_handling_rejects_invalid_values() { - let (_tmp, uri) = create_blob_v2_dataset(); + let (_tmp, uri) = create_blob_v2_dataset(false); let c_uri = c_str(&uri); let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; assert!(!ds.is_null()); @@ -14868,7 +14872,7 @@ fn test_scanner_blob_handling_rejects_invalid_values() { #[test] fn test_scanner_blob_handling_all_binary_with_fragment_ids() { - let (_tmp, uri) = create_blob_v2_dataset(); + let (_tmp, uri) = create_blob_v2_dataset(false); let c_uri = c_str(&uri); let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; assert!(!ds.is_null()); @@ -14916,3 +14920,984 @@ fn test_scanner_blob_handling_all_binary_with_fragment_ids() { unsafe { lance_scanner_close(scanner) }; unsafe { lance_dataset_close(ds) }; } + +// --------------------------------------------------------------------------- +// Blob v2 random access +// --------------------------------------------------------------------------- + +/// Row offset (in `id` order) of the packed blob used by the cursor tests. +const PACKED_BLOB_ROW: usize = 1; +/// Row offset (in `id` order) of the dedicated blob used by the cursor tests. +const DEDICATED_BLOB_ROW: usize = 2; + +/// Expected bytes at row offset `row` (in `id` order); `None` for the null row. +fn expected_blob(row: usize) -> Option> { + let fragment = row / BLOB_ROW_SIZES.len(); + let seed = BLOB_FRAGMENT_BASE_IDS[fragment] as usize; + BLOB_ROW_SIZES[row % BLOB_ROW_SIZES.len()].map(|len| blob_payload(len, seed)) +} + +/// Row ids of every row in `id` order, read through the scanner. +fn scan_blob_row_ids(dataset: *const LanceDataset) -> Vec { + let id_column = c_str("id"); + let columns: [*const c_char; 2] = [id_column.as_ptr(), ptr::null()]; + let scanner = unsafe { lance_scanner_new(dataset, columns.as_ptr(), ptr::null()) }; + assert!(!scanner.is_null()); + assert_eq!(unsafe { lance_scanner_with_row_id(scanner, true) }, 0); + + let mut stream = FFI_ArrowArrayStream::empty(); + assert_eq!( + unsafe { lance_scanner_to_arrow_stream(scanner, &mut stream) }, + 0 + ); + let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut stream) }.unwrap(); + + let mut rows: Vec<(u32, u64)> = Vec::new(); + for batch in reader { + let batch = batch.unwrap(); + let ids = batch + .column_by_name("id") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + let row_ids = batch + .column_by_name("_rowid") + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + for row in 0..batch.num_rows() { + rows.push((ids.value(row), row_ids.value(row))); + } + } + unsafe { lance_scanner_close(scanner) }; + + rows.sort_by_key(|(id, _)| *id); + rows.into_iter().map(|(_, row_id)| row_id).collect() +} + +/// Take every blob of the dataset by row ID, asserting the call succeeds. +fn take_all_blobs(dataset: *const LanceDataset) -> Vec<*mut LanceBlobFile> { + let row_ids = scan_blob_row_ids(dataset); + assert_eq!( + row_ids.len(), + 2 * BLOB_ROW_SIZES.len(), + "two fragments of five rows" + ); + + let column = c_str("blob"); + let mut handles = vec![ptr::null_mut::(); row_ids.len()]; + let rc = unsafe { + lance_dataset_take_blobs( + dataset, + row_ids.as_ptr(), + row_ids.len(), + column.as_ptr(), + handles.as_mut_ptr(), + ) + }; + assert_eq!(rc, 0, "take_blobs failed: {}", take_last_error_message()); + handles +} + +/// Read a handle from its current cursor to the end, asserting success. +fn read_blob_to_end(handle: *mut LanceBlobFile) -> Vec { + let size = unsafe { lance_blob_file_size(handle) }; + let mut cursor = 0u64; + assert_eq!(unsafe { lance_blob_file_tell(handle, &mut cursor) }, 0); + let mut buffer = vec![0u8; size.saturating_sub(cursor) as usize]; + assert_eq!( + unsafe { lance_blob_file_read(handle, buffer.as_mut_ptr(), buffer.len()) }, + 0, + "read failed: {}", + take_last_error_message() + ); + buffer +} + +/// Close every handle; NULL slots are accepted. +fn close_blob_handles(handles: &[*mut LanceBlobFile]) { + for handle in handles { + unsafe { lance_blob_file_close(*handle) }; + } +} + +#[test] +fn test_blob_take_by_row_ids_covers_every_storage_layout() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let handles = take_all_blobs(ds); + + // Input order, both fragments: inline, packed, dedicated, empty, null. + for (row, handle) in handles.iter().copied().enumerate() { + match expected_blob(row) { + None => assert!( + handle.is_null(), + "row {row}: a null blob must yield a NULL slot" + ), + Some(expected) => { + assert!( + !handle.is_null(), + "row {row}: a non-null blob must yield a handle" + ); + assert_eq!( + unsafe { lance_blob_file_size(handle) }, + expected.len() as u64, + "row {row}: size must match the written payload" + ); + assert_eq!(read_blob_to_end(handle), expected, "row {row}: bytes"); + } + } + } + + // An empty blob is a real handle of size 0, not a NULL slot. + let empty = handles[3]; + assert!(!empty.is_null()); + assert_eq!(unsafe { lance_blob_file_size(empty) }, 0); + let mut untouched = [0xABu8; 4]; + assert_eq!( + unsafe { lance_blob_file_read(empty, untouched.as_mut_ptr(), untouched.len()) }, + 0, + "reading an empty blob failed: {}", + take_last_error_message() + ); + assert_eq!(untouched, [0xABu8; 4], "an empty blob must write no bytes"); + + close_blob_handles(&handles); + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_take_by_indices_matches_take_by_row_ids() { + assert_blob_take_by_indices_matches_row_ids(false); +} + +#[test] +fn test_blob_take_by_indices_matches_take_by_row_ids_with_stable_row_ids() { + // With stable row ids a `_rowid` is not the row address. + assert_blob_take_by_indices_matches_row_ids(true); +} + +fn assert_blob_take_by_indices_matches_row_ids(enable_stable_row_ids: bool) { + let (_tmp, uri) = create_blob_v2_dataset(enable_stable_row_ids); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let by_row_id = take_all_blobs(ds); + + let indices = (0..2 * BLOB_ROW_SIZES.len() as u64).collect::>(); + let column = c_str("blob"); + let mut by_index = vec![ptr::null_mut::(); indices.len()]; + let rc = unsafe { + lance_dataset_take_blobs_by_indices( + ds, + indices.as_ptr(), + indices.len(), + column.as_ptr(), + by_index.as_mut_ptr(), + ) + }; + assert_eq!( + rc, + 0, + "take_blobs_by_indices failed: {}", + take_last_error_message() + ); + + for row in 0..indices.len() { + match (by_row_id[row].is_null(), by_index[row].is_null()) { + (true, true) => continue, + (false, false) => assert_eq!( + read_blob_to_end(by_index[row]), + read_blob_to_end(by_row_id[row]), + "row {row}: both addressing schemes must return the same bytes" + ), + (row_id_null, index_null) => panic!( + "row {row}: NULL slots disagree (by row id: {row_id_null}, by index: {index_null})" + ), + } + } + + close_blob_handles(&by_row_id); + close_blob_handles(&by_index); + unsafe { lance_dataset_close(ds) }; +} + +/// Rows requested out of storage order: two fragments, a repeated row, and a +/// null blob in the middle. +const PERMUTED_ROWS: [usize; 5] = [7, 2, 2, 9, 0]; + +#[test] +fn test_blob_take_preserves_permuted_and_duplicated_input_order() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let all_row_ids = scan_blob_row_ids(ds); + let row_ids = PERMUTED_ROWS + .iter() + .map(|row| all_row_ids[*row]) + .collect::>(); + let indices = PERMUTED_ROWS + .iter() + .map(|row| *row as u64) + .collect::>(); + let column = c_str("blob"); + + for (entry_point, ids) in [("row ids", &row_ids), ("indices", &indices)] { + let mut handles = vec![ptr::null_mut::(); ids.len()]; + let rc = if entry_point == "row ids" { + unsafe { + lance_dataset_take_blobs( + ds, + ids.as_ptr(), + ids.len(), + column.as_ptr(), + handles.as_mut_ptr(), + ) + } + } else { + unsafe { + lance_dataset_take_blobs_by_indices( + ds, + ids.as_ptr(), + ids.len(), + column.as_ptr(), + handles.as_mut_ptr(), + ) + } + }; + assert_eq!( + rc, + 0, + "{entry_point}: take failed: {}", + take_last_error_message() + ); + + for (slot, row) in PERMUTED_ROWS.iter().copied().enumerate() { + let handle = handles[slot]; + match expected_blob(row) { + None => assert!( + handle.is_null(), + "{entry_point}: slot {slot} (row {row}) must be NULL" + ), + Some(expected) => { + assert!( + !handle.is_null(), + "{entry_point}: slot {slot} (row {row}) must hold a handle" + ); + assert_eq!( + unsafe { lance_blob_file_size(handle) }, + expected.len() as u64, + "{entry_point}: slot {slot} (row {row}) size" + ); + assert_eq!( + read_blob_to_end(handle), + expected, + "{entry_point}: slot {slot} (row {row}) bytes" + ); + } + } + } + + // Duplicate rows get independent handles with their own cursors. + assert_eq!(unsafe { lance_blob_file_seek(handles[1], 0) }, 0); + let mut first = u64::MAX; + let mut second = u64::MAX; + assert_eq!(unsafe { lance_blob_file_tell(handles[1], &mut first) }, 0); + assert_eq!(unsafe { lance_blob_file_tell(handles[2], &mut second) }, 0); + assert_eq!(first, 0, "{entry_point}: the rewound duplicate"); + assert_eq!( + second, + unsafe { lance_blob_file_size(handles[2]) }, + "{entry_point}: duplicates must not share a cursor" + ); + + close_blob_handles(&handles); + } + + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_fixture_uses_all_three_storage_layouts() { + // The fixture must really produce three storage kinds; only the Rust API + // exposes the kind. + let (_tmp, uri) = create_blob_v2_dataset(false); + let kinds = lance_c::runtime::block_on(async { + let dataset = Arc::new(Dataset::open(&uri).await.unwrap()); + let blobs = dataset + .take_blobs_by_indices(&[0, 1, 2], "blob") + .await + .unwrap(); + blobs + .into_iter() + .map(|blob| blob.unwrap().kind()) + .collect::>() + }); + + use lance_core::datatypes::BlobKind; + assert_eq!( + kinds, + vec![BlobKind::Inline, BlobKind::Packed, BlobKind::Dedicated], + "the 8, 128 and 1024 byte rows must land in three different layouts" + ); +} + +#[test] +fn test_blob_cursor_advances_only_on_sequential_reads() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let handles = take_all_blobs(ds); + let blob = handles[PACKED_BLOB_ROW]; + let payload = expected_blob(PACKED_BLOB_ROW).unwrap(); + let size = unsafe { lance_blob_file_size(blob) }; + assert_eq!(size, payload.len() as u64); + + let mut cursor = u64::MAX; + assert_eq!(unsafe { lance_blob_file_tell(blob, &mut cursor) }, 0); + assert_eq!(cursor, 0, "a fresh handle starts at the beginning"); + + // A short read moves the cursor by exactly what it read. + let mut buffer = vec![0u8; 32]; + let mut bytes_read = usize::MAX; + assert_eq!( + unsafe { + lance_blob_file_read_up_to(blob, buffer.as_mut_ptr(), buffer.len(), &mut bytes_read) + }, + 0, + "read_up_to failed: {}", + take_last_error_message() + ); + assert_eq!(bytes_read, 32); + assert_eq!(buffer, payload[..32]); + assert_eq!(unsafe { lance_blob_file_tell(blob, &mut cursor) }, 0); + assert_eq!(cursor, 32); + + // Asking for more than remains reads only what is left. + let mut rest = vec![0u8; payload.len()]; + assert_eq!( + unsafe { lance_blob_file_read_up_to(blob, rest.as_mut_ptr(), rest.len(), &mut bytes_read) }, + 0, + "read_up_to failed: {}", + take_last_error_message() + ); + assert_eq!(bytes_read, payload.len() - 32); + assert_eq!(&rest[..bytes_read], &payload[32..]); + assert_eq!(unsafe { lance_blob_file_tell(blob, &mut cursor) }, 0); + assert_eq!(cursor, size); + + // At the end, read_up_to reports zero bytes instead of failing. + assert_eq!( + unsafe { lance_blob_file_read_up_to(blob, rest.as_mut_ptr(), rest.len(), &mut bytes_read) }, + 0 + ); + assert_eq!(bytes_read, 0); + + // seek positions the cursor, and read then starts there. + assert_eq!(unsafe { lance_blob_file_seek(blob, 64) }, 0); + assert_eq!(unsafe { lance_blob_file_tell(blob, &mut cursor) }, 0); + assert_eq!(cursor, 64); + let mut tail = vec![0u8; (size - 64) as usize]; + assert_eq!( + unsafe { lance_blob_file_read(blob, tail.as_mut_ptr(), tail.len()) }, + 0, + "read failed: {}", + take_last_error_message() + ); + assert_eq!(tail, payload[64..]); + + // Seeking past the end is allowed; the read that follows writes nothing. + assert_eq!(unsafe { lance_blob_file_seek(blob, size + 16) }, 0); + let mut untouched = [0xCDu8; 8]; + assert_eq!( + unsafe { lance_blob_file_read(blob, untouched.as_mut_ptr(), untouched.len()) }, + 0, + "reading past the end failed: {}", + take_last_error_message() + ); + assert_eq!(untouched, [0xCDu8; 8]); + + // read_range is positional and leaves the cursor wherever it was. + assert_eq!(unsafe { lance_blob_file_seek(blob, 5) }, 0); + let mut window = vec![0u8; 16]; + assert_eq!( + unsafe { lance_blob_file_read_range(blob, 40, window.as_mut_ptr(), window.len()) }, + 0, + "read_range failed: {}", + take_last_error_message() + ); + assert_eq!(window, payload[40..56]); + assert_eq!(unsafe { lance_blob_file_tell(blob, &mut cursor) }, 0); + assert_eq!(cursor, 5, "read_range must not move the cursor"); + + close_blob_handles(&handles); + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_read_rejects_buffer_smaller_than_remaining() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let handles = take_all_blobs(ds); + let blob = handles[PACKED_BLOB_ROW]; + let payload = expected_blob(PACKED_BLOB_ROW).unwrap(); + + // One byte short of the whole blob. + let mut buffer = vec![0xEEu8; payload.len() - 1]; + assert_eq!( + unsafe { lance_blob_file_read(blob, buffer.as_mut_ptr(), buffer.len()) }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("dst_len 127"), "{message}"); + assert!(message.contains("128 bytes remaining"), "{message}"); + assert!(message.contains("cursor 0"), "{message}"); + assert!(message.contains("blob size 128"), "{message}"); + assert!( + buffer.iter().all(|byte| *byte == 0xEE), + "a rejected read must not touch the buffer" + ); + + // The same rejection from a non-zero cursor reports the bytes remaining, + // not the blob size. + assert_eq!(unsafe { lance_blob_file_seek(blob, 100) }, 0); + let mut short = vec![0u8; 27]; + assert_eq!( + unsafe { lance_blob_file_read(blob, short.as_mut_ptr(), short.len()) }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("dst_len 27"), "{message}"); + assert!(message.contains("28 bytes remaining"), "{message}"); + assert!(message.contains("cursor 100"), "{message}"); + assert!(message.contains("blob size 128"), "{message}"); + + // An exactly sized buffer succeeds. + let mut exact = vec![0u8; 28]; + assert_eq!( + unsafe { lance_blob_file_read(blob, exact.as_mut_ptr(), exact.len()) }, + 0, + "read failed: {}", + take_last_error_message() + ); + assert_eq!(exact, payload[100..]); + + close_blob_handles(&handles); + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_read_range_rejects_out_of_bounds() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let handles = take_all_blobs(ds); + let blob = handles[PACKED_BLOB_ROW]; + let size = unsafe { lance_blob_file_size(blob) }; + + // Four bytes past the end. + let mut buffer = vec![0x5Au8; 8]; + assert_eq!( + unsafe { lance_blob_file_read_range(blob, size - 4, buffer.as_mut_ptr(), buffer.len()) }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("132"), "{message}"); + assert!(message.contains("exceeds blob size 128"), "{message}"); + assert!( + buffer.iter().all(|byte| *byte == 0x5A), + "a rejected read_range must not touch the buffer" + ); + + // An offset plus length that overflows 64 bits is rejected before any read. + assert_eq!( + unsafe { lance_blob_file_read_range(blob, u64::MAX, buffer.as_mut_ptr(), 2) }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains(&u64::MAX.to_string()), "{message}"); + assert!(message.contains("len 2"), "{message}"); + + // An empty range succeeds and accepts a NULL destination. + assert_eq!( + unsafe { lance_blob_file_read_range(blob, 0, ptr::null_mut(), 0) }, + 0, + "empty read_range failed: {}", + take_last_error_message() + ); + + close_blob_handles(&handles); + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_handles_outlive_the_dataset() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let handles = take_all_blobs(ds); + + // Handles own their readers; the dataset can go first. + unsafe { lance_dataset_close(ds) }; + + for (row, handle) in handles.iter().copied().enumerate() { + let Some(expected) = expected_blob(row) else { + continue; + }; + assert_eq!( + unsafe { lance_blob_file_size(handle) }, + expected.len() as u64, + "row {row}: size after the dataset was closed" + ); + assert_eq!( + read_blob_to_end(handle), + expected, + "row {row}: read after the dataset was closed" + ); + + if expected.is_empty() { + continue; + } + let mut window = vec![0u8; expected.len().min(16)]; + assert_eq!( + unsafe { lance_blob_file_read_range(handle, 0, window.as_mut_ptr(), window.len()) }, + 0, + "row {row}: read_range after the dataset was closed: {}", + take_last_error_message() + ); + assert_eq!(window, expected[..window.len()], "row {row}: range bytes"); + } + + close_blob_handles(&handles); +} + +#[test] +fn test_blob_take_rejects_invalid_arguments() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let row_ids = scan_blob_row_ids(ds); + let blob_column = c_str("blob"); + + // Sentinel that no rejected call may overwrite; never dereferenced. + let sentinel = ptr::without_provenance_mut::(0xDEAD_BEEF); + let mut out = vec![sentinel; row_ids.len()]; + let assert_out_untouched = |out: &[*mut LanceBlobFile], case: &str| { + for (slot, handle) in out.iter().enumerate() { + assert_eq!(*handle, sentinel, "{case}: slot {slot} was written"); + } + }; + + let missing = c_str("does_not_exist"); + assert_eq!( + unsafe { + lance_dataset_take_blobs( + ds, + row_ids.as_ptr(), + row_ids.len(), + missing.as_ptr(), + out.as_mut_ptr(), + ) + }, + -1 + ); + // Read the code first; taking the message clears the error. + assert_eq!( + lance_last_error_code(), + LanceErrorCode::InvalidArgument, + "a misspelled column is a caller error, not an internal one" + ); + let message = take_last_error_message(); + assert!(message.contains("does_not_exist"), "{message}"); + assert_out_untouched(&out, "missing column"); + + let not_a_blob = c_str("raw"); + assert_eq!( + unsafe { + lance_dataset_take_blobs( + ds, + row_ids.as_ptr(), + row_ids.len(), + not_a_blob.as_ptr(), + out.as_mut_ptr(), + ) + }, + -1 + ); + assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); + let message = take_last_error_message(); + assert!(message.contains("raw"), "{message}"); + assert!(message.contains("not a blob column"), "{message}"); + assert_out_untouched(&out, "non-blob column"); + + // Zero identifiers is a no-op success that writes nothing. + assert_eq!( + unsafe { + lance_dataset_take_blobs(ds, ptr::null(), 0, blob_column.as_ptr(), out.as_mut_ptr()) + }, + 0, + "empty take failed: {}", + take_last_error_message() + ); + assert_out_untouched(&out, "zero row ids"); + assert_eq!( + unsafe { + lance_dataset_take_blobs_by_indices( + ds, + ptr::null(), + 0, + blob_column.as_ptr(), + out.as_mut_ptr(), + ) + }, + 0, + "empty take by index failed: {}", + take_last_error_message() + ); + assert_out_untouched(&out, "zero indices"); + + assert_eq!( + unsafe { + lance_dataset_take_blobs(ds, ptr::null(), 1, blob_column.as_ptr(), out.as_mut_ptr()) + }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("row_ids must not be NULL"), "{message}"); + assert!(message.contains("num_row_ids = 1"), "{message}"); + assert_out_untouched(&out, "NULL row_ids"); + + assert_eq!( + unsafe { + lance_dataset_take_blobs_by_indices( + ds, + ptr::null(), + 1, + blob_column.as_ptr(), + out.as_mut_ptr(), + ) + }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("indices must not be NULL"), "{message}"); + assert!(message.contains("num_indices = 1"), "{message}"); + assert_out_untouched(&out, "NULL indices"); + + assert_eq!( + unsafe { + lance_dataset_take_blobs( + ptr::null(), + row_ids.as_ptr(), + row_ids.len(), + blob_column.as_ptr(), + out.as_mut_ptr(), + ) + }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("dataset must not be NULL"), "{message}"); + assert_out_untouched(&out, "NULL dataset"); + + assert_eq!( + unsafe { + lance_dataset_take_blobs( + ds, + row_ids.as_ptr(), + row_ids.len(), + ptr::null(), + out.as_mut_ptr(), + ) + }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("column must not be NULL"), "{message}"); + assert_out_untouched(&out, "NULL column"); + + assert_eq!( + unsafe { + lance_dataset_take_blobs( + ds, + row_ids.as_ptr(), + row_ids.len(), + blob_column.as_ptr(), + ptr::null_mut(), + ) + }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("out must not be NULL"), "{message}"); + + // Invalid UTF-8 in the column name. + let invalid_utf8 = CString::new(b"bl\xFFob".to_vec()).unwrap(); + assert_eq!( + unsafe { + lance_dataset_take_blobs( + ds, + row_ids.as_ptr(), + row_ids.len(), + invalid_utf8.as_ptr(), + out.as_mut_ptr(), + ) + }, + -1 + ); + assert_out_untouched(&out, "invalid UTF-8 column"); + + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_reads_reject_null_destination_and_out_params() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let handles = take_all_blobs(ds); + let blob = handles[PACKED_BLOB_ROW]; + let size = unsafe { lance_blob_file_size(blob) }; + + // A NULL destination is only legal for a request that reads no bytes. + assert_eq!( + unsafe { lance_blob_file_read(blob, ptr::null_mut(), size as usize) }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("dst must not be NULL"), "{message}"); + + let mut bytes_read = usize::MAX; + assert_eq!( + unsafe { lance_blob_file_read_up_to(blob, ptr::null_mut(), 8, &mut bytes_read) }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("dst must not be NULL"), "{message}"); + assert_eq!( + bytes_read, + usize::MAX, + "a rejected read must not report a length" + ); + + assert_eq!( + unsafe { lance_blob_file_read_range(blob, 0, ptr::null_mut(), 8) }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("dst must not be NULL"), "{message}"); + + let mut pos = u64::MAX; + assert_eq!(unsafe { lance_blob_file_tell(blob, ptr::null_mut()) }, -1); + let message = take_last_error_message(); + assert!(message.contains("pos must not be NULL"), "{message}"); + + // None of the rejections moved the cursor. + assert_eq!(unsafe { lance_blob_file_tell(blob, &mut pos) }, 0); + assert_eq!(pos, 0); + + close_blob_handles(&handles); + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_take_rejects_unknown_row_id() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let column = c_str("blob"); + let sentinel = ptr::without_provenance_mut::(0xDEAD_BEEF); + let mut out = [sentinel]; + let unknown = [u64::MAX - 1]; + + assert_eq!( + unsafe { + lance_dataset_take_blobs(ds, unknown.as_ptr(), 1, column.as_ptr(), out.as_mut_ptr()) + }, + -1 + ); + // The row id decodes to a fragment that does not exist; upstream rejects + // the whole call. + let message = take_last_error_message(); + assert!(message.contains("18446744073709551614"), "{message}"); + assert!(message.contains("non-existent fragment"), "{message}"); + assert_eq!(out[0], sentinel, "a rejected take must not write `out`"); + + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_take_by_indices_rejects_out_of_range_index() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let column = c_str("blob"); + let sentinel = ptr::without_provenance_mut::(0xDEAD_BEEF); + let mut out = [sentinel, sentinel]; + // A valid offset next to one just past the end of the dataset. + let indices = [0u64, 2 * BLOB_ROW_SIZES.len() as u64]; + + assert_eq!( + unsafe { + lance_dataset_take_blobs_by_indices( + ds, + indices.as_ptr(), + indices.len(), + column.as_ptr(), + out.as_mut_ptr(), + ) + }, + -1 + ); + // An offset past the end becomes a tombstone address, which upstream + // rejects; the valid slot is not written either. + assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); + let message = take_last_error_message(); + assert!(message.contains("non-existent fragment"), "{message}"); + assert_eq!( + out, + [sentinel, sentinel], + "a rejected take must not write `out`" + ); + + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_read_up_to_requires_bytes_read_out_param() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let handles = take_all_blobs(ds); + let blob = handles[DEDICATED_BLOB_ROW]; + let mut buffer = [0u8; 8]; + + assert_eq!( + unsafe { + lance_blob_file_read_up_to(blob, buffer.as_mut_ptr(), buffer.len(), ptr::null_mut()) + }, + -1 + ); + let message = take_last_error_message(); + assert!(message.contains("bytes_read must not be NULL"), "{message}"); + + // A zero-length request accepts a NULL destination and reports 0 bytes. + let mut bytes_read = usize::MAX; + assert_eq!( + unsafe { lance_blob_file_read_up_to(blob, ptr::null_mut(), 0, &mut bytes_read) }, + 0, + "zero-length read_up_to failed: {}", + take_last_error_message() + ); + assert_eq!(bytes_read, 0); + + close_blob_handles(&handles); + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_blob_null_handle_is_rejected_without_crashing() { + /// Assert that the pending error names the NULL handle. + fn assert_null_handle_reported() { + let message = take_last_error_message(); + assert!(message.contains("blob must not be NULL"), "{message}"); + } + + assert_eq!(unsafe { lance_blob_file_size(ptr::null()) }, 0); + assert_ne!( + lance_last_error_code(), + LanceErrorCode::Ok, + "size must report a NULL handle through the error channel" + ); + assert_null_handle_reported(); + + let mut buffer = [0u8; 4]; + assert_eq!( + unsafe { lance_blob_file_read(ptr::null_mut(), buffer.as_mut_ptr(), buffer.len()) }, + -1 + ); + assert_null_handle_reported(); + let mut bytes_read = 0usize; + assert_eq!( + unsafe { + lance_blob_file_read_up_to( + ptr::null_mut(), + buffer.as_mut_ptr(), + buffer.len(), + &mut bytes_read, + ) + }, + -1 + ); + assert_null_handle_reported(); + assert_eq!( + unsafe { lance_blob_file_read_range(ptr::null(), 0, buffer.as_mut_ptr(), buffer.len()) }, + -1 + ); + assert_null_handle_reported(); + assert_eq!(unsafe { lance_blob_file_seek(ptr::null_mut(), 0) }, -1); + assert_null_handle_reported(); + let mut pos = 0u64; + assert_eq!(unsafe { lance_blob_file_tell(ptr::null(), &mut pos) }, -1); + assert_null_handle_reported(); + + // Closing NULL is a no-op. + unsafe { lance_blob_file_close(ptr::null_mut()) }; +} + +#[test] +fn test_blob_close_keeps_the_pending_error_readable() { + let (_tmp, uri) = create_blob_v2_dataset(false); + let uri_c = c_str(&uri); + let ds = unsafe { lance_dataset_open(uri_c.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let handles = take_all_blobs(ds); + let blob = handles[PACKED_BLOB_ROW]; + let mut too_small = [0u8; 4]; + assert_eq!( + unsafe { lance_blob_file_read(blob, too_small.as_mut_ptr(), too_small.len()) }, + -1 + ); + + // Closing must not clear an error the caller has not read yet. + unsafe { lance_blob_file_close(blob) }; + assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); + let message = take_last_error_message(); + assert!(message.contains("dst_len 4"), "{message}"); + + let rest = handles + .iter() + .copied() + .filter(|handle| *handle != blob) + .collect::>(); + close_blob_handles(&rest); + unsafe { lance_dataset_close(ds) }; +} diff --git a/tests/cpp/test_c_api.c b/tests/cpp/test_c_api.c index 7570110..d9539e9 100644 --- a/tests/cpp/test_c_api.c +++ b/tests/cpp/test_c_api.c @@ -295,6 +295,94 @@ static void test_scanner_blob_handling(const char *blob_uri) { printf("OK\n"); } +/* Byte `i` of every blob payload in the smoke fixture. */ +static uint8_t blob_byte(size_t i) { return (uint8_t)(i * 7 + 3); } + +/* Check that `bytes` are the payload bytes starting at `offset`. */ +static void assert_blob_payload(const uint8_t *bytes, size_t len, size_t offset) { + for (size_t i = 0; i < len; i++) { + ASSERT(bytes[i] == blob_byte(offset + i), "blob payload mismatch"); + } +} + +static void test_take_blobs(const char *blob_uri) { + printf(" test_take_blobs... "); + + LanceDataset *ds = lance_dataset_open(blob_uri, NULL, 0); + ASSERT(ds != NULL, "blob dataset open failed"); + + /* The first fragment holds an inline, a packed, a dedicated, an empty and + * a null blob, in that order. */ + const uint64_t indices[] = {0, 1, 2, 3, 4}; + LanceBlobFile *blobs[5] = {0}; + int32_t rc = lance_dataset_take_blobs_by_indices(ds, indices, 5, "blob", blobs); + ASSERT(rc == 0, "take_blobs_by_indices failed"); + + const uint64_t sizes[] = {8, 128, 1024, 0}; + uint8_t buffer[1024]; + for (size_t i = 0; i < 4; i++) { + ASSERT(blobs[i] != NULL, "a non-null blob should yield a handle"); + uint64_t size = lance_blob_file_size(blobs[i]); + CHECK_OK(); + ASSERT(size == sizes[i], "blob size mismatch"); + rc = lance_blob_file_read(blobs[i], buffer, (size_t)size); + ASSERT(rc == 0, "blob read failed"); + assert_blob_payload(buffer, (size_t)size, 0); + } + ASSERT(blobs[4] == NULL, "a null blob should yield a NULL slot"); + + /* Cursor and positional reads on the packed blob. */ + LanceBlobFile *packed = blobs[1]; + rc = lance_blob_file_seek(packed, 100); + ASSERT(rc == 0, "seek failed"); + size_t bytes_read = 0; + rc = lance_blob_file_read_up_to(packed, buffer, 64, &bytes_read); + ASSERT(rc == 0, "read_up_to failed"); + ASSERT(bytes_read == 28, "read_up_to should stop at the end of the blob"); + assert_blob_payload(buffer, bytes_read, 100); + uint64_t pos = 0; + rc = lance_blob_file_tell(packed, &pos); + ASSERT(rc == 0, "tell failed"); + ASSERT(pos == 128, "cursor should be at the end"); + rc = lance_blob_file_read_range(packed, 40, buffer, 16); + ASSERT(rc == 0, "read_range failed"); + assert_blob_payload(buffer, 16, 40); + rc = lance_blob_file_tell(packed, &pos); + ASSERT(rc == 0 && pos == 128, "read_range must not move the cursor"); + + /* A buffer smaller than the remaining bytes is rejected, not truncated. */ + rc = lance_blob_file_seek(packed, 0); + ASSERT(rc == 0, "seek failed"); + rc = lance_blob_file_read(packed, buffer, 64); + ASSERT(rc == -1, "a short buffer should be rejected"); + ASSERT(lance_last_error_code() == LANCE_ERR_INVALID_ARGUMENT, "wrong error code"); + const char *msg = lance_last_error_message(); + ASSERT(msg != NULL, "an error message is expected"); + lance_free_string(msg); + + /* A column that is not a blob column is rejected and leaves `out` alone. */ + LanceBlobFile *untouched[5] = {0}; + rc = lance_dataset_take_blobs_by_indices(ds, indices, 5, "raw", untouched); + ASSERT(rc == -1, "a non-blob column should be rejected"); + ASSERT(lance_last_error_code() == LANCE_ERR_INVALID_ARGUMENT, "wrong error code"); + msg = lance_last_error_message(); + if (msg) lance_free_string(msg); + for (size_t i = 0; i < 5; i++) { + ASSERT(untouched[i] == NULL, "out must stay untouched on error"); + } + + /* Handles stay readable after the dataset is closed. */ + lance_dataset_close(ds); + rc = lance_blob_file_read_range(blobs[2], 0, buffer, 16); + ASSERT(rc == 0, "read after the dataset was closed failed"); + assert_blob_payload(buffer, 16, 0); + + for (size_t i = 0; i < 5; i++) { + lance_blob_file_close(blobs[i]); /* NULL-safe for the null slot */ + } + printf("OK\n"); +} + static void test_versions(const char *uri) { printf(" test_versions... "); @@ -1140,6 +1228,7 @@ int main(int argc, char **argv) { test_scan(uri); test_scan_with_limit(uri); test_scanner_blob_handling(blob_uri); + test_take_blobs(blob_uri); test_versions(uri); test_restore_to_current(uri); test_error_handling(); diff --git a/tests/cpp/test_cpp_api.cpp b/tests/cpp/test_cpp_api.cpp index 6f1ebc8..9f4ded0 100644 --- a/tests/cpp/test_cpp_api.cpp +++ b/tests/cpp/test_cpp_api.cpp @@ -17,6 +17,7 @@ #include #include #include +#include #include #include #include @@ -281,6 +282,75 @@ static void test_scanner_blob_handling(const std::string& blob_uri) { PASS(); } +/// Byte `i` of every blob payload in the smoke fixture. +static uint8_t blob_byte(size_t i) { return static_cast(i * 7 + 3); } + +/// Check that `bytes` are the payload bytes starting at `offset`. +static void assert_blob_payload(const std::vector& bytes, size_t offset) { + for (size_t i = 0; i < bytes.size(); i++) { + assert(bytes[i] == blob_byte(offset + i)); + } +} + +static void test_take_blobs(const std::string& blob_uri) { + TEST(test_take_blobs); + + std::vector> survivors; + { + auto ds = lance::Dataset::open(blob_uri); + + // The first fragment holds an inline, a packed, a dedicated, an empty + // and a null blob, in that order. + uint64_t indices[] = {0, 1, 2, 3, 4}; + auto blobs = ds.take_blobs_by_indices(indices, 5, "blob"); + assert(blobs.size() == 5); + const uint64_t sizes[] = {8, 128, 1024, 0}; + for (size_t i = 0; i < 4; i++) { + assert(blobs[i].has_value()); + assert(blobs[i]->size() == sizes[i]); + assert_blob_payload(blobs[i]->read(), 0); + assert(blobs[i]->tell() == sizes[i]); + } + assert(!blobs[4].has_value()); + + // Cursor and positional reads on the packed blob. + lance::BlobFile& packed = *blobs[1]; + packed.seek(100); + auto tail = packed.read_up_to(64); + assert(tail.size() == 28); + assert_blob_payload(tail, 100); + assert(packed.tell() == 128); + auto window = packed.read_range(40, 16); + assert(window.size() == 16); + assert_blob_payload(window, 40); + assert(packed.tell() == 128); + + // The same column by row ID. Without stable row ids a row id is the + // row address, so the second fragment starts at 1 << 32. + uint64_t row_ids[] = {0, (uint64_t{1} << 32) | 2}; + survivors = ds.take_blobs(row_ids, 2, "blob"); + assert(survivors.size() == 2); + assert(survivors[0]->size() == 8); + assert(survivors[1]->size() == 1024); + + // A column that is not a blob column is rejected. + bool caught = false; + try { + ds.take_blobs_by_indices(indices, 5, "raw"); + } catch (const lance::Error& e) { + caught = true; + assert(e.code == LANCE_ERR_INVALID_ARGUMENT); + } + assert(caught); + } + + // Handles stay readable after the Dataset is gone. + assert_blob_payload(survivors[1]->read(), 0); + assert(survivors[1]->tell() == 1024); + + PASS(); +} + static void test_dataset_take(const std::string& uri) { TEST(test_dataset_take); @@ -1085,6 +1155,7 @@ int main(int argc, char** argv) { test_scanner_fluent(uri); test_scanner_async_stream_ownership(uri); test_scanner_blob_handling(blob_uri); + test_take_blobs(blob_uri); test_dataset_take(uri); test_dataset_take_rows(uri); test_raii_cleanup(uri);