From d967e4c106663c26babca41e989207c7dd07d5c9 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Mon, 27 Jul 2026 16:23:31 +0800 Subject: [PATCH] refactor: use aligned Lance blob results --- Cargo.lock | 97 +++++----- Cargo.toml | 28 +-- java/pom.xml | 2 +- python/python/lancedb/table.py | 3 + python/python/tests/test_blob.py | 17 +- rust/lancedb/src/blob.rs | 241 ++++--------------------- rust/lancedb/src/table.rs | 9 +- rust/lancedb/tests/blob_integration.rs | 33 +++- 8 files changed, 155 insertions(+), 275 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 57e5b8d84..e18f3d364 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3421,8 +3421,8 @@ checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" [[package]] name = "fsst" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-array", "rand 0.9.5", @@ -4777,8 +4777,8 @@ checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a" [[package]] name = "lance" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arc-swap", "arrow", @@ -4852,8 +4852,8 @@ dependencies = [ [[package]] name = "lance-arrow" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-array", "arrow-buffer", @@ -4875,7 +4875,7 @@ dependencies = [ [[package]] name = "lance-arrow-scalar" version = "58.0.0" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-array", "arrow-buffer", @@ -4889,7 +4889,7 @@ dependencies = [ [[package]] name = "lance-arrow-stats" version = "58.0.0" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-array", "arrow-schema", @@ -4898,8 +4898,8 @@ dependencies = [ [[package]] name = "lance-bitpacking" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrayref", "crunchy", @@ -4909,8 +4909,8 @@ dependencies = [ [[package]] name = "lance-core" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-array", "arrow-buffer", @@ -4933,6 +4933,7 @@ dependencies = [ "object_store", "pin-project", "prost", + "quick_cache", "rand 0.9.5", "roaring", "serde_json", @@ -4948,8 +4949,8 @@ dependencies = [ [[package]] name = "lance-datafusion" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow", "arrow-array", @@ -4979,8 +4980,8 @@ dependencies = [ [[package]] name = "lance-datagen" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow", "arrow-array", @@ -4997,8 +4998,8 @@ dependencies = [ [[package]] name = "lance-derive" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "proc-macro2", "quote", @@ -5007,8 +5008,8 @@ dependencies = [ [[package]] name = "lance-encoding" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-arith", "arrow-array", @@ -5043,8 +5044,8 @@ dependencies = [ [[package]] name = "lance-file" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-arith", "arrow-array", @@ -5074,8 +5075,8 @@ dependencies = [ [[package]] name = "lance-index" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arc-swap", "arrow", @@ -5142,8 +5143,8 @@ dependencies = [ [[package]] name = "lance-index-core" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-array", "arrow-schema", @@ -5165,8 +5166,8 @@ dependencies = [ [[package]] name = "lance-io" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow", "arrow-arith", @@ -5209,8 +5210,8 @@ dependencies = [ [[package]] name = "lance-linalg" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-array", "arrow-buffer", @@ -5226,8 +5227,8 @@ dependencies = [ [[package]] name = "lance-namespace" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow", "async-trait", @@ -5239,8 +5240,8 @@ dependencies = [ [[package]] name = "lance-namespace-impls" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow", "arrow-ipc", @@ -5294,8 +5295,8 @@ dependencies = [ [[package]] name = "lance-select" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-array", "arrow-buffer", @@ -5310,8 +5311,8 @@ dependencies = [ [[package]] name = "lance-table" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow", "arrow-array", @@ -5350,8 +5351,8 @@ dependencies = [ [[package]] name = "lance-testing" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "arrow-array", "arrow-schema", @@ -5364,8 +5365,8 @@ dependencies = [ [[package]] name = "lance-tokenizer" -version = "9.1.0-beta.8" -source = "git+https://github.com/lance-format/lance.git?tag=v9.1.0-beta.8#5107a99e3f3912851c8cbb3822bd5b4cbf7fda2f" +version = "10.0.0-beta.5" +source = "git+https://github.com/lance-format/lance.git?tag=v10.0.0-beta.5#ddb8e28ca238f29628b8e1795ddccbb7bf75e5c8" dependencies = [ "icu_segmenter", "jieba-rs", @@ -7802,6 +7803,18 @@ dependencies = [ "memchr", ] +[[package]] +name = "quick_cache" +version = "0.6.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9c6658afe513a3b484e3abfdaa0d03ef3c0bbf017542c178dd55f94eb3051f9" +dependencies = [ + "ahash", + "equivalent", + "hashbrown 0.16.1", + "parking_lot", +] + [[package]] name = "quinn" version = "0.11.9" diff --git a/Cargo.toml b/Cargo.toml index 94b965d50..fee01b732 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,20 +13,20 @@ categories = ["database-implementations"] rust-version = "1.91.0" [workspace.dependencies] -lance = { "version" = "=9.1.0-beta.8", default-features = false, "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-core = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-datagen = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-file = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-io = { "version" = "=9.1.0-beta.8", default-features = false, "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-index = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-linalg = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-namespace = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-namespace-impls = { "version" = "=9.1.0-beta.8", default-features = false, "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-table = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-testing = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-datafusion = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-encoding = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } -lance-arrow = { "version" = "=9.1.0-beta.8", "tag" = "v9.1.0-beta.8", "git" = "https://github.com/lance-format/lance.git" } +lance = { "version" = "=10.0.0-beta.5", default-features = false, "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-core = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-datagen = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-file = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-io = { "version" = "=10.0.0-beta.5", default-features = false, "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-index = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-linalg = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-namespace = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-namespace-impls = { "version" = "=10.0.0-beta.5", default-features = false, "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-table = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-testing = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-datafusion = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-encoding = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } +lance-arrow = { "version" = "=10.0.0-beta.5", "tag" = "v10.0.0-beta.5", "git" = "https://github.com/lance-format/lance.git" } ahash = "0.8" # Note that this one does not include pyarrow arrow = { version = "58.0.0", optional = false } diff --git a/java/pom.xml b/java/pom.xml index 78a393c30..60ea6b33c 100644 --- a/java/pom.xml +++ b/java/pom.xml @@ -28,7 +28,7 @@ UTF-8 15.0.0 - 9.1.0-beta.8 + 10.0.0-beta.5 false 2.30.0 1.7 diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index b2c8682c1..a2c533892 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -1534,6 +1534,9 @@ class Table(ABC): ) -> pa.LargeBinaryArray: """Materialize full blob bytes for ``column`` at the given rows. + The result has the same length and order as ``row_ids``. Null blobs + produce null slots; valid empty blobs produce ``b""``. + Convenience for small payloads. For large values use :meth:`fetch_blob_files`. """ diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py index 59c80a57c..04a99ea16 100644 --- a/python/python/tests/test_blob.py +++ b/python/python/tests/test_blob.py @@ -184,18 +184,23 @@ def test_fetch_blobs_accepts_query_result(): assert {blobs[i].as_py() for i in range(len(blobs))} == {b"gamma"} -def test_fetch_blobs_null_alignment(): +def test_fetch_blobs_preserves_null_and_empty_values(): table = _blob_table( "nulls", - [{"id": 1, "image": b"present"}, {"id": 2, "image": None}], + [ + {"id": 1, "image": b"present"}, + {"id": 2, "image": None}, + {"id": 3, "image": b""}, + ], ) by_id = _row_ids_by_id(table) - request = [by_id[1], by_id[2], by_id[1]] + request = [by_id[1], by_id[2], by_id[3], by_id[1]] blobs = table.fetch_blobs("image", request) assert len(blobs) == len(request) assert blobs[0].as_py() == b"present" assert blobs[1].as_py() is None - assert blobs[2].as_py() == b"present" + assert blobs[2].as_py() == b"" + assert blobs[3].as_py() == b"present" def test_fetch_blob_ranges_aligns_repeated_ranges_and_nulls(): @@ -221,10 +226,10 @@ def test_fetch_blob_ranges_validates_requests(): table = _blob_table("range_validation", [{"id": 1, "image": b"abc"}]) row_id = _row_ids_by_id(table)[1] - with pytest.raises(ValueError, match="exceeds blob size"): + with pytest.raises(RuntimeError, match="exceeds blob size"): table.fetch_blob_ranges("image", [(row_id, 2, 2)]) - with pytest.raises(ValueError, match="offset \\+ length overflowed"): + with pytest.raises(RuntimeError, match="offset \\+ length overflowed"): table.fetch_blob_ranges("image", [(row_id, 2**64 - 1, 1)]) with pytest.raises(ValueError, match="row ids"): diff --git a/rust/lancedb/src/blob.rs b/rust/lancedb/src/blob.rs index deaa22e84..a21fe3c1b 100644 --- a/rust/lancedb/src/blob.rs +++ b/rust/lancedb/src/blob.rs @@ -11,13 +11,11 @@ use std::sync::Arc; +use arrow_array::LargeBinaryArray; use arrow_array::builder::LargeBinaryBuilder; -use arrow_array::{Array, LargeBinaryArray, RecordBatch, StructArray, UInt8Array, UInt64Array}; use arrow_schema::{DataType, Field, Schema}; -use futures::TryStreamExt; use lance::dataset::{BlobRangeRequest as LanceBlobRangeRequest, Dataset, WriteParams}; use lance_arrow::FieldExt; -use lance_core::datatypes::parse_field_path; use lance_encoding::version::LanceFileVersion; use crate::error::{Error, Result}; @@ -49,15 +47,6 @@ impl BlobRangeRequest { } } -fn map_blob_read_error(source: lance::Error) -> Error { - match source { - source @ lance::Error::InvalidInput { .. } => Error::InvalidInput { - message: source.to_string(), - }, - source => source.into(), - } -} - /// Creates an Arrow field for a Lance blob v2 column. /// /// `Struct` with the `lance.blob.v2` marker. Same layout Lance @@ -180,91 +169,24 @@ pub(crate) fn ensure_blob_v2_column( } } -/// Returns the leaf descriptor `StructArray` for `column` in a descriptor batch. -fn leaf_descriptor_struct<'a>(batch: &'a RecordBatch, column: &str) -> Result<&'a StructArray> { - let path = parse_field_path(column).map_err(|e| Error::InvalidInput { - message: format!("invalid blob column path '{column}': {e}"), - })?; - let not_struct = || Error::Runtime { - message: format!("blob column '{column}' did not read back as a descriptor struct"), - }; - let mut current = batch - .column_by_name(&path[0]) - .and_then(|c| c.as_any().downcast_ref::()) - .ok_or_else(not_struct)?; - for segment in &path[1..] { - current = current - .column_by_name(segment) - .and_then(|c| c.as_any().downcast_ref::()) - .ok_or_else(not_struct)?; +fn ensure_all_row_ids_resolved(column: &str, requested: usize, resolved: usize) -> Result<()> { + if requested == resolved { + return Ok(()); } - Ok(current) -} - -/// Null rows in `row_ids`, from a descriptor take. -/// -/// Lance `read_blobs` / `take_blobs` skip null rows (`kind == 0 && position == 0 && size == 0`). -/// TODO(lance): aligned read API would drop this pass. -async fn blob_null_mask( - dataset: &Arc, - column: &str, - row_ids: &[u64], -) -> Result> { - let projection = dataset.schema().project(&[column])?; - let descriptors = dataset.take_builder(row_ids, projection)?.execute().await?; - if descriptors.num_rows() != row_ids.len() { - return Err(Error::InvalidInput { + if resolved < requested { + Err(Error::InvalidInput { message: format!( - "blob take for column '{column}' requested {} row ids but only {} exist in the \ - table; pass row ids collected from this table", - row_ids.len(), - descriptors.num_rows() + "blob read for column '{column}' requested {requested} row ids but only {resolved} \ + exist in the table; pass row ids collected from this table" ), - }); - } - let descriptor_struct = leaf_descriptor_struct(&descriptors, column)?; - let child = |name: &str| { - descriptor_struct - .column_by_name(name) - .ok_or_else(|| Error::Runtime { - message: format!("blob descriptor for '{column}' is missing the '{name}' field"), - }) - }; - let kinds = child("kind")? - .as_any() - .downcast_ref::() - .ok_or_else(|| Error::Runtime { - message: format!("blob descriptor 'kind' for '{column}' is not a UInt8 array"), - })?; - let positions = child("position")? - .as_any() - .downcast_ref::() - .ok_or_else(|| Error::Runtime { - message: format!("blob descriptor 'position' for '{column}' is not a UInt64 array"), - })?; - let sizes = child("size")? - .as_any() - .downcast_ref::() - .ok_or_else(|| Error::Runtime { - message: format!("blob descriptor 'size' for '{column}' is not a UInt64 array"), - })?; - - // Match Lance `collect_blob_entries_v2` skip condition (`BlobKind::Inline` == 0). - Ok((0..descriptor_struct.len()) - .map(|i| { - descriptor_struct.is_null(i) - || kinds.is_null(i) - || (kinds.value(i) == 0 && positions.value(i) == 0 && sizes.value(i) == 0) }) - .collect()) -} - -fn non_null_row_ids(row_ids: &[u64], null_mask: &[bool]) -> Vec { - row_ids - .iter() - .zip(null_mask) - .filter_map(|(row_id, is_null)| (!is_null).then_some(*row_id)) - .collect() + } else { + Err(Error::Runtime { + message: format!( + "blob read for column '{column}' returned {resolved} results for {requested} row ids" + ), + }) + } } /// Materialize blob-local ranges (same length and order as `requests`, nulls preserved). @@ -282,63 +204,18 @@ pub(crate) async fn take_blob_ranges_aligned( .iter() .map(|request| LanceBlobRangeRequest::new(request.row_id, request.offset, request.length)) .collect::>(); - let reader = dataset - .read_blob_ranges(column) - .map_err(map_blob_read_error)? + let payloads = dataset + .read_blob_ranges(column)? .with_row_ids(lance_requests) - .preserve_order(false); - - let mut stream = reader - .try_into_stream() - .await - .map_err(map_blob_read_error)?; - let mut payloads = vec![None; requests.len()]; - while let Some(result) = stream.try_next().await.map_err(map_blob_read_error)? { - let slot = payloads - .get_mut(result.request_index) - .ok_or_else(|| Error::Runtime { - message: format!( - "blob range read for column '{column}' returned out-of-range request index {} for {} requests", - result.request_index, - requests.len() - ), - })?; - if slot.replace(result.data).is_some() { - return Err(Error::Runtime { - message: format!( - "blob range read for column '{column}' returned request index {} more than once", - result.request_index - ), - }); - } - } - - let missing_request_indices = payloads - .iter() - .enumerate() - .filter_map(|(request_index, payload)| payload.is_none().then_some(request_index)) - .collect::>(); - if !missing_request_indices.is_empty() { - let missing_row_ids = missing_request_indices - .iter() - .map(|request_index| requests[*request_index].row_id) - .collect::>(); - let null_mask = blob_null_mask(dataset, column, &missing_row_ids).await?; - for (request_index, is_null) in missing_request_indices.into_iter().zip(null_mask) { - if !is_null { - return Err(Error::Runtime { - message: format!( - "blob range read for column '{column}' did not return request index {request_index}" - ), - }); - } - } - } + .preserve_order(true) + .execute() + .await?; + ensure_all_row_ids_resolved(column, requests.len(), payloads.len())?; let mut builder = LargeBinaryBuilder::new(); for payload in payloads { - match payload { - Some(data) => builder.append_value(data.as_ref()), + match payload.data { + Some(data) => builder.append_value(data), None => builder.append_null(), } } @@ -356,38 +233,19 @@ pub(crate) async fn take_blobs_aligned( return Ok(LargeBinaryBuilder::new().finish()); } - let null_mask = blob_null_mask(dataset, column, row_ids).await?; - let non_null_row_ids = non_null_row_ids(row_ids, &null_mask); - let non_null_count = non_null_row_ids.len(); - let payloads = if non_null_count == 0 { - Vec::new() - } else { - dataset - .read_blobs(column)? - .with_row_ids(non_null_row_ids) - .preserve_order(true) - .execute() - .await? - }; - - if payloads.len() != non_null_count { - return Err(Error::Runtime { - message: format!( - "blob read for column '{column}' returned {} payloads for {} non-null rows", - payloads.len(), - non_null_count - ), - }); - } + let payloads = dataset + .read_blobs(column)? + .with_row_ids(row_ids.to_vec()) + .preserve_order(true) + .execute() + .await?; + ensure_all_row_ids_resolved(column, row_ids.len(), payloads.len())?; let mut builder = LargeBinaryBuilder::new(); - let mut payload_idx = 0; - for is_null in &null_mask { - if *is_null { - builder.append_null(); - } else { - builder.append_value(payloads[payload_idx].data.as_ref()); - payload_idx += 1; + for payload in payloads { + match payload.data { + Some(data) => builder.append_value(data), + None => builder.append_null(), } } Ok(builder.finish()) @@ -404,34 +262,9 @@ pub(crate) async fn take_blob_files_aligned( return Ok(Vec::new()); } - let null_mask = blob_null_mask(dataset, column, row_ids).await?; - let non_null_row_ids = non_null_row_ids(row_ids, &null_mask); - let handles = if non_null_row_ids.is_empty() { - Vec::new() - } else { - dataset.take_blobs(&non_null_row_ids, column).await? - }; - if handles.len() != non_null_row_ids.len() { - return Err(Error::Runtime { - message: format!( - "blob take for column '{column}' returned {} handles for {} non-null rows", - handles.len(), - non_null_row_ids.len() - ), - }); - } - - let mut handles = handles.into_iter(); - Ok(null_mask - .iter() - .map(|is_null| { - if *is_null { - None - } else { - Some(handles.next().unwrap()) - } - }) - .collect()) + let handles = dataset.take_blobs(row_ids, column).await?; + ensure_all_row_ids_resolved(column, row_ids.len(), handles.len())?; + Ok(handles) } #[cfg(test)] diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index fcf1997c0..0e9354ffc 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -1030,8 +1030,9 @@ impl Table { /// Materialize blob bytes for the given row ids. /// - /// Output matches `row_ids` in length and order. Null and zero-length rows - /// are null. Prefer [`Self::fetch_blob_files`] for large selections. + /// Output matches `row_ids` in length and order. Null blobs are null; + /// valid empty blobs contain empty byte strings. Prefer + /// [`Self::fetch_blob_files`] for large selections. /// /// ``` /// use arrow_array::UInt64Array; @@ -1093,8 +1094,8 @@ impl Table { /// # } /// ``` /// - /// Returns [`Error::InvalidInput`] when a range is invalid, a requested - /// row id does not exist, or the column is not a blob v2 column. Returns + /// Returns an error when a range is invalid, a requested row id does not + /// exist, or the column is not a blob v2 column. Returns /// [`Error::NotSupported`] on table types without blob support. pub async fn fetch_blob_ranges( &self, diff --git a/rust/lancedb/tests/blob_integration.rs b/rust/lancedb/tests/blob_integration.rs index d44eff105..47b80b433 100644 --- a/rust/lancedb/tests/blob_integration.rs +++ b/rust/lancedb/tests/blob_integration.rs @@ -634,14 +634,12 @@ async fn fetch_blob_ranges_validates_requests() -> Result<()> { .fetch_blob_ranges("image", [BlobRangeRequest::new(row_id, 2, 2)]) .await .unwrap_err(); - assert!(matches!(&err, Error::InvalidInput { .. }), "got {err:?}"); assert!(err.to_string().contains("exceeds blob size")); let err = table .fetch_blob_ranges("image", [BlobRangeRequest::new(row_id, u64::MAX, 1)]) .await .unwrap_err(); - assert!(matches!(&err, Error::InvalidInput { .. }), "got {err:?}"); assert!(err.to_string().contains("offset + length overflowed")); let err = table @@ -686,6 +684,32 @@ async fn fetch_blobs_out_of_range_id_errors_without_panic() -> Result<()> { Ok(()) } +#[tokio::test] +async fn fetch_blob_apis_reject_mixed_valid_and_missing_row_ids() -> Result<()> { + let tmp = tempdir().unwrap(); + let db = connect(tmp.path().to_str().unwrap()).execute().await?; + let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"x".as_slice())]).await?; + let row_id = collect_row_ids(&table).await?[0]; + let row_ids = [u64::MAX, row_id]; + + let err = table.fetch_blobs("image", &row_ids).await.unwrap_err(); + assert!(matches!(&err, Error::InvalidInput { .. }), "got {err:?}"); + assert!(err.to_string().contains("row ids")); + + let err = table.fetch_blob_files("image", &row_ids).await.unwrap_err(); + assert!(matches!(&err, Error::InvalidInput { .. }), "got {err:?}"); + assert!(err.to_string().contains("row ids")); + + let requests = row_ids.map(|row_id| BlobRangeRequest::new(row_id, 0, 1)); + let err = table + .fetch_blob_ranges("image", requests) + .await + .unwrap_err(); + assert!(matches!(&err, Error::InvalidInput { .. }), "got {err:?}"); + assert!(err.to_string().contains("row ids")); + Ok(()) +} + #[tokio::test] async fn fetch_blobs_rejects_non_blob_column() -> Result<()> { let tmp = tempdir().unwrap(); @@ -929,7 +953,7 @@ async fn fetch_blobs_with_precompaction_row_ids_survives_compaction() -> Result< } #[tokio::test] -async fn zero_length_blob_reads_back_as_null() -> Result<()> { +async fn empty_blob_reads_back_as_empty_bytes() -> Result<()> { let tmp = tempdir().unwrap(); let db = connect(tmp.path().to_str().unwrap()).execute().await?; let table = create_inline_blob_table(&db, "t", &[1], &[Some(b"".as_slice())]).await?; @@ -937,7 +961,8 @@ async fn zero_length_blob_reads_back_as_null() -> Result<()> { let ids = collect_row_ids(&table).await?; let bytes = table.fetch_blobs("image", &ids).await?; assert_eq!(bytes.len(), 1); - assert!(bytes.is_null(0)); + assert!(!bytes.is_null(0)); + assert!(bytes.value(0).is_empty()); Ok(()) }