diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 5f87302ad..d42d70c4a 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -581,6 +581,9 @@ pylance, LanceDB uses a matching `lance.blob.v2` extension type so blob columns still work. Native table queries return descriptors. Call [`fetch_blob_files`][lancedb.table.Table.fetch_blob_files] for lazy reads or [`fetch_blobs`][lancedb.table.Table.fetch_blobs] for eager bytes. +Local tables support these fetch APIs for both legacy blob v1 columns +(`lance-encoding:blob`) and blob v2 columns. For legacy columns, obtain row IDs +with `search().with_row_id(True).to_arrow()` and pass the result to the fetch API. One missing or deleted row ID rejects the entire blob-read batch; the APIs do not return partial results. diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 6df109d6c..7bef43c0d 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -2516,7 +2516,7 @@ class Table(ABC): @abstractmethod def blob_columns(self) -> list[str]: - """Names of blob v2 columns that can be fetched by row id. + """Names of blob v1 and v2 columns that can be fetched by row id. Struct-nested blobs use dotted paths. Blobs inside lists are omitted because a row id does not identify an individual list element. @@ -2530,6 +2530,7 @@ class Table(ABC): ) -> pa.LargeBinaryArray: """Materialize full blob bytes for ``column`` at the given rows. + Local tables support both legacy blob v1 and blob v2 columns. The result has the same length and order as ``row_ids``. Null blobs produce null slots; valid empty blobs produce ``b""``. @@ -2553,8 +2554,9 @@ class Table(ABC): column: str, requests: Sequence[Sequence[int]], ) -> pa.LargeBinaryArray: - """Materialize row-specific byte ranges from a blob v2 column. + """Materialize row-specific byte ranges from a blob column. + Local tables support both legacy blob v1 and blob v2 columns. Each request is a three-item ``(row_id, offset, length)`` sequence. Requests may be repeated or reordered, including multiple ranges for the same blob. @@ -2579,6 +2581,7 @@ class Table(ABC): ) -> "list[Optional[BlobFile]]": """Open lazy, seekable :class:`~lancedb._blob.BlobFile` handles. + Local tables support both legacy blob v1 and blob v2 columns. Prefer this over :meth:`fetch_blobs` for large payloads. ``row_ids`` is a ``list[int]``, an integer Arrow array or chunked array, or a query ``pyarrow.Table`` carrying row identity via ``_rowid`` or a diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py index da362f053..f4c6a9e24 100644 --- a/python/python/tests/test_blob.py +++ b/python/python/tests/test_blob.py @@ -1903,10 +1903,74 @@ def test_legacy_v1_query_omits_auto_row_id(): assert "_rowid" not in hits.column_names -def test_fetch_blobs_rejects_legacy_v1_column(): +def test_fetch_legacy_v1_blobs(): table = _legacy_v1_table("legacy_fetch") - with pytest.raises(ValueError, match="legacy blob column.*blob v2"): - table.fetch_blobs("legacy", [0]) + table.add( + [ + {"id": 2, "legacy": None}, + {"id": 3, "legacy": b""}, + {"id": 4, "legacy": b"another fragment"}, + ] + ) + hits = table.search().select(["id", "legacy"]).with_row_id(True).to_arrow() + assert pa.types.is_struct(hits["legacy"].type) + by_id = dict(zip(hits["id"].to_pylist(), hits["_rowid"].to_pylist())) + row_ids = [by_id[i] for i in [4, 2, 1, 3, 4]] + expected = [b"another fragment", None, b"old", b"", b"another fragment"] + assert table.fetch_blobs("legacy", row_ids).to_pylist() == expected + assert table.blob_columns() == ["legacy"] + assert table.fetch_blobs("legacy", hits).to_pylist() == [ + b"old", + None, + b"", + b"another fragment", + ] + assert table.fetch_blobs("legacy", []).to_pylist() == [] + assert table.fetch_blob_files("legacy", []) == [] + + handles = table.fetch_blob_files("legacy", row_ids) + assert len(handles) == len(expected) + for handle, payload in zip(handles, expected): + if payload is None: + assert handle is None + else: + with handle: + assert handle.size() == len(payload) + assert handle.read(2) == payload[:2] + assert handle.seek(0) == 0 + assert handle.read() == payload + + assert table.fetch_blob_ranges( + "legacy", [(by_id[4], 8, 8), (by_id[2], 0, 1), (by_id[3], 0, 0)] + ).to_pylist() == [b"fragment", None, b""] + + table.delete("id = 2") + for fetch in [table.fetch_blobs, table.fetch_blob_files]: + with pytest.raises(ValueError, match="row ids but some do not exist"): + fetch("legacy", [by_id[1], by_id[2]]) + + +@pytest.mark.asyncio +async def test_async_fetch_legacy_v1_blobs(tmp_path): + table = _legacy_v1_table("legacy_async_source") + # Open a dataset written by Lance independently of the LanceDB connection. + lance.write_dataset( + pa.Table.from_pylist( + [{"id": 1, "legacy": b"old"}, {"id": 2, "legacy": None}], + schema=table.schema, + ), + tmp_path / "legacy.lance", + data_storage_version="2.0", + ) + db = await lancedb.connect_async(tmp_path) + table = await db.open_table("legacy") + assert await table.blob_columns() == ["legacy"] + hits = await table.query().with_row_id().to_arrow() + assert (await table.fetch_blobs("legacy", hits)).to_pylist() == [b"old", None] + handles = await table.fetch_blob_files("legacy", hits) + assert handles[0].read() == b"old" + handles[0].close() + assert handles[1] is None @pytest.mark.asyncio diff --git a/rust/lancedb/src/blob.rs b/rust/lancedb/src/blob.rs index 07808d56c..ce124e891 100644 --- a/rust/lancedb/src/blob.rs +++ b/rust/lancedb/src/blob.rs @@ -1,13 +1,15 @@ // SPDX-License-Identifier: Apache-2.0 // SPDX-FileCopyrightText: Copyright The LanceDB Authors -//! Lance blob v2 columns store large binary payloads out of line. +//! Lance blob columns store large binary payloads out of line. //! -//! Declare a column with [`blob`]. On write, [`crate::table::Table::add`] coerces +//! The read APIs support both legacy blob v1 and blob v2 columns. +//! +//! Declare a blob v2 column with [`blob`]. On write, [`crate::table::Table::add`] coerces //! raw `Binary` / `LargeBinary` into the blob struct layout. Queries return //! small descriptors, not bytes. //! -//! Blob tables require Lance file format >= 2.2. `_rowid` values stay valid +//! Blob v2 tables require Lance file format >= 2.2. `_rowid` values stay valid //! after compaction when the table has stable row ids. Overwrite is a new //! create and does not keep the previous table's stable row id setting. @@ -467,14 +469,27 @@ pub(crate) fn field_tree_has_blob_v2(field: &Field) -> bool { } } -/// Collects the dotted paths of row-addressable blob v2 columns under `field`. +fn field_tree_has_blob(field: &Field) -> bool { + if field.is_blob() { + return true; + } + match field.data_type() { + DataType::Struct(children) => children.iter().any(|c| field_tree_has_blob(c)), + DataType::List(child) | DataType::LargeList(child) | DataType::FixedSizeList(child, _) => { + field_tree_has_blob(child) + } + _ => false, + } +} + +/// Collects the dotted paths of row-addressable blob columns under `field`. fn collect_blob_paths(field: &Field, prefix: &str, paths: &mut Vec) { let path = if prefix.is_empty() { field.name().clone() } else { format!("{prefix}.{}", field.name()) }; - if field.is_blob_v2() { + if field.is_blob() { paths.push(path); return; } @@ -491,7 +506,7 @@ fn is_list_blob_path(field: &Field, prefix: &str, column: &str, inside_list: boo } else { format!("{prefix}.{}", field.name()) }; - if field.is_blob_v2() { + if field.is_blob() { return inside_list && column == path; } match field.data_type() { @@ -499,7 +514,7 @@ fn is_list_blob_path(field: &Field, prefix: &str, column: &str, inside_list: boo .iter() .any(|child| is_list_blob_path(child, &path, column, inside_list)), DataType::List(child) | DataType::LargeList(child) | DataType::FixedSizeList(child, _) => { - (column == path && field_tree_has_blob_v2(child)) + (column == path && field_tree_has_blob(child)) || is_list_blob_path(child, &path, column, true) } _ => false, @@ -527,7 +542,7 @@ pub(crate) fn has_blob_columns(schema: &Schema) -> bool { schema.fields().iter().any(|f| field_tree_has_blob_v2(f)) } -/// Row-addressable blob v2 column paths in `schema`, in declaration order. +/// Row-addressable blob v1 and v2 column paths in `schema`, in declaration order. /// Struct-nested blobs use dotted paths (e.g. `info.blob`); list-nested blobs /// are omitted because a row id cannot identify an individual list element. pub(crate) fn blob_column_names(schema: &Schema) -> Vec { @@ -557,15 +572,37 @@ pub(crate) fn ensure_blob_storage_version(schema: &Schema, params: &mut WritePar } } -/// Validate that `column` exists and is a blob v2 column. -/// -/// Legacy v1 columns (`lance-encoding:blob`) error with a migration hint. -pub(crate) fn ensure_blob_v2_column( +/// Validate that `column` exists and is a row-addressable blob v1 or v2 column. +pub(crate) fn ensure_blob_column( schema: &lance_core::datatypes::Schema, column: &str, ) -> Result<()> { // A top-level blob cannot have a list ancestor. Keep the common read path // from converting the entire schema just to validate that condition. + if column.contains('.') + || schema + .field(column) + .is_none_or(|field| !field.is_blob() && !field.is_blob_v2()) + { + reject_list_blob_column(&Schema::from(schema), column)?; + } + match schema.field(column) { + Some(field) if field.is_blob() || field.is_blob_v2() => Ok(()), + Some(_) => Err(Error::InvalidInput { + message: format!("column '{column}' is not a blob column"), + }), + None => Err(Error::InvalidInput { + message: format!("no column named '{column}' in this table"), + }), + } +} + +/// Validate a blob v2 descriptor source column. +/// +/// Stored SQL descriptors only describe blob v2 layouts. Keep legacy v1 +/// support on the row-id fetch APIs instead of broadening descriptor +/// reopening to a storage format it cannot encode. +fn ensure_blob_v2_column(schema: &lance_core::datatypes::Schema, column: &str) -> Result<()> { if column.contains('.') || schema.field(column).is_none_or(|field| !field.is_blob_v2()) { reject_list_blob_column(&Schema::from(schema), column)?; } @@ -708,7 +745,7 @@ pub(crate) async fn take_blob_ranges_aligned( column: &str, requests: &[BlobRangeRequest], ) -> Result { - ensure_blob_v2_column(dataset.schema(), column)?; + ensure_blob_column(dataset.schema(), column)?; if requests.is_empty() { return Ok(LargeBinaryBuilder::new().finish()); } @@ -749,7 +786,7 @@ pub(crate) async fn take_blobs_aligned( column: &str, row_ids: &[u64], ) -> Result { - ensure_blob_v2_column(dataset.schema(), column)?; + ensure_blob_column(dataset.schema(), column)?; if row_ids.is_empty() { return Ok(LargeBinaryBuilder::new().finish()); } @@ -782,7 +819,7 @@ pub(crate) async fn take_blob_files_aligned( column: &str, row_ids: &[u64], ) -> Result>> { - ensure_blob_v2_column(dataset.schema(), column)?; + ensure_blob_column(dataset.schema(), column)?; if row_ids.is_empty() { return Ok(Vec::new()); } @@ -930,20 +967,35 @@ mod tests { } #[test] - fn legacy_v1_blob_column_is_rejected_with_migration_hint() { + fn legacy_v1_blob_columns_are_row_addressable() { let legacy = Field::new("image", DataType::LargeBinary, true).with_metadata( std::collections::HashMap::from([( "lance-encoding:blob".to_string(), "true".to_string(), )]), ); - let arrow_schema = Schema::new(vec![legacy]); + let arrow_schema = Schema::new(vec![ + legacy.clone(), + Field::new("info", DataType::Struct(vec![legacy.clone()].into()), true), + Field::new("images", DataType::List(Arc::new(legacy)), true), + ]); let lance_schema = lance_core::datatypes::Schema::try_from(&arrow_schema).unwrap(); - let err = ensure_blob_v2_column(&lance_schema, "image").unwrap_err(); - assert!(matches!(err, Error::InvalidInput { .. })); - assert!(err.to_string().contains("legacy blob column")); - assert!(err.to_string().contains("lance.blob.v2")); + ensure_blob_column(&lance_schema, "image").unwrap(); + ensure_blob_column(&lance_schema, "info.image").unwrap(); + assert_eq!( + blob_column_names(&arrow_schema), + vec!["image", "info.image"] + ); + for path in ["images", "images.image"] { + let err = ensure_blob_column(&lance_schema, path).unwrap_err(); + assert!( + err.to_string() + .contains("blobs inside lists cannot be fetched") + ); + } + // Legacy columns do not require the blob v2 write format. + assert!(!has_blob_columns(&arrow_schema)); } #[test] @@ -951,10 +1003,10 @@ mod tests { let arrow_schema = Schema::new(vec![Field::new("id", DataType::Int64, false)]); let lance_schema = lance_core::datatypes::Schema::try_from(&arrow_schema).unwrap(); - let err = ensure_blob_v2_column(&lance_schema, "id").unwrap_err(); + let err = ensure_blob_column(&lance_schema, "id").unwrap_err(); assert!(err.to_string().contains("'id' is not a blob column")); - let err = ensure_blob_v2_column(&lance_schema, "missing").unwrap_err(); + let err = ensure_blob_column(&lance_schema, "missing").unwrap_err(); assert!(err.to_string().contains("no column named 'missing'")); } diff --git a/rust/lancedb/src/remote/table/blobs.rs b/rust/lancedb/src/remote/table/blobs.rs index e3cdfc6a6..65ab13935 100644 --- a/rust/lancedb/src/remote/table/blobs.rs +++ b/rust/lancedb/src/remote/table/blobs.rs @@ -405,7 +405,7 @@ impl RemoteTable { error } - /// Blob v2 columns are marked in field metadata, which `describe` returns. Reading + /// Blob columns are marked in field metadata, which `describe` returns. Reading /// them from the cached schema needs no route of its own and no version gate. pub(super) async fn blob_columns_impl(&self) -> Result> { let schema = self.schema().await?; diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index d6575b959..7097644e4 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -766,7 +766,7 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync { async fn close_rte_writers(&self) -> Result<()> { Ok(()) } - /// Names of the blob v2 columns in this table, in declaration order. + /// Names of the blob v1 and v2 columns in this table, in declaration order. async fn blob_columns(&self) -> Result> { Err(Error::NotSupported { message: "blob_columns is not supported on this table type".into(), @@ -1251,7 +1251,7 @@ impl Table { self.inner.count_rows(filter).await } - /// Names of the row-addressable blob v2 columns, in declaration order. + /// Names of the row-addressable blob v1 and v2 columns, in declaration order. /// /// Blobs nested in structs use dotted paths (e.g. `info.blob`). Blobs /// inside lists are omitted because row ids cannot identify their elements. @@ -1262,6 +1262,7 @@ impl Table { /// Materialize blob bytes for the given row ids. /// + /// Native tables support both legacy blob v1 and blob v2 columns. /// Output matches `row_ids` in length and order. Null blobs are null; /// valid empty blobs contain empty byte strings. Cloud limits individual /// requests to 1024 row ids and 64 MiB of blob bytes; the remote client @@ -1297,7 +1298,7 @@ impl Table { /// /// Blobs inside lists cannot be fetched by row id. Returns /// [`Error::InvalidInput`] for those paths, when the column does not exist, - /// or when it is not a blob v2 column. Returns [`Error::NotSupported`] on + /// or when it is not a blob column. Returns [`Error::NotSupported`] on /// table types without blob support. pub async fn fetch_blobs( &self, @@ -1307,8 +1308,9 @@ impl Table { self.inner.fetch_blobs(column.as_ref(), row_ids).await } - /// Materialize row-specific ranges from a blob v2 column. + /// Materialize row-specific ranges from a blob column. /// + /// Native tables support both legacy blob v1 and blob v2 columns. /// Each request contains a row id and a blob-local offset and length. /// Requests may be duplicated or reordered, including multiple /// ranges for the same blob. The output has the same length and order as @@ -1341,7 +1343,7 @@ impl Table { /// ``` /// /// 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 + /// exist, or the column is not a blob column. Returns /// [`Error::NotSupported`] on table types without blob support. pub async fn fetch_blob_ranges( &self, @@ -1356,6 +1358,7 @@ impl Table { /// Open lazy [`BlobFile`] handles for the given row ids. /// + /// Native tables support both legacy blob v1 and blob v2 columns. /// Same length and order as `row_ids`. Null rows are `None`. Bytes are not /// read from disk until a call to [`BlobFile::read`]. /// Remote blob requests share a connection-wide concurrency limit. Set diff --git a/rust/lancedb/tests/blob_integration.rs b/rust/lancedb/tests/blob_integration.rs index 8559a0e43..0a4240038 100644 --- a/rust/lancedb/tests/blob_integration.rs +++ b/rust/lancedb/tests/blob_integration.rs @@ -947,12 +947,15 @@ async fn fetch_blobs_rejects_unknown_column() -> Result<()> { Ok(()) } +#[rstest::rstest] +#[case("2.0")] +#[case("2.1")] #[tokio::test] -async fn fetch_blobs_rejects_legacy_v1_blob_column() -> Result<()> { +async fn fetch_blob_apis_read_legacy_v1_blob_column(#[case] storage_version: &str) -> Result<()> { let tmp = tempdir().unwrap(); // Legacy v1 blob columns are only writable at file version <= 2.1. let db = connect(tmp.path().to_str().unwrap()) - .storage_options([(OPT_NEW_TABLE_STORAGE_VERSION, "2.1")]) + .storage_options([(OPT_NEW_TABLE_STORAGE_VERSION, storage_version)]) .execute() .await?; let legacy = Field::new("image", DataType::LargeBinary, true).with_metadata( @@ -963,9 +966,62 @@ async fn fetch_blobs_rejects_legacy_v1_blob_column() -> Result<()> { legacy, ])); let table = db.create_empty_table("t", schema).execute().await?; + table + .add(binary_input_batch( + &[1, 2, 3, 4], + &[Some(b"abcdef"), None, Some(b""), Some(b"uvwxyz")], + )) + .execute() + .await?; - let err = table.fetch_blobs("image", &[0]).await.unwrap_err(); - assert!(err.to_string().contains("legacy blob column")); + assert!(!supports_blob_v2(storage_format_version(&table).await)); + assert_eq!(table.blob_columns().await?, vec!["image"]); + + let pairs = collect_id_rowid(&table).await?; + let by_id = |want: i64| pairs.iter().find(|(id, _)| *id == want).unwrap().1; + let row_ids = [by_id(4), by_id(2), by_id(1), by_id(3), by_id(4)]; + let expected: [Option<&[u8]>; 5] = [ + Some(b"uvwxyz"), + None, + Some(b"abcdef"), + Some(b""), + Some(b"uvwxyz"), + ]; + let bytes = table.fetch_blobs("image", &row_ids).await?; + assert_eq!(bytes.iter().collect::>(), expected); + + let files = table.fetch_blob_files("image", &row_ids).await?; + assert_eq!(files.len(), expected.len()); + for (file, payload) in files.iter().zip(expected) { + match payload { + Some(payload) => { + let file = file.as_ref().unwrap(); + assert_eq!(file.size(), payload.len() as u64); + assert_eq!(file.read().await?.as_ref(), payload); + } + None => assert!(file.is_none()), + } + } + assert_eq!( + files[2].as_ref().unwrap().read_range(1..4).await?.as_ref(), + b"bcd" + ); + + let ranges = table + .fetch_blob_ranges( + "image", + [ + BlobRangeRequest::new(by_id(4), 1, 3), + BlobRangeRequest::new(by_id(2), 0, 0), + BlobRangeRequest::new(by_id(1), 2, 3), + BlobRangeRequest::new(by_id(3), 0, 0), + BlobRangeRequest::new(by_id(4), 0, 2), + ], + ) + .await?; + let expected_ranges: [Option<&[u8]>; 5] = + [Some(b"vwx"), None, Some(b"cde"), Some(b""), Some(b"uv")]; + assert_eq!(ranges.iter().collect::>(), expected_ranges); Ok(()) } @@ -1406,7 +1462,7 @@ async fn v1_blob_descriptors(table: &Table) -> Result Result>)>> { let pairs = sorted_id_rowid(table).await?; let row_ids: Vec = pairs.iter().map(|(_, row_id)| *row_id).collect();