diff --git a/docs/web/curation/create-function.mdx b/docs/web/curation/create-function.mdx index 8e4bdc657..1a35e31be 100644 --- a/docs/web/curation/create-function.mdx +++ b/docs/web/curation/create-function.mdx @@ -217,6 +217,10 @@ schema still holds the right bytes, but skips the Managed storage path and `fetch_blobs`. Declare it explicitly on every function whose output should be a blob. +A Blob v2 field may also be an element of an output list or large list. These +values are queryable as Blob descriptors, but the row-id blob fetch APIs and +`blob_mode="bytes"` cannot fetch payloads inside lists. + ## What to read next {#next} diff --git a/docs/web/management/function.mdx b/docs/web/management/function.mdx index e110612ea..7a16f8530 100644 --- a/docs/web/management/function.mdx +++ b/docs/web/management/function.mdx @@ -313,6 +313,11 @@ blob as a value, use a named-struct output schema with a nullable blob field; the struct itself records whether the result has been filled. Read a nested blob with `table.fetch_blobs("payload.image", hits)`. +A list or large-list output may also contain Blob v2 elements; inputs and +fixed-size lists do not support this layout. List-nested values are queryable +as descriptors, but `fetch_blobs` and `blob_mode="bytes"` cannot fetch their +payloads. + `output_schema=lancedb.blob(...)` is what makes a persisted result Blob v2 rather than plain Arrow `binary`. Returning `bytes` from a function whose diff --git a/python/python/lancedb/functions.py b/python/python/lancedb/functions.py index a17fcfa94..c77f45e5e 100644 --- a/python/python/lancedb/functions.py +++ b/python/python/lancedb/functions.py @@ -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: diff --git a/python/python/tests/test_first_class_function_slice2.py b/python/python/tests/test_first_class_function_slice2.py index b17e3208b..4478fd3e5 100644 --- a/python/python/tests/test_first_class_function_slice2.py +++ b/python/python/tests/test_first_class_function_slice2.py @@ -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.""" diff --git a/rust/lancedb/src/table/computed_columns.rs b/rust/lancedb/src/table/computed_columns.rs index 9a99f3c04..0e02e7ce2 100644 --- a/rust/lancedb/src/table/computed_columns.rs +++ b/rust/lancedb/src/table/computed_columns.rs @@ -1039,7 +1039,7 @@ fn canonical_input_arrow_type(field: &JsonArrowField) -> Result { if is_blob_v2 || field.r#type.fields.is_some() { let arrow_field = lance_namespace::schema::convert_json_arrow_field(field) .map_err(|e| invalid_function(format!("invalid Function input field: {e}")))?; - validate_function_blob_nesting(&arrow_field, false)?; + validate_function_blob_nesting(&arrow_field, false, false)?; if is_blob_v2 { return Ok(FUNCTION_BLOB_V2_TYPE.to_string()); } @@ -1061,11 +1061,16 @@ fn has_supported_blob_v2_layout(field: &ArrowField) -> bool { ) } -fn validate_function_blob_nesting(field: &ArrowField, inside_collection: bool) -> Result<()> { +/// Blob v2 under a collection is accepted only where `collection_blobs` is set (outputs). +fn validate_function_blob_nesting( + field: &ArrowField, + inside_collection: bool, + collection_blobs: bool, +) -> Result<()> { if field.is_blob_v2() { - if inside_collection { + if inside_collection && !collection_blobs { return Err(invalid_function(format!( - "Function field '{}' nests Blob v2 under a collection, which Function signatures do not support", + "Function field '{}' nests Blob v2 under a collection; only a Function output may, under a list or large_list", field.name() ))); } @@ -1078,13 +1083,16 @@ fn validate_function_blob_nesting(field: &ArrowField, inside_collection: bool) - return Ok(()); } match field.data_type() { - DataType::Struct(fields) => fields - .iter() - .try_for_each(|field| validate_function_blob_nesting(field, inside_collection)), - DataType::List(field) - | DataType::LargeList(field) - | DataType::FixedSizeList(field, _) - | DataType::Map(field, _) => validate_function_blob_nesting(field, true), + DataType::Struct(fields) => fields.iter().try_for_each(|field| { + validate_function_blob_nesting(field, inside_collection, collection_blobs) + }), + DataType::List(field) | DataType::LargeList(field) => { + validate_function_blob_nesting(field, true, collection_blobs) + } + // Lance stores Blob v2 under List and LargeList only. + DataType::FixedSizeList(field, _) | DataType::Map(field, _) => { + validate_function_blob_nesting(field, true, false) + } _ => Ok(()), } } @@ -1183,7 +1191,7 @@ fn function_output_field(name: &str, nullable: bool, raw: &str) -> Result item, + other => panic!("expected a list output, got {other}"), + }; + assert!(item.is_blob_v2()); + } + + let rejected = [ + DataType::FixedSizeList(Arc::new(crate::blob("item", true)), 2), + DataType::Map( + Arc::new(ArrowField::new( + "entries", + DataType::Struct(Fields::from(vec![ + ArrowField::new("key", DataType::Utf8, false), + crate::blob("value", true), + ])), + false, + )), + false, + ), + ]; + for data_type in rejected { + let output_type = exact_arrow_type(ArrowField::new("frames", data_type, false)); + let output = serde_json::json!({ + "kind": "scalar", + "arrow_type": output_type, + "nullable": false + }); + let application = blob_application(&output.to_string()); + let error = plan_function_application( + &ArrowSchema::new(vec![input.clone()]), + &application, + Some("frames"), + ) + .unwrap_err(); + assert!(error.to_string().contains("nests Blob v2"), "{error}"); + } + } + #[test] fn test_blob_children_under_collections_are_rejected() { let collections = vec![