mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-13 14:16:54 +00:00
fix(python): add a Blob V2 array construction helper (#4734)
Fixes #4733 PyArrow expects Blob V2 storage structs when constructing a record batch and cannot infer the extension array from raw bytes. LanceDB already converts binary values during ingestion, but batch writers had no supported array constructor. Add `lancedb.schema.blob_array(payloads)` for iterables of inline bytes and nulls, using the same extension type resolver as `blob()`. The helper preserves empty blobs and null blobs and works with both pylance's type and LanceDB's fallback. Document the descriptor fields, exclusive inline/external representations, batch construction, and existing binary conversion in synchronous and asynchronous `create_table` / `add` APIs. Validation: - Reproduced the reported `ArrowTypeError` on the original code; both new regression cases fail against the original schema module. - 256 focused blob tests and schema doctests passed; one optional PyAV test skipped. - Both new regression cases also passed with PyArrow 25, with and without pylance imports. - Python package Ruff checks and formatting passed. - MkDocs build passed, with an existing unrelated `SecretReference` cross-reference warning. Repository-wide Ruff reports 48 existing errors in `docs/web-tests`. <!-- lance-gatekeeper-fix:v1 agent=20d1a296d76ff5eb914c0848428a747a generation=1 --> Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
This commit is contained in:
@@ -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".
|
||||
|
||||
@@ -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.
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user