mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-03 20:18:54 +00:00
feat: support nested blob function signatures
This commit is contained in:
@@ -521,6 +521,12 @@ class RefreshColumnResult(_RemoteValue):
|
||||
|
||||
_FUNCTION_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_.-]*$")
|
||||
_FUNCTION_BLOB_V2_TYPE = "blob_v2"
|
||||
_ARROW_EXTENSION_NAME_KEY = "ARROW:extension:name"
|
||||
_BLOB_V2_EXTENSION_NAME = "lance.blob.v2"
|
||||
_NESTED_BLOB_COLLECTION_ERROR = (
|
||||
"unsupported Arrow type for Function signature: Blob v2 fields nested under "
|
||||
"collection types are not supported"
|
||||
)
|
||||
|
||||
|
||||
_GRAMMAR_PRIMITIVES = (
|
||||
@@ -591,6 +597,19 @@ def _validate_exact_arrow_field(field: pa.Field) -> None:
|
||||
"unsupported Arrow type for Function signature: lance.blob.v2 "
|
||||
f"requires a supported Blob storage layout, got {field}"
|
||||
)
|
||||
metadata = {
|
||||
(key.decode() if isinstance(key, bytes) else key): (
|
||||
value.decode() if isinstance(value, bytes) else value
|
||||
)
|
||||
for key, value in (field.metadata or {}).items()
|
||||
}
|
||||
if metadata and metadata != {
|
||||
_ARROW_EXTENSION_NAME_KEY: _BLOB_V2_EXTENSION_NAME
|
||||
}:
|
||||
raise TypeError(
|
||||
"unsupported Arrow type for Function signature: lance.blob.v2 "
|
||||
"field metadata must contain only its canonical extension marker"
|
||||
)
|
||||
elif field.metadata:
|
||||
raise TypeError(
|
||||
"unsupported Arrow type for Function signature: field metadata "
|
||||
@@ -655,23 +674,84 @@ def _canonical_arrow_field(field: pa.Field) -> str:
|
||||
return _canonical_arrow_type(field.type)
|
||||
|
||||
|
||||
def _exact_arrow_field(field: pa.Field) -> dict[str, Any]:
|
||||
def _blob_storage_type(field: pa.Field) -> pa.DataType:
|
||||
data_type = field.type
|
||||
if isinstance(data_type, pa.ExtensionType):
|
||||
return data_type.storage_type
|
||||
return data_type
|
||||
|
||||
|
||||
def _exact_blob_storage_type(field: pa.Field) -> dict[str, Any]:
|
||||
storage = _blob_storage_type(field)
|
||||
if not pa.types.is_struct(storage):
|
||||
raise TypeError(
|
||||
"unsupported Arrow type for Function signature: lance.blob.v2 "
|
||||
"requires struct storage"
|
||||
)
|
||||
return {
|
||||
"type": "struct",
|
||||
"fields": [
|
||||
{
|
||||
"name": child.name,
|
||||
"nullable": child.nullable,
|
||||
"type": (
|
||||
{"type": "large_binary"}
|
||||
if pa.types.is_large_binary(child.type)
|
||||
else _exact_arrow_type(child.type)
|
||||
),
|
||||
}
|
||||
for child in storage
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _data_type_has_blob_v2(data_type: pa.DataType) -> bool:
|
||||
if pa.types.is_struct(data_type):
|
||||
return any(
|
||||
_is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
|
||||
for field in data_type
|
||||
)
|
||||
if (
|
||||
pa.types.is_list(data_type)
|
||||
or pa.types.is_large_list(data_type)
|
||||
or pa.types.is_fixed_size_list(data_type)
|
||||
):
|
||||
field = data_type.value_field
|
||||
return _is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
|
||||
if pa.types.is_map(data_type):
|
||||
return any(
|
||||
_is_blob_v2_field(field) or _data_type_has_blob_v2(field.type)
|
||||
for field in (data_type.key_field, data_type.item_field)
|
||||
)
|
||||
return False
|
||||
|
||||
|
||||
def _exact_arrow_field(
|
||||
field: pa.Field, *, inside_collection: bool = False
|
||||
) -> dict[str, Any]:
|
||||
_validate_exact_arrow_field(field)
|
||||
if _is_blob_v2_field(field):
|
||||
raise TypeError(
|
||||
"unsupported Arrow type for Function signature: nested Blob v2 "
|
||||
"fields are not supported; declare Blob parameters or named result "
|
||||
"fields directly"
|
||||
)
|
||||
if inside_collection:
|
||||
raise TypeError(_NESTED_BLOB_COLLECTION_ERROR)
|
||||
return {
|
||||
"name": field.name,
|
||||
"nullable": field.nullable,
|
||||
"type": _exact_blob_storage_type(field),
|
||||
"metadata": {
|
||||
_ARROW_EXTENSION_NAME_KEY: _BLOB_V2_EXTENSION_NAME,
|
||||
},
|
||||
}
|
||||
value = {
|
||||
"name": field.name,
|
||||
"nullable": field.nullable,
|
||||
"type": _exact_arrow_type(field.type),
|
||||
"type": _exact_arrow_type(field.type, inside_collection=inside_collection),
|
||||
}
|
||||
return value
|
||||
|
||||
|
||||
def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
|
||||
def _exact_arrow_type(
|
||||
data_type: pa.DataType, *, inside_collection: bool = False
|
||||
) -> dict[str, Any]:
|
||||
for candidate, name in _GRAMMAR_PRIMITIVES:
|
||||
if data_type == candidate:
|
||||
return {"type": name}
|
||||
@@ -685,7 +765,10 @@ def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
|
||||
)
|
||||
return {
|
||||
"type": "struct",
|
||||
"fields": [_exact_arrow_field(field) for field in fields],
|
||||
"fields": [
|
||||
_exact_arrow_field(field, inside_collection=inside_collection)
|
||||
for field in fields
|
||||
],
|
||||
}
|
||||
if (
|
||||
pa.types.is_list(data_type)
|
||||
@@ -710,11 +793,15 @@ def _exact_arrow_type(data_type: pa.DataType) -> dict[str, Any]:
|
||||
if pa.types.is_large_list(data_type)
|
||||
else "fixed_size_list"
|
||||
),
|
||||
"fields": [_exact_arrow_field(data_type.value_field)],
|
||||
"fields": [
|
||||
_exact_arrow_field(data_type.value_field, inside_collection=True)
|
||||
],
|
||||
}
|
||||
if pa.types.is_fixed_size_list(data_type):
|
||||
value["length"] = data_type.list_size
|
||||
return value
|
||||
if pa.types.is_map(data_type) and _data_type_has_blob_v2(data_type):
|
||||
raise TypeError(_NESTED_BLOB_COLLECTION_ERROR)
|
||||
raise TypeError(f"unsupported Arrow type for Function signature: {data_type}")
|
||||
|
||||
|
||||
|
||||
@@ -668,6 +668,167 @@ def test_blob_fields_use_the_scalar_function_semantic_type():
|
||||
assert signature.output.arrow_type == "blob_v2"
|
||||
|
||||
|
||||
def test_whole_named_struct_function_can_include_a_blob_result_field():
|
||||
@udf(
|
||||
input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
|
||||
output_schema=pa.field(
|
||||
"payload",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field("mime_type", pa.string(), nullable=False),
|
||||
lancedb.blob("image", nullable=False),
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
),
|
||||
)
|
||||
def inspect_blob(image):
|
||||
return {"mime_type": "image/png", "image": image}
|
||||
|
||||
output = inspect_blob.registration_request.signature.output
|
||||
assert output.kind == "named_struct"
|
||||
assert [(field.name, field.arrow_type) for field in output.fields] == [
|
||||
("mime_type", "utf8"),
|
||||
("image", "blob_v2"),
|
||||
]
|
||||
|
||||
|
||||
def test_struct_blob_signature_fields_preserve_exact_metadata_and_nullability():
|
||||
nested_input = pa.field(
|
||||
"payload",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field("mime_type", pa.string(), nullable=False),
|
||||
pa.field(
|
||||
"nested",
|
||||
pa.struct([lancedb.blob("image", nullable=True)]),
|
||||
nullable=True,
|
||||
),
|
||||
]
|
||||
),
|
||||
nullable=True,
|
||||
)
|
||||
nested_output = pa.field(
|
||||
"result",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field("mime_type", pa.string(), nullable=False),
|
||||
pa.field(
|
||||
"nested",
|
||||
pa.struct([lancedb.blob("image", nullable=True)]),
|
||||
nullable=False,
|
||||
),
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
)
|
||||
|
||||
@udf(input_schema=pa.schema([nested_input]), output_schema=nested_output)
|
||||
def copy_payload(payload):
|
||||
return payload
|
||||
|
||||
signature = copy_payload.registration_request.signature
|
||||
input_type = json.loads(signature.inputs[0].arrow_type)
|
||||
assert input_type["fields"][1]["nullable"] is True
|
||||
input_blob = input_type["fields"][1]["type"]["fields"][0]
|
||||
assert input_blob["nullable"] is True
|
||||
assert input_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
|
||||
|
||||
assert signature.output.kind == "named_struct"
|
||||
nested_result = next(
|
||||
field for field in signature.output.fields if field.name == "nested"
|
||||
)
|
||||
output_type = json.loads(nested_result.arrow_type)
|
||||
output_blob = output_type["fields"][0]
|
||||
assert output_blob["nullable"] is True
|
||||
assert output_blob["metadata"] == {"ARROW:extension:name": "lance.blob.v2"}
|
||||
|
||||
|
||||
def test_struct_blob_signature_supports_multiple_struct_levels():
|
||||
recursive = pa.field(
|
||||
"value",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field(
|
||||
"level_1",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field(
|
||||
"level_2",
|
||||
pa.struct([lancedb.blob("image", nullable=False)]),
|
||||
nullable=False,
|
||||
)
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
)
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
)
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([recursive]),
|
||||
output_schema=pa.field("size", pa.int64(), nullable=False),
|
||||
)
|
||||
def blob_size(value):
|
||||
return len(value["level_1"]["level_2"]["image"])
|
||||
|
||||
encoded = json.loads(blob_size.registration_request.signature.inputs[0].arrow_type)
|
||||
blob = encoded["fields"][0]["type"]["fields"][0]["type"]["fields"][0]
|
||||
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_(lancedb.blob("item", nullable=False), 2),
|
||||
pa.map_(pa.string(), lancedb.blob("value", nullable=False).type),
|
||||
],
|
||||
)
|
||||
def test_blob_signature_rejects_collection_ancestors(data_type):
|
||||
with pytest.raises(
|
||||
TypeError,
|
||||
match="Blob v2 fields nested under collection types are not supported",
|
||||
):
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([pa.field("value", data_type, nullable=False)]),
|
||||
output_schema=pa.field("size", pa.int64(), nullable=False),
|
||||
)
|
||||
def blob_size(value):
|
||||
return len(value)
|
||||
|
||||
|
||||
def test_blob_signature_rejects_collection_below_a_struct():
|
||||
nested = pa.field(
|
||||
"value",
|
||||
pa.struct(
|
||||
[
|
||||
pa.field(
|
||||
"images",
|
||||
pa.list_(lancedb.blob("item", nullable=False)),
|
||||
nullable=False,
|
||||
)
|
||||
]
|
||||
),
|
||||
nullable=False,
|
||||
)
|
||||
with pytest.raises(
|
||||
TypeError,
|
||||
match="Blob v2 fields nested under collection types are not supported",
|
||||
):
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([nested]),
|
||||
output_schema=pa.field("size", pa.int64(), nullable=False),
|
||||
)
|
||||
def blob_size(value):
|
||||
return len(value["images"])
|
||||
|
||||
|
||||
def test_named_struct_function_can_include_a_blob_result_field():
|
||||
@udf(
|
||||
input_schema=pa.schema([lancedb.blob("image", nullable=False)]),
|
||||
@@ -729,22 +890,6 @@ def test_blob_marker_rejects_invalid_storage_layout():
|
||||
return len(image)
|
||||
|
||||
|
||||
def test_nested_blob_signature_field_has_a_clear_error():
|
||||
nested = pa.field(
|
||||
"value",
|
||||
pa.struct([lancedb.blob("image", nullable=False)]),
|
||||
nullable=False,
|
||||
)
|
||||
with pytest.raises(TypeError, match="nested Blob v2 fields are not supported"):
|
||||
|
||||
@udf(
|
||||
input_schema=pa.schema([nested]),
|
||||
output_schema=pa.field("size", pa.int64(), nullable=False),
|
||||
)
|
||||
def blob_size(value):
|
||||
return len(value["image"])
|
||||
|
||||
|
||||
def test_nested_non_blob_extension_is_not_silently_unwrapped():
|
||||
class TestExtension(pa.ExtensionType):
|
||||
def __init__(self):
|
||||
|
||||
Reference in New Issue
Block a user