diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index f4e835547..2bdd57fe9 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -261,6 +261,101 @@ IndexConfigType = Union[ KNOWN_METRICS = {"l2", "cosine", "dot", "hamming"} +def _blob_value_to_storage(value: Any) -> Optional[dict]: + """Keep a Python blob's inline data or external URI before Arrow infers it.""" + if value is None: + return None + if isinstance(value, (bytes, bytearray, memoryview)): + return {"data": bytes(value)} + if isinstance(value, str): + if not value: + raise ValueError("Blob uri cannot be empty") + return {"uri": value} + if isinstance(value, dict): + unknown = value.keys() - {"data", "uri", "position", "size"} + if unknown: + raise ValueError(f"Unknown blob fields: {sorted(unknown)}") + return value + + try: + from lance.blob import Blob + except ModuleNotFoundError as err: + if err.name not in ("lance", "lance.blob"): + raise + else: + if isinstance(value, Blob): + return { + "data": value.data, + "uri": value.uri, + "position": value.position, + "size": value.size, + } + raise TypeError(f"Unsupported blob value: {type(value).__name__}") + + +def _blob_input_to_arrow(data: Any, schema: Optional[pa.Schema]) -> Optional[pa.Table]: + """Convert Python rows with blob fields before Arrow loses their value types.""" + if schema is None: + return None + + if isinstance(data, list) and data and isinstance(data[0], dict): + names = { + field.name + for field in schema + if is_blob_v2_field(field) + and any(isinstance(row, dict) and field.name in row for row in data) + } + if not names: + return None + values = {name: [row.get(name) for row in data] for name in names} + rows = [{**row, **{name: None for name in names}} for row in data] + table = pa.Table.from_pylist(rows) + elif _check_for_pandas(data) and isinstance(data, pd.DataFrame): + names = { + field.name + for field in schema + if is_blob_v2_field(field) and field.name in data.columns + } + if not names: + return None + values = { + name: [ + None + if value is pd.NA or isinstance(value, float) and np.isnan(value) + else value + for value in data[name] + ] + for name in names + } + table = pa.Table.from_pandas( + data.assign(**{name: None for name in names}), preserve_index=False + ).replace_schema_metadata(None) + else: + return None + + for field in schema: + if field.name not in names: + continue + storage_type = ( + field.type.storage_type + if isinstance(field.type, pa.ExtensionType) + else field.type + ) + storage = pa.array( + [_blob_value_to_storage(value) for value in values[field.name]], + type=storage_type, + ) + column = ( + pa.ExtensionArray.from_storage(field.type, storage) + if isinstance(field.type, pa.ExtensionType) + else storage + ) + table = table.set_column( + table.schema.get_field_index(field.name), field, column + ) + return table + + def _into_pyarrow_reader( data, schema: Optional[pa.Schema] = None ) -> pa.RecordBatchReader: @@ -303,9 +398,14 @@ def _into_pyarrow_reader( return pa.Table.from_batches(data).to_reader() else: data = _serialize_json_values(data, schema) - return pa.Table.from_pylist(data).to_reader() + table = _blob_input_to_arrow(data, schema) + return ( + table if table is not None else pa.Table.from_pylist(data) + ).to_reader() elif _check_for_pandas(data) and isinstance(data, pd.DataFrame): - table = pa.Table.from_pandas(data, preserve_index=False) + table = _blob_input_to_arrow(data, schema) + if table is None: + table = pa.Table.from_pandas(data, preserve_index=False) # Do not serialize Pandas metadata meta = table.schema.metadata if table.schema.metadata is not None else {} meta = {k: v for k, v in meta.items() if k != b"pandas"} @@ -5891,6 +5991,9 @@ class AsyncTable: """ schema = await self.schema() data = _serialize_json_values(data, schema) + blob_table = _blob_input_to_arrow(data, schema) + if blob_table is not None: + data = blob_table if on_bad_vectors is None: on_bad_vectors = "error" if fill_value is None: diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py index cada6e83e..e100597cf 100644 --- a/python/python/tests/test_blob.py +++ b/python/python/tests/test_blob.py @@ -7,10 +7,11 @@ import sys import textwrap import lance +import pandas as pd import pyarrow as pa import pyarrow.compute as pc import pytest -from lance.blob import BlobType as LanceBlobType +from lance.blob import Blob, BlobType as LanceBlobType import lancedb from lancedb._blob import ( @@ -1432,3 +1433,129 @@ def test_add_external_uri_string_round_trips_with_flag(tmp_path): hits = table.search().to_arrow() blobs = table.fetch_blobs("image", hits) assert blobs[0].as_py() == payload + + +@pytest.mark.parametrize("as_pandas", [False, True]) +def test_add_bytes_and_uri_in_one_batch_preserves_external_uri(tmp_path, as_pandas): + payload = b"external-payload" + blob_path = tmp_path / "payload.bin" + blob_path.write_bytes(payload) + rows = [ + {"id": 1, "image": b"inline"}, + {"id": 2, "image": blob_path.as_uri()}, + ] + db = lancedb.connect(tmp_path / "db") + schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")]) + table = db.create_table("mixed", schema=schema) + + table.add( + pd.DataFrame(rows) if as_pandas else rows, + allow_external_blob_outside_bases=True, + ) + + by_id = _row_ids_by_id(table) + assert table.fetch_blobs("image", [by_id[1], by_id[2]]).to_pylist() == [ + b"inline", + payload, + ] + + +@pytest.mark.parametrize("as_pandas", [False, True]) +def test_add_mixed_python_blob_values_preserves_external_uris(tmp_path, as_pandas): + payload = b"external-payload" + blob_path = tmp_path / "payload.bin" + blob_path.write_bytes(payload) + uri = blob_path.as_uri() + rows = [ + {"id": 1, "image": b"inline"}, + {"id": 2, "image": uri}, + {"id": 3, "image": {"data": b"from-dict"}}, + {"id": 4, "image": Blob.from_bytes(b"from-blob")}, + {"id": 5, "image": Blob.from_uri(uri)}, + {"id": 6, "image": {"uri": uri}}, + {"id": 7, "image": None}, + ] + db = lancedb.connect(tmp_path / "db") + schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")]) + table = db.create_table("mixed", schema=schema) + + table.add( + pd.DataFrame(rows) if as_pandas else rows, + allow_external_blob_outside_bases=True, + ) + + by_id = _row_ids_by_id(table) + assert table.fetch_blobs("image", [by_id[i] for i in range(1, 8)]).to_pylist() == [ + b"inline", + payload, + b"from-dict", + b"from-blob", + payload, + payload, + None, + ] + + +@pytest.mark.parametrize("as_pandas", [False, True]) +def test_merge_insert_mixed_python_blobs_validates_external_uri(tmp_path, as_pandas): + blob_path = tmp_path / "payload.bin" + blob_path.write_bytes(b"external-payload") + table = _blob_table("merge_mixed_uri", [{"id": 1, "image": b"before"}]) + rows = [ + {"id": 1, "image": b"updated"}, + {"id": 2, "image": blob_path.as_uri()}, + ] + + with pytest.raises(ValueError, match="allow_external_blob_outside_bases"): + ( + table.merge_insert("id") + .when_matched_update_all() + .when_not_matched_insert_all() + .execute(pd.DataFrame(rows) if as_pandas else rows) + ) + assert table.count_rows() == 1 + + +@pytest.mark.parametrize("as_pandas", [False, True]) +def test_merge_insert_python_blob_dicts_and_objects(as_pandas): + table = _blob_table("merge_python_blobs", [{"id": 1, "image": b"before"}]) + rows = [ + {"id": 1, "image": b"updated"}, + {"id": 2, "image": {"data": b"from-dict"}}, + {"id": 3, "image": Blob.from_bytes(b"from-blob")}, + ] + + result = ( + table.merge_insert("id") + .when_matched_update_all() + .when_not_matched_insert_all() + .execute(pd.DataFrame(rows) if as_pandas else rows) + ) + + assert result.num_updated_rows == 1 + assert result.num_inserted_rows == 2 + by_id = _row_ids_by_id(table) + assert table.fetch_blobs("image", [by_id[i] for i in range(1, 4)]).to_pylist() == [ + b"updated", + b"from-dict", + b"from-blob", + ] + + +def test_create_table_with_python_blob_dict_and_object(): + db = lancedb.connect("memory:///") + schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")]) + table = db.create_table( + "create_python_blobs", + data=[ + {"id": 1, "image": {"data": b"from-dict"}}, + {"id": 2, "image": Blob.from_bytes(b"from-blob")}, + ], + schema=schema, + ) + + by_id = _row_ids_by_id(table) + assert table.fetch_blobs("image", [by_id[1], by_id[2]]).to_pylist() == [ + b"from-dict", + b"from-blob", + ]