mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-13 14:16:54 +00:00
feat(python): allow Blob v2 under a list in Function outputs
A Function that produces several blobs per row, such as frames from a video, could not declare `pa.list_(lancedb.blob("item"))` as its output: signatures rejected Blob v2 under any collection.
Outputs now accept Blob v2 inside list and large-list fields; the server stores each element as a Blob v2 value. Inputs, fixed-size lists and maps still reject it (lance stores Blob v2 under neither), and the `udf()` docstring now says where Blob v2 may appear.
This commit is contained in:
@@ -747,12 +747,14 @@ _GRAMMAR_PRIMITIVES = (
|
||||
)
|
||||
|
||||
|
||||
def _canonical_arrow_type(data_type: pa.DataType) -> str:
|
||||
def _canonical_arrow_type(
|
||||
data_type: pa.DataType, *, collection_blobs: bool = False
|
||||
) -> str:
|
||||
"""The compact Function grammar, or canonical exact JSON for nested types."""
|
||||
grammar = _grammar_arrow_type(data_type)
|
||||
if grammar is not None:
|
||||
return grammar
|
||||
exact = _exact_arrow_type(data_type)
|
||||
exact = _exact_arrow_type(data_type, collection_blobs=collection_blobs)
|
||||
return json.dumps(exact, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
||||
|
||||
|
||||
@@ -864,11 +866,12 @@ def _has_supported_blob_v2_layout(field: pa.Field) -> bool:
|
||||
)
|
||||
|
||||
|
||||
def _canonical_arrow_field(field: pa.Field) -> str:
|
||||
def _canonical_arrow_field(field: pa.Field, *, collection_blobs: bool = False) -> str:
|
||||
"""``collection_blobs`` admits Blob v2 under a list, which only outputs support."""
|
||||
_validate_exact_arrow_field(field)
|
||||
if _is_blob_v2_field(field):
|
||||
return _FUNCTION_BLOB_V2_TYPE
|
||||
return _canonical_arrow_type(field.type)
|
||||
return _canonical_arrow_type(field.type, collection_blobs=collection_blobs)
|
||||
|
||||
|
||||
def _blob_storage_type(field: pa.Field) -> pa.DataType:
|
||||
@@ -920,11 +923,14 @@ def _data_type_has_blob_v2(data_type: pa.DataType) -> bool:
|
||||
|
||||
|
||||
def _exact_arrow_field(
|
||||
field: pa.Field, *, inside_collection: bool = False
|
||||
field: pa.Field,
|
||||
*,
|
||||
inside_collection: bool = False,
|
||||
collection_blobs: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
_validate_exact_arrow_field(field)
|
||||
if _is_blob_v2_field(field):
|
||||
if inside_collection:
|
||||
if inside_collection and not collection_blobs:
|
||||
raise TypeError(_NESTED_BLOB_COLLECTION_ERROR)
|
||||
return {
|
||||
"name": field.name,
|
||||
@@ -937,13 +943,20 @@ def _exact_arrow_field(
|
||||
value = {
|
||||
"name": field.name,
|
||||
"nullable": field.nullable,
|
||||
"type": _exact_arrow_type(field.type, inside_collection=inside_collection),
|
||||
"type": _exact_arrow_type(
|
||||
field.type,
|
||||
inside_collection=inside_collection,
|
||||
collection_blobs=collection_blobs,
|
||||
),
|
||||
}
|
||||
return value
|
||||
|
||||
|
||||
def _exact_arrow_type(
|
||||
data_type: pa.DataType, *, inside_collection: bool = False
|
||||
data_type: pa.DataType,
|
||||
*,
|
||||
inside_collection: bool = False,
|
||||
collection_blobs: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
for candidate, name in _GRAMMAR_PRIMITIVES:
|
||||
if data_type == candidate:
|
||||
@@ -959,7 +972,11 @@ def _exact_arrow_type(
|
||||
return {
|
||||
"type": "struct",
|
||||
"fields": [
|
||||
_exact_arrow_field(field, inside_collection=inside_collection)
|
||||
_exact_arrow_field(
|
||||
field,
|
||||
inside_collection=inside_collection,
|
||||
collection_blobs=collection_blobs,
|
||||
)
|
||||
for field in fields
|
||||
],
|
||||
}
|
||||
@@ -987,7 +1004,13 @@ def _exact_arrow_type(
|
||||
else "fixed_size_list"
|
||||
),
|
||||
"fields": [
|
||||
_exact_arrow_field(data_type.value_field, inside_collection=True)
|
||||
_exact_arrow_field(
|
||||
data_type.value_field,
|
||||
inside_collection=True,
|
||||
# Lance stores Blob v2 under list and large_list only.
|
||||
collection_blobs=collection_blobs
|
||||
and not pa.types.is_fixed_size_list(data_type),
|
||||
)
|
||||
],
|
||||
}
|
||||
if pa.types.is_fixed_size_list(data_type):
|
||||
@@ -1108,7 +1131,7 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
|
||||
raise ValueError("Function output must be non-nullable")
|
||||
return FunctionOutput(
|
||||
kind="scalar",
|
||||
arrow_type=_canonical_arrow_field(field),
|
||||
arrow_type=_canonical_arrow_field(field, collection_blobs=True),
|
||||
nullable=False,
|
||||
)
|
||||
|
||||
@@ -1124,7 +1147,7 @@ def _function_output(output: pa.DataType | pa.Field | pa.Schema) -> FunctionOutp
|
||||
fields=tuple(
|
||||
FunctionResultField(
|
||||
name=field.name,
|
||||
arrow_type=_canonical_arrow_field(field),
|
||||
arrow_type=_canonical_arrow_field(field, collection_blobs=True),
|
||||
nullable=field.nullable,
|
||||
)
|
||||
for field in fields
|
||||
@@ -2353,6 +2376,13 @@ def udf(
|
||||
mode (and as ``bytes`` in row mode). Return a ``LargeBinaryArray`` with the
|
||||
declared nullability. Use :func:`lancedb.blob` in the explicit schemas;
|
||||
plain ``pyarrow.large_binary()`` does not carry Blob v2 semantics.
|
||||
|
||||
A Blob v2 field may sit at the top level or inside a struct in inputs and
|
||||
outputs. Only outputs may nest it in a list or large list (for example
|
||||
``pa.list_(lancedb.blob("item"))``, returned as a list of ``bytes``);
|
||||
fixed-size lists and maps are not supported. List-nested Blob v2 values
|
||||
are queryable as Blob descriptors, but the row-id blob fetch APIs and
|
||||
``blob_mode="bytes"`` cannot fetch payloads inside lists.
|
||||
"""
|
||||
|
||||
def decorate(target: Any) -> UdfDefinition:
|
||||
|
||||
@@ -31,6 +31,7 @@ from lancedb.functions import (
|
||||
SecretReference,
|
||||
UdfDefinition,
|
||||
_canonical_arrow_type,
|
||||
_exact_arrow_type,
|
||||
_GRAMMAR_PRIMITIVES,
|
||||
udf,
|
||||
)
|
||||
@@ -1213,6 +1214,74 @@ def test_struct_blob_signature_supports_multiple_struct_levels():
|
||||
assert blob["metadata"]["ARROW:extension:name"] == "lance.blob.v2"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"data_type",
|
||||
[
|
||||
pa.list_(lancedb.blob("item", nullable=False)),
|
||||
pa.large_list(lancedb.blob("item", nullable=False)),
|
||||
pa.list_(pa.struct([lancedb.blob("image", nullable=False)])),
|
||||
],
|
||||
)
|
||||
def test_blob_output_accepts_list_ancestors(data_type):
|
||||
@udf(
|
||||
input_schema=pa.schema([lancedb.blob("video", nullable=False)]),
|
||||
output_schema=pa.field("frames", data_type, nullable=False),
|
||||
)
|
||||
def extract_frames(video: bytes) -> list:
|
||||
return [video]
|
||||
|
||||
output = extract_frames.registration_request.signature.output
|
||||
assert json.loads(output.arrow_type) == _exact_arrow_type(
|
||||
data_type, collection_blobs=True
|
||||
)
|
||||
assert "lance.blob.v2" in output.arrow_type
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"data_type",
|
||||
[
|
||||
pa.map_(pa.string(), lancedb.blob("value", nullable=False).type),
|
||||
pa.list_(lancedb.blob("item", nullable=False), 2),
|
||||
pa.list_(pa.list_(lancedb.blob("item", nullable=False), 2)),
|
||||
],
|
||||
)
|
||||
def test_blob_output_rejects_map_and_fixed_size_list_ancestors(data_type):
|
||||
with pytest.raises(TypeError, match="nested under collection types"):
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([pa.field("id", pa.int64(), nullable=False)]),
|
||||
output_schema=pa.field("frames", data_type, nullable=False),
|
||||
)
|
||||
def frames(id: int) -> dict:
|
||||
return {}
|
||||
|
||||
|
||||
def test_blob_output_named_struct_accepts_list_ancestor():
|
||||
frames_type = pa.list_(lancedb.blob("item", nullable=False))
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([pa.field("id", pa.int64(), nullable=False)]),
|
||||
output_schema=pa.schema(
|
||||
[
|
||||
pa.field("frames", frames_type, nullable=False),
|
||||
pa.field("count", pa.int64(), nullable=False),
|
||||
]
|
||||
),
|
||||
)
|
||||
def extract_frames(id: int) -> dict:
|
||||
return {"frames": [], "count": 0}
|
||||
|
||||
output = extract_frames.registration_request.signature.output
|
||||
assert output.kind == "named_struct"
|
||||
assert [field.name for field in output.fields] == ["frames", "count"]
|
||||
assert json.loads(output.fields[0].arrow_type) == _exact_arrow_type(
|
||||
frames_type, collection_blobs=True
|
||||
)
|
||||
assert "lance.blob.v2" in output.fields[0].arrow_type
|
||||
assert output.fields[1].arrow_type == "int64"
|
||||
assert output.fields[1].nullable is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"data_type",
|
||||
[
|
||||
@@ -1939,6 +2008,86 @@ def test_vectorized_blob_udf_executes_packaged_artifact_and_publishes(tmp_path):
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("declared_list", "physical_list", "array_type", "declared_name"),
|
||||
[
|
||||
(pa.list_, pa.list_, pa.ListArray, "list"),
|
||||
(pa.large_list, pa.large_list, pa.LargeListArray, "large_list"),
|
||||
],
|
||||
ids=["list", "large_list"],
|
||||
)
|
||||
def test_blob_list_output_fresh_artifact_survives_local_reopen(
|
||||
tmp_path, declared_list, physical_list, array_type, declared_name
|
||||
):
|
||||
input_type = physical_list(pa.int64())
|
||||
item = lancedb.blob("item", nullable=True)
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([pa.field("frames", input_type, nullable=False)]),
|
||||
output_schema=pa.field("frames", declared_list(item), nullable=False),
|
||||
)
|
||||
def extract_frames(frames: pa.Array) -> pa.Array:
|
||||
output_type = (
|
||||
pa.large_list(pa.large_binary())
|
||||
if isinstance(frames, pa.LargeListArray)
|
||||
else pa.list_(pa.large_binary())
|
||||
)
|
||||
return pa.array([[b"abc", None, b""], []], type=output_type)
|
||||
|
||||
request = json.loads(extract_frames.registration_request.to_canonical_json())
|
||||
declared = json.loads(request["signature"]["output"]["arrow_type"])
|
||||
assert declared["type"] == declared_name
|
||||
item_field = declared["fields"][0]
|
||||
assert item_field["metadata"]["ARROW:extension:name"] == "lance.blob.v2"
|
||||
|
||||
result = _execute_registered_array_udf_v1(
|
||||
request,
|
||||
pa.RecordBatch.from_arrays(
|
||||
[pa.array([[0], [1]], type=input_type)],
|
||||
names=["frames"],
|
||||
),
|
||||
)
|
||||
assert isinstance(result, array_type)
|
||||
assert result.type == physical_list(pa.large_binary())
|
||||
assert result.offsets.to_pylist() == [0, 3, 3]
|
||||
assert result.values.to_pylist() == [b"abc", None, b""]
|
||||
assert result.to_pylist() == [[b"abc", None, b""], []]
|
||||
|
||||
db = lancedb.connect(tmp_path)
|
||||
table = db.create_table(
|
||||
"fresh_artifact_frames",
|
||||
schema=pa.schema(
|
||||
[
|
||||
pa.field("id", pa.int64(), nullable=False),
|
||||
pa.field("frames", declared_list(item), nullable=False),
|
||||
]
|
||||
),
|
||||
)
|
||||
table.add(
|
||||
pa.Table.from_arrays(
|
||||
[pa.array([1, 2], type=pa.int64()), result],
|
||||
names=["id", "frames"],
|
||||
)
|
||||
)
|
||||
|
||||
reopened = lancedb.connect(tmp_path).open_table("fresh_artifact_frames")
|
||||
reopened_item = reopened.schema.field("frames").type.value_field
|
||||
assert reopened_item.type.extension_name == "lance.blob.v2"
|
||||
assert reopened.blob_columns() == []
|
||||
hits = reopened.search().limit(2).to_arrow().sort_by([("id", "ascending")])
|
||||
lists = hits["frames"].combine_chunks()
|
||||
assert isinstance(lists, array_type)
|
||||
assert lists.offsets.to_pylist() == [0, 3, 3]
|
||||
|
||||
rows = lists.to_pylist()
|
||||
assert rows[1] == []
|
||||
assert [None if value is None else value["size"] for value in rows[0]] == [
|
||||
3,
|
||||
None,
|
||||
0,
|
||||
]
|
||||
|
||||
|
||||
def test_a_function_namespace_path_must_be_a_list_of_segments():
|
||||
"""A bare string is refused rather than read as a sequence of one-character
|
||||
segments, and nothing reaches the service."""
|
||||
|
||||
Reference in New Issue
Block a user