diff --git a/docs/src/python/python.md b/docs/src/python/python.md index af3f9d598..b3c0c4672 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -589,6 +589,34 @@ with `search().with_row_id(True).to_arrow()` and pass the result to the fetch AP One missing or deleted row ID rejects the entire blob-read batch; the APIs do not return partial results. +To construct inline Blob V2 arrays for ingestion in record batches, use +[blob_array][lancedb.schema.blob_array]. It accepts bytes and `None`, preserving +the distinction between an empty blob (`b""`) and a null blob: + +```python +import pyarrow as pa +from lancedb.schema import blob, blob_array + +schema = pa.schema([pa.field("video_id", pa.string()), blob("video")]) +batch = pa.RecordBatch.from_arrays( + [pa.array(["a", "b", "c"]), blob_array([b"abc", None, b""])], + schema=schema, +) +table = db.create_table("videos", data=iter([batch]), schema=schema) +table.add(iter([batch])) +``` + +`create_table(..., schema=...)` also casts plain binary values into blob columns +declared in the schema. `table.add(...)` casts them using the table's schema. +These conversions apply to Python rows and plain binary Arrow columns. +PyArrow's `RecordBatch.from_pylist(..., schema=...)` cannot convert raw bytes +into the blob extension type; use `blob_array` and `RecordBatch.from_arrays` +when constructing batches explicitly. + +For external blob references, construct the descriptor struct described by +[blob][lancedb.schema.blob], then wrap it with `pa.ExtensionArray.from_storage`. +Inline `data` and external `uri` representations are mutually exclusive. + LanceDB Distributed SQL returns blob bytes by default. Select `blob_descriptor(video)` to return an openable reference instead, then use [`open_blob_file`][lancedb.table.Table.open_blob_file] or the lazy iterable @@ -647,6 +675,8 @@ so the fetch APIs cannot read their bytes yet. ::: lancedb.blob +::: lancedb.schema.blob_array + ::: lancedb._blob.BlobFile options: show_root_full_path: false diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 32f7dc43c..e13e3dc9c 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -416,6 +416,10 @@ class DBConnection(EnforceOverrides): - pyarrow.Schema - [LanceModel][lancedb.pydantic.LanceModel] + + Plain binary values are cast into blob columns declared with + [blob][lancedb.schema.blob]. To construct Arrow record batches + explicitly, use [blob_array][lancedb.schema.blob_array]. mode: str; default "create" The mode to use when creating the table. Can be "create", "overwrite", or "exist_ok". @@ -2598,6 +2602,10 @@ class AsyncConnection(object): - pyarrow.Schema - [LanceModel][lancedb.pydantic.LanceModel] + + Plain binary values are cast into blob columns declared with + [blob][lancedb.schema.blob]. To construct Arrow record batches + explicitly, use [blob_array][lancedb.schema.blob_array]. mode: Literal["create", "overwrite", "exist_ok"]; default "create" The mode to use when creating the table. Can be "create", "overwrite", or "exist_ok". diff --git a/python/python/lancedb/schema.py b/python/python/lancedb/schema.py index 342053d6a..d92f7f692 100644 --- a/python/python/lancedb/schema.py +++ b/python/python/lancedb/schema.py @@ -6,7 +6,7 @@ import ctypes import importlib -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING, Iterable, Optional import pyarrow as pa import pyarrow.ipc @@ -229,8 +229,30 @@ def blob( Maximum size in bytes of a single packed blob sidecar file before a new one is started. Must be positive. + Notes + ----- Thresholds are validated and stored as Arrow field metadata, with the same format as ``lance.blob_field``. They do not require pylance. + + The extension's storage is a struct with nullable fields ``data`` + (large_binary), ``uri`` (string), ``position`` (uint64), and ``size`` + (uint64). Each non-null blob must set exactly one of ``data`` and ``uri``: + + - Inline bytes use ``data``, leaving all other fields null. ``b""`` is a + valid empty blob; use a null struct for a null blob. + - External blobs use ``uri``, leaving ``data`` null. Set both ``position`` + and ``size`` to select a byte range, or leave both null for the whole + object. A range's ``size`` must be greater than zero. + + Use [blob_array][lancedb.schema.blob_array] to build inline blob arrays + for Arrow record batches. For external descriptors, build the struct + array with ``field.type.storage_type`` and wrap it with + ``pa.ExtensionArray.from_storage(field.type, storage)``. + + ``create_table(..., schema=...)`` casts plain binary values into blob + columns declared by this field, and ``table.add(...)`` does the same + using the table's schema. PyArrow's ``RecordBatch.from_pylist`` does not + perform this conversion; construct blob arrays explicitly instead. """ metadata = {} for threshold, value, allow_zero, key in ( @@ -269,6 +291,50 @@ def blob( return field.with_metadata(metadata) if metadata else field +def blob_array(payloads: Iterable[Optional[bytes]]) -> pa.ExtensionArray: + """Build a Lance blob v2 extension array from inline bytes. + + Uses the same extension type as [blob][lancedb.schema.blob], with or + without pylance installed. + + Parameters + ---------- + payloads: + An iterable of ``bytes`` or ``None``. ``None`` creates a null blob; + ``b""`` creates a valid empty blob. + + Returns + ------- + pa.ExtensionArray + Inline blob values ready to use in Arrow tables or record batches. + + Raises + ------ + TypeError + If a payload is neither ``bytes`` nor ``None``. + + Examples + -------- + >>> import pyarrow as pa + >>> from lancedb.schema import blob, blob_array + >>> schema = pa.schema([blob("video")]) + >>> videos = blob_array([b"abc", None, b""]) + >>> batch = pa.RecordBatch.from_arrays([videos], schema=schema) + >>> batch.num_rows + 3 + """ + blob_type = _resolve_blob_type()() + descriptors = [] + for payload in payloads: + if payload is not None and not isinstance(payload, bytes): + raise TypeError( + f"payloads must contain bytes or None, got {type(payload).__name__}" + ) + descriptors.append(None if payload is None else {"data": payload}) + storage = pa.array(descriptors, type=blob_type.storage_type) + return pa.ExtensionArray.from_storage(blob_type, storage) + + def vector(dimension: int, value_type: pa.DataType = pa.float32()) -> pa.DataType: """A help function to create a vector type. diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 98a841bd0..be42dc802 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -2183,6 +2183,10 @@ class Table(ABC): ) -> AddResult: """Add more data to the [Table][lancedb.table.Table]. + Plain binary values are cast into blob columns using the table's schema. + To construct Arrow record batches explicitly, use + [blob_array][lancedb.schema.blob_array]. + Parameters ---------- data: DATA @@ -6475,6 +6479,10 @@ class AsyncTable: ) -> AddResult: """Add more data to the [AsyncTable][lancedb.table.AsyncTable]. + Plain binary values are cast into blob columns using the table's schema. + To construct Arrow record batches explicitly, use + [blob_array][lancedb.schema.blob_array]. + Parameters ---------- data: DATA diff --git a/python/python/tests/test_blob_array.py b/python/python/tests/test_blob_array.py new file mode 100644 index 000000000..9eb8221f0 --- /dev/null +++ b/python/python/tests/test_blob_array.py @@ -0,0 +1,85 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The LanceDB Authors + +import subprocess +import sys +import textwrap + +import pytest + + +@pytest.mark.parametrize("with_pylance", [False, True]) +def test_blob_array(with_pylance): + # Use fresh processes so each extension class can own Arrow's registry. + script = textwrap.dedent( + """\ + import importlib.abc + import sys + + if sys.argv[1] == "False": + class MissingLance(importlib.abc.MetaPathFinder): + def find_spec(self, fullname, path, target=None): + if fullname == "lance" or fullname.startswith("lance."): + raise ModuleNotFoundError(fullname, name="lance") + sys.meta_path.insert(0, MissingLance()) + + import lancedb + import pyarrow as pa + from lancedb.schema import blob, blob_array + + if sys.argv[1] == "True": + from lance.blob import BlobType + else: + from lancedb.schema import _FallbackBlobType as BlobType + + schema = pa.schema([blob("video")]) + payloads = [b"abc", None, b"", b"\\x00\\xff"] + for values in ([], [None, None], payloads): + for source in (values, iter(values)): + array = blob_array(source) + assert isinstance(array, pa.ExtensionArray) + assert type(array.type) is BlobType + assert array.type == schema.field("video").type + assert array.null_count == values.count(None) + assert array.storage.to_pylist() == [ + None if value is None else { + "data": value, "uri": None, "position": None, "size": None + } + for value in values + ] + batch = pa.RecordBatch.from_arrays([array], schema=schema) + with pa.BufferOutputStream() as sink: + with pa.ipc.new_stream(sink, schema) as writer: + writer.write_batch(batch) + restored = pa.ipc.open_stream(sink.getvalue()).read_next_batch() + assert restored.equals(batch) + + for invalid in ("file:///video", {"data": b"abc"}, 123): + try: + blob_array([invalid]) + except TypeError as error: + assert "bytes or None" in str(error) + else: + raise AssertionError(f"accepted unsupported payload: {invalid!r}") + + # Batch iterators are the ingestion path that needs explicit arrays. + batches = [ + pa.RecordBatch.from_arrays([blob_array([value])], schema=schema) + for value in payloads + ] + table = lancedb.connect("memory:///").create_table( + "videos", data=iter(batches[:2]), schema=schema + ) + table.add(iter(batches[2:])) + row_ids = table.search().with_row_id(True).to_arrow()["_rowid"] + assert table.fetch_blobs("video", row_ids).to_pylist() == payloads + if sys.argv[1] == "False": + assert "lance" not in sys.modules + """ + ) + result = subprocess.run( + [sys.executable, "-c", script, str(with_pylance)], + capture_output=True, + text=True, + ) + assert result.returncode == 0, result.stderr