From fab2829f1cf6ec0b5c2980174abf5e48b5c482fb Mon Sep 17 00:00:00 2001 From: liuxiaoyu Date: Thu, 10 Sep 2026 11:10:15 +0800 Subject: [PATCH 1/5] feat: expose scanner blob handling Adds LanceBlobHandling and lance_scanner_set_blob_handling(), applied in build_scanner() right after project(). Values outside the enum and calls after the scan has started are rejected. --- include/lance/lance.h | 32 +++ include/lance/lance.hpp | 8 + src/scanner.rs | 88 ++++++ tests/c_api_test.rs | 526 +++++++++++++++++++++++++++++++++- tests/compile_and_run_test.rs | 92 +++++- tests/cpp/test_c_api.c | 85 +++++- tests/cpp/test_cpp_api.cpp | 76 ++++- 7 files changed, 894 insertions(+), 13 deletions(-) diff --git a/include/lance/lance.h b/include/lance/lance.h index 8173ae5..fdf0dca 100644 --- a/include/lance/lance.h +++ b/include/lance/lance.h @@ -1059,6 +1059,38 @@ int32_t lance_scanner_set_include_deleted_rows( bool include_deleted_rows ); +/** How blob columns are materialized by a scan. Validated as an integer. */ +typedef enum { + /** + * Default: blob columns are returned as descriptor structs and every + * other binary column is returned as bytes. The descriptor layout + * depends on the storage format of the column: Blob v2 columns yield + * (kind, position, size, blob_id, blob_uri), while legacy blob columns + * (large_binary tagged `lance-encoding: blob`) yield (position, size). + */ + LANCE_BLOB_HANDLING_BLOBS_DESCRIPTIONS = 0, + /** Every blob column is materialized as bytes (LargeBinary). */ + LANCE_BLOB_HANDLING_ALL_BINARY = 1, + /** + * Requests descriptors for every binary column. On lance v11.0.0 only + * columns carrying blob metadata are affected; other binary columns keep + * their bytes, so this behaves like + * LANCE_BLOB_HANDLING_BLOBS_DESCRIPTIONS. + */ + LANCE_BLOB_HANDLING_ALL_DESCRIPTIONS = 2, +} LanceBlobHandling; + +/** + * Choose how blob columns are materialized by this scan. Default: + * LANCE_BLOB_HANDLING_BLOBS_DESCRIPTIONS. ALL_BINARY pulls the full payload + * into the batches, so keep descriptors for large values. Columns without + * blob metadata keep their bytes under every mode. + * + * Must be set before scanning starts; values outside the enum are rejected. + * @return 0 on success, -1 on error + */ +int32_t lance_scanner_set_blob_handling(LanceScanner* scanner, LanceBlobHandling handling); + /** * Restrict scan to the given fragment IDs. Must be called before iteration. * @param ids Array of fragment IDs diff --git a/include/lance/lance.hpp b/include/lance/lance.hpp index c12c0c6..64693b1 100644 --- a/include/lance/lance.hpp +++ b/include/lance/lance.hpp @@ -1290,6 +1290,14 @@ class Scanner { return *this; } + /// Choose how blob columns are materialized (default: descriptors for blob + /// columns, bytes for every other binary column). + Scanner& blob_handling(LanceBlobHandling handling) { + if (lance_scanner_set_blob_handling(handle_.get(), handling) != 0) + check_error(); + return *this; + } + /// Enable/disable row ID in output. Scanner& with_row_id(bool enable = true) { if (lance_scanner_with_row_id(handle_.get(), enable) != 0) diff --git a/src/scanner.rs b/src/scanner.rs index 4ceeb0e..8c2276d 100644 --- a/src/scanner.rs +++ b/src/scanner.rs @@ -20,6 +20,7 @@ use lance::dataset::scanner::{ }; use lance::io::exec::fts::{FlatMatchQueryExec, MatchQueryExec, PhraseQueryExec}; use lance_core::Result; +use lance_core::datatypes::BlobHandling; use lance_index::scalar::FullTextSearchQuery; use lance_index::vector::ApproxMode; use lance_io::stream::RecordBatchStream; @@ -91,6 +92,7 @@ pub struct LanceScanner { filter: Option, substrait_filter: Option>, additional_sql_filters: Vec, + blob_handling: Option, limit: Option, offset: Option, batch_size: Option, @@ -239,6 +241,7 @@ impl LanceScanner { filter: None, substrait_filter: None, additional_sql_filters: Vec::new(), + blob_handling: None, limit: None, offset: None, batch_size: None, @@ -360,6 +363,9 @@ impl LanceScanner { if let Some(cols) = &self.columns { scanner.project(cols)?; } + if let Some(handling) = &self.blob_handling { + scanner.blob_handling(handling.clone()); + } if self.limit.is_some() || self.offset.is_some() { scanner.limit(self.limit, self.offset)?; } @@ -1269,6 +1275,46 @@ unsafe fn scanner_set_use_stats_inner(scanner: *mut LanceScanner, use_stats: boo Ok(0) } +/// Set how blob columns are materialized. `handling` is the C enum +/// `LanceBlobHandling` as an integer: 0 descriptors for blob columns (the +/// default), 1 bytes for every blob column, 2 descriptors for every binary +/// column. Other values are rejected. Must be set before the scan starts. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn lance_scanner_set_blob_handling( + scanner: *mut LanceScanner, + handling: i32, +) -> i32 { + scanner_poison_check!(scanner, -1); + scanner_ffi_try!(scanner, unsafe { + scanner_set_blob_handling_inner(scanner, handling) + }) +} + +unsafe fn scanner_set_blob_handling_inner( + scanner: *mut LanceScanner, + handling: i32, +) -> Result { + if scanner.is_null() { + return Err(lance_core::Error::invalid_input_source( + "scanner is NULL".into(), + )); + } + let scanner = unsafe { &mut *scanner }; + scanner.ensure_scan_not_started("blob_handling")?; + let parsed = match handling { + 0 => BlobHandling::BlobsDescriptions, + 1 => BlobHandling::AllBinary, + 2 => BlobHandling::AllDescriptions, + other => { + return Err(lance_core::Error::invalid_input(format!( + "blob_handling must be 0 (blobs as descriptions), 1 (all binary) or 2 (all descriptions); got {other}" + ))); + } + }; + scanner.blob_handling = Some(parsed); + Ok(0) +} + /// Enable or disable row ID in scan output. Returns 0. #[unsafe(no_mangle)] pub unsafe extern "C" fn lance_scanner_with_row_id( @@ -3128,6 +3174,48 @@ mod tests { } } + #[test] + fn set_blob_handling_stores_the_matching_upstream_variant() { + // A scan cannot tell AllDescriptions from BlobsDescriptions on lance + // v11, so check the stored variant directly. + let (_tmp, uri) = create_test_dataset(); + let (dataset, scanner) = open_dataset_and_scanner(&uri); + + assert_eq!( + unsafe { &*scanner }.blob_handling, + None, + "blob handling should be unset until the setter is called" + ); + + for (handling, expected) in [ + (0, BlobHandling::BlobsDescriptions), + (1, BlobHandling::AllBinary), + (2, BlobHandling::AllDescriptions), + ] { + assert_eq!( + unsafe { lance_scanner_set_blob_handling(scanner, handling) }, + 0 + ); + assert_eq!( + unsafe { &*scanner }.blob_handling, + Some(expected), + "blob handling {handling} stored the wrong variant" + ); + } + + // A rejected value leaves the last accepted mode in place. + assert_eq!(unsafe { lance_scanner_set_blob_handling(scanner, 3) }, -1); + assert_eq!( + unsafe { &*scanner }.blob_handling, + Some(BlobHandling::AllDescriptions) + ); + + unsafe { + lance_scanner_close(scanner); + lance_dataset_close(dataset); + } + } + #[test] fn null_poll_waker_is_rejected_and_clears_out() { let (_tmp, uri) = create_test_dataset(); diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs index 3b3424b..288be7a 100644 --- a/tests/c_api_test.rs +++ b/tests/c_api_test.rs @@ -17,7 +17,10 @@ use arrow::ffi::{FFI_ArrowArray, FFI_ArrowSchema}; use arrow::ffi_stream::ArrowArrayStreamReader; use arrow::ffi_stream::FFI_ArrowArrayStream; use arrow::record_batch::RecordBatchReader; -use arrow_array::{Array, Float32Array, Int32Array, RecordBatch, StringArray, UInt64Array}; +use arrow_array::{ + Array, BinaryArray, Float32Array, Int32Array, LargeBinaryArray, RecordBatch, StringArray, + UInt32Array, UInt64Array, +}; use arrow_schema::{DataType, Field, Schema}; use lance::Dataset; use lance_c::*; @@ -12513,3 +12516,524 @@ fn test_add_columns_stream_null_dataset_consumes_stream() { assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); assert_stream_consumed(&stream, &drop_count); } + +// --------------------------------------------------------------------------- +// Scanner blob handling +// --------------------------------------------------------------------------- + +// Mirror of the C enum `LanceBlobHandling`; the FFI parameter is an int32. +const BLOB_HANDLING_BLOBS_DESCRIPTIONS: i32 = 0; +const BLOB_HANDLING_ALL_BINARY: i32 = 1; +const BLOB_HANDLING_ALL_DESCRIPTIONS: i32 = 2; + +/// Sub-fields of a Blob v2 description struct, in schema order. Legacy blob +/// columns use a two-field layout instead, which this fixture does not write. +const BLOB_DESCRIPTION_FIELDS: [&str; 5] = ["kind", "position", "size", "blob_id", "blob_uri"]; + +/// Blob storage thresholds used by [`create_blob_v2_dataset`]. Given +/// explicitly so the tests do not depend on the library defaults. +const BLOB_INLINE_THRESHOLD: usize = 16; +const BLOB_DEDICATED_THRESHOLD: usize = 256; + +/// Payload sizes of the five rows written into each fragment by +/// [`create_blob_v2_dataset`]; `None` is a null blob. Against the thresholds +/// above, 8 bytes stays inline in the data file, 128 bytes goes to packed +/// blob storage, 1024 bytes gets a dedicated blob file, and the fourth row is +/// a valid but empty blob. +const BLOB_ROW_SIZES: [Option; 5] = [Some(8), Some(128), Some(1024), Some(0), None]; + +/// `id` of the first row of each fragment written by [`create_blob_v2_dataset`]. +/// The gap lets a row's id, and its payload bytes, identify its fragment. +const BLOB_FRAGMENT_BASE_IDS: [u32; 2] = [0, 100]; + +/// Deterministic blob payload: byte `i` is `(i * 7 + 3 + seed) as u8`. +/// `seed` is the fragment's base id so a row's bytes identify its fragment. +fn blob_payload(len: usize, seed: usize) -> Vec { + (0..len).map(|i| (i * 7 + 3 + seed) as u8).collect() +} + +/// One five-row batch of the blob dataset, with ids `base_id..base_id + 5` +/// and the blob rows described by [`BLOB_ROW_SIZES`]. The plain binary column +/// holds `raw-` and is null in the same row as the blob column. +fn blob_batch(schema: &Arc, base_id: u32) -> RecordBatch { + let seed = base_id as usize; + let mut blobs = lance::BlobArrayBuilder::new(BLOB_ROW_SIZES.len()); + for size in BLOB_ROW_SIZES { + match size { + Some(0) => blobs.push_empty().unwrap(), + Some(len) => blobs.push_bytes(blob_payload(len, seed)).unwrap(), + None => blobs.push_null().unwrap(), + } + } + + let ids: Vec = (0..BLOB_ROW_SIZES.len() as u32) + .map(|row| base_id + row) + .collect(); + let raw: Vec> = ids + .iter() + .map(|id| format!("raw-{id}").into_bytes()) + .collect(); + let raw_array = BinaryArray::from_iter( + raw.iter() + .zip(BLOB_ROW_SIZES) + .map(|(value, size)| size.map(|_| value.as_slice())), + ); + + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(UInt32Array::from(ids)), + blobs.finish().unwrap(), + Arc::new(raw_array), + ], + ) + .unwrap() +} + +/// Helper: two-fragment dataset in the v2.2 storage format holding a blob +/// column (`blob`) and a plain binary column (`raw`) next to an id column. +/// Each fragment holds the five rows of [`BLOB_ROW_SIZES`], with ids starting +/// at [`BLOB_FRAGMENT_BASE_IDS`]. +fn create_blob_v2_dataset() -> (tempfile::TempDir, String) { + let tmp = tempfile::tempdir().unwrap(); + let uri = tmp.path().join("blob_ds").to_str().unwrap().to_string(); + + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::UInt32, false), + lance::blob_field_with_options( + "blob", + true, + lance::BlobFieldOptions { + inline_size_threshold: Some(BLOB_INLINE_THRESHOLD), + dedicated_size_threshold: std::num::NonZeroUsize::new(BLOB_DEDICATED_THRESHOLD), + }, + ), + Field::new("raw", DataType::Binary, true), + ])); + + lance_c::runtime::block_on(async { + for (fragment, base_id) in BLOB_FRAGMENT_BASE_IDS.into_iter().enumerate() { + let params = lance::dataset::WriteParams { + mode: if fragment == 0 { + lance::dataset::WriteMode::Create + } else { + lance::dataset::WriteMode::Append + }, + // Blob v2 is a 2.2 storage feature. + data_storage_version: Some(lance_file::version::LanceFileVersion::V2_2), + ..Default::default() + }; + Dataset::write( + arrow::record_batch::RecordBatchIterator::new( + vec![Ok(blob_batch(&schema, base_id))], + schema.clone(), + ), + &uri, + Some(params), + ) + .await + .unwrap(); + } + }); + + (tmp, uri) +} + +/// Materialize a scanner through the C Arrow stream entry point and return the +/// stream schema together with every batch it produced. The stream's `release` +/// callback runs exactly once, when the reader is dropped. +fn scan_stream(scanner: *mut LanceScanner) -> (Schema, Vec) { + let mut ffi_stream = FFI_ArrowArrayStream::empty(); + assert_eq!( + unsafe { lance_scanner_to_arrow_stream(scanner, &mut ffi_stream) }, + 0, + "to_arrow_stream should succeed" + ); + let reader = unsafe { ArrowArrayStreamReader::from_raw(&mut ffi_stream) }.unwrap(); + let schema = reader.schema().as_ref().clone(); + let batches: Vec = reader.map(|batch| batch.unwrap()).collect(); + (schema, batches) +} + +/// Collect `(id, blob bytes)` pairs from batches whose blob column was +/// materialized as bytes, sorted by id. +fn collect_blob_bytes(batches: &[RecordBatch]) -> Vec<(u32, Option>)> { + let mut rows = Vec::new(); + for batch in batches { + let ids = batch + .column_by_name("id") + .expect("id column") + .as_any() + .downcast_ref::() + .expect("id is UInt32"); + let blobs = batch + .column_by_name("blob") + .expect("blob column") + .as_any() + .downcast_ref::() + .expect("blob is LargeBinary"); + for row in 0..batch.num_rows() { + let value = (!blobs.is_null(row)).then(|| blobs.value(row).to_vec()); + rows.push((ids.value(row), value)); + } + } + rows.sort_by_key(|(id, _)| *id); + rows +} + +/// Collect `(id, raw bytes)` pairs from the plain binary column, sorted by id. +/// That column keeps its bytes under every blob handling mode. +fn collect_raw_bytes(batches: &[RecordBatch]) -> Vec<(u32, Option>)> { + let mut rows = Vec::new(); + for batch in batches { + let ids = batch + .column_by_name("id") + .expect("id column") + .as_any() + .downcast_ref::() + .expect("id is UInt32"); + let raw = batch + .column_by_name("raw") + .expect("raw column") + .as_any() + .downcast_ref::() + .expect("raw is Binary"); + for row in 0..batch.num_rows() { + let value = (!raw.is_null(row)).then(|| raw.value(row).to_vec()); + rows.push((ids.value(row), value)); + } + } + rows.sort_by_key(|(id, _)| *id); + rows +} + +/// Assert that the plain binary column of the fragment based at `base_id` +/// round-tripped: `raw-` bytes, and null in the last row. +fn assert_raw_bytes_of_fragment(rows: &[(u32, Option>)], base_id: u32) { + let row = |id: u32| -> &Option> { + &rows + .iter() + .find(|(row_id, _)| *row_id == id) + .unwrap_or_else(|| panic!("row {id} missing from scan output")) + .1 + }; + + for offset in 0..4 { + let id = base_id + offset; + assert_eq!( + row(id).as_deref(), + Some(format!("raw-{id}").as_bytes()), + "plain binary payload of row {id} must round-trip byte for byte" + ); + } + assert_eq!( + row(base_id + 4), + &None, + "null plain binary value must stay null" + ); +} + +/// Assert that the five rows written for `base_id` round-tripped byte for byte. +fn assert_blob_bytes_of_fragment(rows: &[(u32, Option>)], base_id: u32) { + let row = |id: u32| -> &Option> { + &rows + .iter() + .find(|(row_id, _)| *row_id == id) + .unwrap_or_else(|| panic!("row {id} missing from scan output")) + .1 + }; + let seed = base_id as usize; + + assert_eq!( + row(base_id).as_deref(), + Some(blob_payload(8, seed).as_slice()), + "inline blob (8 bytes) must round-trip byte for byte" + ); + assert_eq!( + row(base_id + 1).as_deref(), + Some(blob_payload(128, seed).as_slice()), + "packed blob (128 bytes) must round-trip byte for byte" + ); + assert_eq!( + row(base_id + 2).as_deref(), + Some(blob_payload(1024, seed).as_slice()), + "dedicated blob (1024 bytes) must round-trip byte for byte" + ); + assert_eq!( + row(base_id + 3).as_deref(), + Some([].as_slice()), + "empty blob must be a zero-length, non-null value" + ); + assert_eq!(row(base_id + 4), &None, "null blob must stay null"); +} + +/// Assert that the named field is a blob description struct. +fn assert_blob_description_field(schema: &Schema, name: &str) { + let field = schema.field_with_name(name).expect("field exists"); + match field.data_type() { + DataType::Struct(children) => { + let names: Vec<&str> = children.iter().map(|c| c.name().as_str()).collect(); + assert_eq!( + names, BLOB_DESCRIPTION_FIELDS, + "{name} should be a blob description struct" + ); + } + other => panic!("{name} should be a blob description struct, got {other:?}"), + } +} + +#[test] +fn test_scanner_blob_handling_all_binary_materializes_bytes() { + let (_tmp, uri) = create_blob_v2_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) }; + assert!(!scanner.is_null()); + assert_eq!( + unsafe { lance_scanner_set_blob_handling(scanner, BLOB_HANDLING_ALL_BINARY) }, + 0 + ); + + let (schema, batches) = scan_stream(scanner); + let blob_field = schema.field_with_name("blob").expect("blob column"); + assert_eq!( + *blob_field.data_type(), + DataType::LargeBinary, + "ALL_BINARY should materialize the blob column as bytes" + ); + + // Measured against lance v11.0.0: neither of the two keys that mark a + // column as a blob survives materialization, so a C consumer cannot tell + // a materialized blob from a plain binary column by metadata alone. The + // field only keeps the two threshold keys echoed from `BlobFieldOptions`. + let metadata = blob_field.metadata(); + assert!( + !metadata.contains_key("lance-encoding:blob"), + "the blob marker should not survive materialization: {metadata:?}" + ); + assert!( + !metadata.contains_key("ARROW:extension:name"), + "the blob v2 extension name should not survive materialization: {metadata:?}" + ); + + let rows = collect_blob_bytes(&batches); + assert_eq!(rows.len(), 10, "both fragments should be scanned"); + assert_blob_bytes_of_fragment(&rows, 0); + assert_blob_bytes_of_fragment(&rows, 100); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_scanner_blob_handling_defaults_to_descriptions() { + let (_tmp, uri) = create_blob_v2_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + // Without the setter, and with an explicit BLOBS_DESCRIPTIONS, the blob + // column is a description struct while plain binary columns stay bytes. + for handling in [None, Some(BLOB_HANDLING_BLOBS_DESCRIPTIONS)] { + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) }; + assert!(!scanner.is_null()); + if let Some(handling) = handling { + assert_eq!( + unsafe { lance_scanner_set_blob_handling(scanner, handling) }, + 0 + ); + } + + let (schema, batches) = scan_stream(scanner); + assert_blob_description_field(&schema, "blob"); + assert_eq!( + *schema + .field_with_name("raw") + .expect("raw column") + .data_type(), + DataType::Binary, + "a plain binary column stays bytes under {handling:?}" + ); + assert_eq!( + batches.iter().map(|b| b.num_rows()).sum::(), + 10, + "both fragments should be scanned under {handling:?}" + ); + + let raw_rows = collect_raw_bytes(&batches); + assert_raw_bytes_of_fragment(&raw_rows, 0); + assert_raw_bytes_of_fragment(&raw_rows, 100); + + unsafe { lance_scanner_close(scanner) }; + } + + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_scanner_blob_handling_all_descriptions() { + let (_tmp, uri) = create_blob_v2_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) }; + assert!(!scanner.is_null()); + assert_eq!( + unsafe { lance_scanner_set_blob_handling(scanner, BLOB_HANDLING_ALL_DESCRIPTIONS) }, + 0 + ); + + let (schema, batches) = scan_stream(scanner); + assert_blob_description_field(&schema, "blob"); + // Measured against lance v11.0.0: a column that is not marked as a blob + // keeps its bytes under ALL_DESCRIPTIONS. `BlobHandling::should_unload` + // does select every binary-like field, but the rewrite it triggers, + // `Field::unloaded_mut`, only replaces fields for which `Field::is_blob` + // holds, i.e. fields carrying blob metadata. So on this version + // ALL_DESCRIPTIONS and BLOBS_DESCRIPTIONS agree on the output schema. + assert_eq!( + *schema + .field_with_name("raw") + .expect("raw column") + .data_type(), + DataType::Binary, + "a column without blob metadata is not turned into a description" + ); + assert_eq!( + batches.iter().map(|b| b.num_rows()).sum::(), + 10, + "both fragments should be scanned" + ); + + let raw_rows = collect_raw_bytes(&batches); + assert_raw_bytes_of_fragment(&raw_rows, 0); + assert_raw_bytes_of_fragment(&raw_rows, 100); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_scanner_blob_handling_rejected_after_scan_started() { + let (_tmp, uri) = create_blob_v2_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) }; + assert!(!scanner.is_null()); + + let mut ffi_stream = FFI_ArrowArrayStream::empty(); + assert_eq!( + unsafe { lance_scanner_to_arrow_stream(scanner, &mut ffi_stream) }, + 0 + ); + // Drop the reader so the stream's release callback runs exactly once. + drop(unsafe { ArrowArrayStreamReader::from_raw(&mut ffi_stream) }.unwrap()); + + assert_eq!( + unsafe { lance_scanner_set_blob_handling(scanner, BLOB_HANDLING_ALL_BINARY) }, + -1, + "blob handling must not change once the scan has started" + ); + let message = take_last_error_message(); + assert!( + message.contains("blob_handling must be set before the scan starts"), + "unexpected error: {message}" + ); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_scanner_blob_handling_rejects_invalid_values() { + let (_tmp, uri) = create_blob_v2_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) }; + assert!(!scanner.is_null()); + + for invalid in [3, -1] { + assert_eq!( + unsafe { lance_scanner_set_blob_handling(scanner, invalid) }, + -1, + "blob_handling {invalid} should be rejected" + ); + assert_eq!(lance_last_error_code(), LanceErrorCode::InvalidArgument); + let message = take_last_error_message(); + assert!( + message.contains(&format!("got {invalid}")), + "error for {invalid} should name the rejected value: {message}" + ); + } + + assert_eq!( + unsafe { lance_scanner_set_blob_handling(ptr::null_mut(), BLOB_HANDLING_ALL_BINARY) }, + -1, + "NULL scanner should be rejected" + ); + + // A rejected value leaves the default handling in place. + let (schema, _batches) = scan_stream(scanner); + assert_blob_description_field(&schema, "blob"); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} + +#[test] +fn test_scanner_blob_handling_all_binary_with_fragment_ids() { + let (_tmp, uri) = create_blob_v2_dataset(); + let c_uri = c_str(&uri); + let ds = unsafe { lance_dataset_open(c_uri.as_ptr(), ptr::null(), 0) }; + assert!(!ds.is_null()); + assert_eq!(unsafe { lance_dataset_fragment_count(ds) }, 2); + + let mut fragment_ids = vec![0u64; 2]; + assert_eq!( + unsafe { lance_dataset_fragment_ids(ds, fragment_ids.as_mut_ptr()) }, + 0 + ); + + let scanner = unsafe { lance_scanner_new(ds, ptr::null(), ptr::null()) }; + assert!(!scanner.is_null()); + assert_eq!( + unsafe { lance_scanner_set_fragment_ids(scanner, fragment_ids[1..].as_ptr(), 1) }, + 0 + ); + assert_eq!( + unsafe { lance_scanner_set_blob_handling(scanner, BLOB_HANDLING_ALL_BINARY) }, + 0 + ); + + let (schema, batches) = scan_stream(scanner); + assert_eq!( + *schema + .field_with_name("blob") + .expect("blob column") + .data_type(), + DataType::LargeBinary + ); + + let rows = collect_blob_bytes(&batches); + assert_eq!( + rows.len(), + 5, + "only the selected fragment should be scanned" + ); + assert!( + rows.iter().all(|(id, _)| (100..105).contains(id)), + "unexpected rows from the unselected fragment: {:?}", + rows.iter().map(|(id, _)| *id).collect::>() + ); + assert_blob_bytes_of_fragment(&rows, 100); + + unsafe { lance_scanner_close(scanner) }; + unsafe { lance_dataset_close(ds) }; +} diff --git a/tests/compile_and_run_test.rs b/tests/compile_and_run_test.rs index b419ac9..1bef8c8 100644 --- a/tests/compile_and_run_test.rs +++ b/tests/compile_and_run_test.rs @@ -16,9 +16,14 @@ use std::path::{Path, PathBuf}; use std::process::Command; use std::sync::Arc; -use arrow_array::{FixedSizeListArray, Float32Array, Int32Array, RecordBatch, StringArray}; +use arrow_array::{ + BinaryArray, FixedSizeListArray, Float32Array, Int32Array, RecordBatch, StringArray, + UInt32Array, +}; use arrow_schema::{DataType, Field, Schema}; use lance::Dataset; +use lance::dataset::{WriteMode, WriteParams}; +use lance_file::version::LanceFileVersion; /// Build the lance-c cdylib and return the path to the shared library and include dir. fn build_lance_c() -> (PathBuf, PathBuf) { @@ -123,6 +128,77 @@ fn create_test_dataset_on_disk() -> (tempfile::TempDir, String) { (tmp, uri) } +/// Create a two-fragment Blob v2 dataset on disk and return (TempDir, path_string). +/// +/// Each fragment holds five rows next to an `id` column and a plain `raw` +/// binary column: blobs of 8, 128 and 1024 bytes, which the 16 / 256 byte +/// thresholds route to inline, packed and dedicated storage, then an empty +/// blob and a null. Byte `i` of every payload is `(i * 7 + 3) as u8`, the +/// formula the C and C++ programs check the bytes they read against. +fn create_blob_dataset_on_disk() -> (tempfile::TempDir, String) { + let tmp = tempfile::tempdir().unwrap(); + let uri = tmp.path().join("blob_ds").to_str().unwrap().to_string(); + + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::UInt32, false), + lance::blob_field_with_options( + "blob", + true, + lance::BlobFieldOptions { + inline_size_threshold: Some(16), + dedicated_size_threshold: std::num::NonZeroUsize::new(256), + }, + ), + Field::new("raw", DataType::Binary, true), + ])); + + let make_batch = |first_id: u32| { + let mut blobs = lance::BlobArrayBuilder::new(5); + for len in [8usize, 128, 1024] { + let payload: Vec = (0..len).map(|i| (i * 7 + 3) as u8).collect(); + blobs.push_bytes(payload).unwrap(); + } + blobs.push_empty().unwrap(); + blobs.push_null().unwrap(); + + let ids: Vec = (first_id..first_id + 5).collect(); + // The plain binary column is null in the same row as the blob column. + let raw = BinaryArray::from_iter((0..5).map(|row| (row < 4).then_some(&b"raw"[..]))); + RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(UInt32Array::from(ids)), + blobs.finish().unwrap(), + Arc::new(raw), + ], + ) + .unwrap() + }; + + lance_c::runtime::block_on(async { + for (first_id, mode) in [(0u32, WriteMode::Create), (100u32, WriteMode::Append)] { + let params = WriteParams { + mode, + // Blob v2 is a 2.2 storage feature. + data_storage_version: Some(LanceFileVersion::V2_2), + ..Default::default() + }; + Dataset::write( + arrow::record_batch::RecordBatchIterator::new( + vec![Ok(make_batch(first_id))], + schema.clone(), + ), + &uri, + Some(params), + ) + .await + .unwrap(); + } + }); + + (tmp, uri) +} + /// Compile a C source file, linking against lance-c. fn compile_c_test(source: &Path, output: &Path, include_dir: &Path, lib_path: &Path) -> bool { let lib_dir = lib_path.parent().unwrap(); @@ -174,12 +250,14 @@ fn compile_cpp_test(source: &Path, output: &Path, include_dir: &Path, lib_path: .success() } -/// Run a compiled test binary with the source dataset URI and a destination URI -/// for the write test. The destination path must not pre-exist. -fn run_test_binary(binary: &Path, dataset_uri: &str, write_uri: &str) { +/// Run a compiled test binary with the source dataset URI, a destination URI +/// for the write test and the URI of a Blob v2 dataset. The destination path +/// must not pre-exist. +fn run_test_binary(binary: &Path, dataset_uri: &str, write_uri: &str, blob_uri: &str) { let output = Command::new(binary) .arg(dataset_uri) .arg(write_uri) + .arg(blob_uri) .output() .unwrap_or_else(|e| panic!("Failed to run {}: {e}", binary.display())); @@ -207,6 +285,7 @@ fn test_c_compilation_and_execution() { let (lib_path, include_dir) = build_lance_c(); let (tmp, dataset_uri) = create_test_dataset_on_disk(); let write_uri = tmp.path().join("c_write_ds").to_str().unwrap().to_string(); + let (_blob_tmp, blob_uri) = create_blob_dataset_on_disk(); let build_dir = tempfile::tempdir().unwrap(); let source = PathBuf::from(env!("CARGO_MANIFEST_DIR")) @@ -220,7 +299,7 @@ fn test_c_compilation_and_execution() { "C test compilation failed" ); - run_test_binary(&binary, &dataset_uri, &write_uri); + run_test_binary(&binary, &dataset_uri, &write_uri, &blob_uri); } #[test] @@ -234,6 +313,7 @@ fn test_cpp_compilation_and_execution() { .to_str() .unwrap() .to_string(); + let (_blob_tmp, blob_uri) = create_blob_dataset_on_disk(); let build_dir = tempfile::tempdir().unwrap(); let source = PathBuf::from(env!("CARGO_MANIFEST_DIR")) @@ -247,5 +327,5 @@ fn test_cpp_compilation_and_execution() { "C++ test compilation failed" ); - run_test_binary(&binary, &dataset_uri, &write_uri); + run_test_binary(&binary, &dataset_uri, &write_uri, &blob_uri); } diff --git a/tests/cpp/test_c_api.c b/tests/cpp/test_c_api.c index c49ecfa..0e524d7 100644 --- a/tests/cpp/test_c_api.c +++ b/tests/cpp/test_c_api.c @@ -8,7 +8,7 @@ * This file is compiled by the Rust integration test to verify that * lance.h is valid C and the API works end-to-end. * - * Usage: test_c_api + * Usage: test_c_api */ #include "lance/lance.h" @@ -218,6 +218,83 @@ static void test_scan_with_limit(const char *uri) { printf("OK\n"); } +/* Copy the Arrow C Data Interface format of the `blob` column of a stream's + * schema into `out`; `out` is empty when the column is missing. */ +static void blob_column_format(struct ArrowArrayStream *stream, char *out, size_t out_len) { + struct ArrowSchema schema; + memset(&schema, 0, sizeof(schema)); + int rc = stream->get_schema(stream, &schema); + ASSERT(rc == 0, "get_schema from stream failed"); + out[0] = '\0'; + for (int64_t i = 0; i < schema.n_children; i++) { + if (strcmp(schema.children[i]->name, "blob") == 0) { + snprintf(out, out_len, "%s", schema.children[i]->format); + } + } + if (schema.release) schema.release(&schema); +} + +static void test_scanner_blob_handling(const char *blob_uri) { + printf(" test_scanner_blob_handling... "); + + LanceDataset *ds = lance_dataset_open(blob_uri, NULL, 0); + ASSERT(ds != NULL, "blob dataset open failed"); + uint64_t expected_rows = lance_dataset_count_rows(ds); + CHECK_OK(); + + char format[16]; + struct ArrowArrayStream stream; + + /* By default a blob column arrives as its description struct. */ + LanceScanner *scanner = lance_scanner_new(ds, NULL, NULL); + ASSERT(scanner != NULL, "scanner creation failed"); + memset(&stream, 0, sizeof(stream)); + int32_t rc = lance_scanner_to_arrow_stream(scanner, &stream); + ASSERT(rc == 0, "to_arrow_stream failed"); + blob_column_format(&stream, format, sizeof(format)); + ASSERT(strcmp(format, "+s") == 0, "default blob column should be a struct"); + if (stream.release) stream.release(&stream); + lance_scanner_close(scanner); + + /* ALL_BINARY materializes the bytes as LargeBinary and keeps every row. */ + scanner = lance_scanner_new(ds, NULL, NULL); + ASSERT(scanner != NULL, "scanner creation failed"); + rc = lance_scanner_set_blob_handling(scanner, LANCE_BLOB_HANDLING_ALL_BINARY); + ASSERT(rc == 0, "set_blob_handling failed"); + memset(&stream, 0, sizeof(stream)); + rc = lance_scanner_to_arrow_stream(scanner, &stream); + ASSERT(rc == 0, "to_arrow_stream failed"); + blob_column_format(&stream, format, sizeof(format)); + ASSERT(strcmp(format, "Z") == 0, "ALL_BINARY blob column should be LargeBinary"); + + uint64_t total_rows = 0; + while (1) { + struct ArrowArray array; + memset(&array, 0, sizeof(array)); + rc = stream.get_next(&stream, &array); + ASSERT(rc == 0, "get_next failed"); + if (array.release == NULL) { + break; + } + total_rows += (uint64_t)array.length; + array.release(&array); + } + ASSERT(total_rows == expected_rows, "row count mismatch"); + if (stream.release) stream.release(&stream); + + /* Once the scan has started the setting is rejected. */ + rc = lance_scanner_set_blob_handling(scanner, LANCE_BLOB_HANDLING_BLOBS_DESCRIPTIONS); + ASSERT(rc == -1, "set_blob_handling after the scan started should fail"); + ASSERT(lance_last_error_code() == LANCE_ERR_INVALID_ARGUMENT, "wrong error code"); + const char *msg = lance_last_error_message(); + if (msg) lance_free_string(msg); + + printf("rows=%llu... ", (unsigned long long)total_rows); + lance_scanner_close(scanner); + lance_dataset_close(ds); + printf("OK\n"); +} + static void test_versions(const char *uri) { printf(" test_versions... "); @@ -962,19 +1039,21 @@ static void test_delete(const char *write_uri) { } int main(int argc, char **argv) { - if (argc < 3) { - fprintf(stderr, "Usage: %s \n", argv[0]); + if (argc < 4) { + fprintf(stderr, "Usage: %s \n", argv[0]); return 1; } const char *uri = argv[1]; const char *write_uri = argv[2]; + const char *blob_uri = argv[3]; printf("Running C API tests with dataset: %s\n", uri); test_open_and_metadata(uri); test_shared_session(uri); test_scan(uri); test_scan_with_limit(uri); + test_scanner_blob_handling(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 60b8fc4..c85d201 100644 --- a/tests/cpp/test_cpp_api.cpp +++ b/tests/cpp/test_cpp_api.cpp @@ -7,7 +7,7 @@ * * Tests the RAII wrappers, exception handling, and builder pattern. * - * Usage: test_cpp_api + * Usage: test_cpp_api */ #include "lance/lance.hpp" @@ -214,6 +214,74 @@ static void test_scanner_async_stream_ownership(const std::string& uri) { PASS(); } +/// Arrow C Data Interface format of the `blob` column in a stream's schema, +/// or an empty string when the column is missing. +static std::string blob_column_format(ArrowArrayStream& stream) { + ArrowSchema schema; + memset(&schema, 0, sizeof(schema)); + int rc = stream.get_schema(&stream, &schema); + assert(rc == 0); + std::string format; + for (int64_t i = 0; i < schema.n_children; i++) { + if (strcmp(schema.children[i]->name, "blob") == 0) { + format = schema.children[i]->format; + } + } + if (schema.release) schema.release(&schema); + return format; +} + +static void test_scanner_blob_handling(const std::string& blob_uri) { + TEST(test_scanner_blob_handling); + + auto ds = lance::Dataset::open(blob_uri); + + // By default a blob column arrives as its description struct ("+s"). + { + auto scanner = ds.scan(); + ArrowArrayStream stream; + memset(&stream, 0, sizeof(stream)); + scanner.to_arrow_stream(&stream); + assert(blob_column_format(stream) == "+s"); + if (stream.release) stream.release(&stream); + } + + // ALL_BINARY materializes it as LargeBinary ("Z") and still yields every + // row of both fragments. + auto scanner = ds.scan(); + scanner.blob_handling(LANCE_BLOB_HANDLING_ALL_BINARY); + ArrowArrayStream stream; + memset(&stream, 0, sizeof(stream)); + scanner.to_arrow_stream(&stream); + assert(blob_column_format(stream) == "Z"); + + uint64_t total = 0; + while (true) { + ArrowArray arr; + memset(&arr, 0, sizeof(arr)); + int rc = stream.get_next(&stream, &arr); + assert(rc == 0); + if (!arr.release) break; + total += (uint64_t)arr.length; + arr.release(&arr); + } + assert(total == ds.count_rows()); + if (stream.release) stream.release(&stream); + + // Once the scan has started the setting is rejected. + bool caught = false; + try { + scanner.blob_handling(LANCE_BLOB_HANDLING_BLOBS_DESCRIPTIONS); + } catch (const lance::Error& e) { + caught = true; + assert(e.code == LANCE_ERR_INVALID_ARGUMENT); + } + assert(caught); + + printf("rows=%llu... ", (unsigned long long)total); + PASS(); +} + static void test_dataset_take(const std::string& uri) { TEST(test_dataset_take); @@ -926,13 +994,14 @@ static void test_delete_rows(const std::string& dst_uri) { } int main(int argc, char** argv) { - if (argc < 3) { - fprintf(stderr, "Usage: %s \n", argv[0]); + if (argc < 4) { + fprintf(stderr, "Usage: %s \n", argv[0]); return 1; } std::string uri(argv[1]); std::string write_uri(argv[2]); + std::string blob_uri(argv[3]); printf("Running C++ API tests with dataset: %s\n", uri.c_str()); test_dataset_open(uri); @@ -940,6 +1009,7 @@ int main(int argc, char** argv) { test_dataset_schema(uri); test_scanner_fluent(uri); test_scanner_async_stream_ownership(uri); + test_scanner_blob_handling(blob_uri); test_dataset_take(uri); test_dataset_take_rows(uri); test_raii_cleanup(uri); From c1ec76d6be10f806880c2bf61feecc99637252cd Mon Sep 17 00:00:00 2001 From: liuxiaoyu Date: Thu, 10 Sep 2026 11:10:15 +0800 Subject: [PATCH 2/5] test: cover scanner blob handling Two-fragment Blob v2 fixture (inline, packed, dedicated, empty and null rows) and six tests. The C and C++ smoke programs get a blob dataset as a third argument and check the setter too. --- tests/c_api_test.rs | 52 ++++++++++++----------------------- tests/compile_and_run_test.rs | 9 +++--- tests/cpp/test_cpp_api.cpp | 3 +- 3 files changed, 22 insertions(+), 42 deletions(-) diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs index 288be7a..fa9638f 100644 --- a/tests/c_api_test.rs +++ b/tests/c_api_test.rs @@ -12526,35 +12526,28 @@ const BLOB_HANDLING_BLOBS_DESCRIPTIONS: i32 = 0; const BLOB_HANDLING_ALL_BINARY: i32 = 1; const BLOB_HANDLING_ALL_DESCRIPTIONS: i32 = 2; -/// Sub-fields of a Blob v2 description struct, in schema order. Legacy blob -/// columns use a two-field layout instead, which this fixture does not write. +/// Sub-fields of a Blob v2 description struct, in schema order. const BLOB_DESCRIPTION_FIELDS: [&str; 5] = ["kind", "position", "size", "blob_id", "blob_uri"]; -/// Blob storage thresholds used by [`create_blob_v2_dataset`]. Given -/// explicitly so the tests do not depend on the library defaults. +/// Blob storage thresholds used by [`create_blob_v2_dataset`]. const BLOB_INLINE_THRESHOLD: usize = 16; const BLOB_DEDICATED_THRESHOLD: usize = 256; -/// Payload sizes of the five rows written into each fragment by -/// [`create_blob_v2_dataset`]; `None` is a null blob. Against the thresholds -/// above, 8 bytes stays inline in the data file, 128 bytes goes to packed -/// blob storage, 1024 bytes gets a dedicated blob file, and the fourth row is -/// a valid but empty blob. +/// Blob sizes of the five rows in each fragment: inline, packed and dedicated +/// against the thresholds above, then an empty blob and a null. const BLOB_ROW_SIZES: [Option; 5] = [Some(8), Some(128), Some(1024), Some(0), None]; -/// `id` of the first row of each fragment written by [`create_blob_v2_dataset`]. -/// The gap lets a row's id, and its payload bytes, identify its fragment. +/// First `id` of each fragment; also seeds its payloads. const BLOB_FRAGMENT_BASE_IDS: [u32; 2] = [0, 100]; -/// Deterministic blob payload: byte `i` is `(i * 7 + 3 + seed) as u8`. -/// `seed` is the fragment's base id so a row's bytes identify its fragment. +/// Blob payload: byte `i` is `(i * 7 + 3 + seed) as u8`. fn blob_payload(len: usize, seed: usize) -> Vec { (0..len).map(|i| (i * 7 + 3 + seed) as u8).collect() } -/// One five-row batch of the blob dataset, with ids `base_id..base_id + 5` -/// and the blob rows described by [`BLOB_ROW_SIZES`]. The plain binary column -/// holds `raw-` and is null in the same row as the blob column. +/// One fragment's batch: ids `base_id..base_id + 5`, blobs per +/// [`BLOB_ROW_SIZES`], `raw-` in the plain binary column (null where the +/// blob is null). fn blob_batch(schema: &Arc, base_id: u32) -> RecordBatch { let seed = base_id as usize; let mut blobs = lance::BlobArrayBuilder::new(BLOB_ROW_SIZES.len()); @@ -12590,10 +12583,8 @@ fn blob_batch(schema: &Arc, base_id: u32) -> RecordBatch { .unwrap() } -/// Helper: two-fragment dataset in the v2.2 storage format holding a blob -/// column (`blob`) and a plain binary column (`raw`) next to an id column. -/// Each fragment holds the five rows of [`BLOB_ROW_SIZES`], with ids starting -/// at [`BLOB_FRAGMENT_BASE_IDS`]. +/// 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) { let tmp = tempfile::tempdir().unwrap(); let uri = tmp.path().join("blob_ds").to_str().unwrap().to_string(); @@ -12639,9 +12630,7 @@ fn create_blob_v2_dataset() -> (tempfile::TempDir, String) { (tmp, uri) } -/// Materialize a scanner through the C Arrow stream entry point and return the -/// stream schema together with every batch it produced. The stream's `release` -/// callback runs exactly once, when the reader is dropped. +/// Run the scanner through the C Arrow stream; return its schema and batches. fn scan_stream(scanner: *mut LanceScanner) -> (Schema, Vec) { let mut ffi_stream = FFI_ArrowArrayStream::empty(); assert_eq!( @@ -12682,7 +12671,6 @@ fn collect_blob_bytes(batches: &[RecordBatch]) -> Vec<(u32, Option>)> { } /// Collect `(id, raw bytes)` pairs from the plain binary column, sorted by id. -/// That column keeps its bytes under every blob handling mode. fn collect_raw_bytes(batches: &[RecordBatch]) -> Vec<(u32, Option>)> { let mut rows = Vec::new(); for batch in batches { @@ -12804,10 +12792,8 @@ fn test_scanner_blob_handling_all_binary_materializes_bytes() { "ALL_BINARY should materialize the blob column as bytes" ); - // Measured against lance v11.0.0: neither of the two keys that mark a - // column as a blob survives materialization, so a C consumer cannot tell - // a materialized blob from a plain binary column by metadata alone. The - // field only keeps the two threshold keys echoed from `BlobFieldOptions`. + // Neither blob marker survives materialization (lance v11), so a C caller + // cannot tell a materialized blob from a plain binary column by metadata. let metadata = blob_field.metadata(); assert!( !metadata.contains_key("lance-encoding:blob"), @@ -12888,12 +12874,8 @@ fn test_scanner_blob_handling_all_descriptions() { let (schema, batches) = scan_stream(scanner); assert_blob_description_field(&schema, "blob"); - // Measured against lance v11.0.0: a column that is not marked as a blob - // keeps its bytes under ALL_DESCRIPTIONS. `BlobHandling::should_unload` - // does select every binary-like field, but the rewrite it triggers, - // `Field::unloaded_mut`, only replaces fields for which `Field::is_blob` - // holds, i.e. fields carrying blob metadata. So on this version - // ALL_DESCRIPTIONS and BLOBS_DESCRIPTIONS agree on the output schema. + // On lance v11 ALL_DESCRIPTIONS only rewrites fields with blob metadata + // (`Field::unloaded_mut` is gated on `is_blob`), so `raw` keeps its bytes. assert_eq!( *schema .field_with_name("raw") @@ -12931,7 +12913,7 @@ fn test_scanner_blob_handling_rejected_after_scan_started() { unsafe { lance_scanner_to_arrow_stream(scanner, &mut ffi_stream) }, 0 ); - // Drop the reader so the stream's release callback runs exactly once. + // Release the stream; the scan has started either way. drop(unsafe { ArrowArrayStreamReader::from_raw(&mut ffi_stream) }.unwrap()); assert_eq!( diff --git a/tests/compile_and_run_test.rs b/tests/compile_and_run_test.rs index 1bef8c8..5eb5dc9 100644 --- a/tests/compile_and_run_test.rs +++ b/tests/compile_and_run_test.rs @@ -130,11 +130,10 @@ fn create_test_dataset_on_disk() -> (tempfile::TempDir, String) { /// Create a two-fragment Blob v2 dataset on disk and return (TempDir, path_string). /// -/// Each fragment holds five rows next to an `id` column and a plain `raw` -/// binary column: blobs of 8, 128 and 1024 bytes, which the 16 / 256 byte -/// thresholds route to inline, packed and dedicated storage, then an empty -/// blob and a null. Byte `i` of every payload is `(i * 7 + 3) as u8`, the -/// formula the C and C++ programs check the bytes they read against. +/// Each fragment has five rows: blobs of 8, 128 and 1024 bytes (inline, packed +/// and dedicated under the 16 / 256 thresholds), an empty blob and a null, +/// next to `id` and a plain `raw` binary column. Payload byte `i` is +/// `(i * 7 + 3) as u8`. fn create_blob_dataset_on_disk() -> (tempfile::TempDir, String) { let tmp = tempfile::tempdir().unwrap(); let uri = tmp.path().join("blob_ds").to_str().unwrap().to_string(); diff --git a/tests/cpp/test_cpp_api.cpp b/tests/cpp/test_cpp_api.cpp index c85d201..fd4acd4 100644 --- a/tests/cpp/test_cpp_api.cpp +++ b/tests/cpp/test_cpp_api.cpp @@ -246,8 +246,7 @@ static void test_scanner_blob_handling(const std::string& blob_uri) { if (stream.release) stream.release(&stream); } - // ALL_BINARY materializes it as LargeBinary ("Z") and still yields every - // row of both fragments. + // ALL_BINARY: LargeBinary ("Z"), and every row is still returned. auto scanner = ds.scan(); scanner.blob_handling(LANCE_BLOB_HANDLING_ALL_BINARY); ArrowArrayStream stream; From 6513805f11c37959f3919192cd490c7baa2c1ad0 Mon Sep 17 00:00:00 2001 From: liuxiaoyu Date: Thu, 10 Sep 2026 11:11:15 +0800 Subject: [PATCH 3/5] feat: add blob take and BlobFile read APIs lance_dataset_take_blobs / _by_indices fill a caller array of LanceBlobFile* (NULL for null values, untouched on error). The handle has size, read, read_up_to, read_range, seek, tell and close, and wraps upstream BlobFile only, so it outlives the dataset. --- include/lance/lance.h | 175 +++++++++++++++++++++ include/lance/lance.hpp | 117 ++++++++++++++ src/blob.rs | 340 ++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 2 + 4 files changed, 634 insertions(+) create mode 100644 src/blob.rs diff --git a/include/lance/lance.h b/include/lance/lance.h index fdf0dca..c2266b0 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 64693b1..ba64048 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,19 @@ 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. + 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 +846,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..b1cd3ed --- /dev/null +++ b/src/blob.rs @@ -0,0 +1,340 @@ +// 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() { + let dst = unsafe { std::slice::from_raw_parts_mut(dst, dst_len) }; + dst[..bytes.len()].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() { + let dst = unsafe { std::slice::from_raw_parts_mut(dst, len) }; + dst[..bytes.len()].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 8b212f5..c9e417b 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; @@ -50,6 +51,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::*; From d0fd2c0871861bebe1f353e9f974e6569c47a468 Mon Sep 17 00:00:00 2001 From: liuxiaoyu Date: Thu, 10 Sep 2026 11:11:15 +0800 Subject: [PATCH 4/5] test: cover blob take and reads Reuses the fixture from the scanner blob handling PR, with a stable row id switch. --- tests/c_api_test.rs | 999 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 992 insertions(+), 7 deletions(-) diff --git a/tests/c_api_test.rs b/tests/c_api_test.rs index fa9638f..5febe52 100644 --- a/tests/c_api_test.rs +++ b/tests/c_api_test.rs @@ -12585,7 +12585,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(); @@ -12612,6 +12615,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( @@ -12772,7 +12776,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()); @@ -12815,7 +12819,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()); @@ -12860,7 +12864,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()); @@ -12900,7 +12904,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()); @@ -12933,7 +12937,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()); @@ -12971,7 +12975,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()); @@ -13019,3 +13023,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) }; +} From 96656bc3647fb32b7cbf429b346086fbc13fbee2 Mon Sep 17 00:00:00 2001 From: liuxiaoyu Date: Thu, 10 Sep 2026 11:11:15 +0800 Subject: [PATCH 5/5] test: exercise blob take from C and C++ --- tests/cpp/test_c_api.c | 89 ++++++++++++++++++++++++++++++++++++++ tests/cpp/test_cpp_api.cpp | 71 ++++++++++++++++++++++++++++++ 2 files changed, 160 insertions(+) diff --git a/tests/cpp/test_c_api.c b/tests/cpp/test_c_api.c index 0e524d7..156c960 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... "); @@ -1054,6 +1142,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 fd4acd4..abe60ab 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); @@ -1009,6 +1079,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);