From 8219e1d80b34f66dd33197d8f45790d4d100fe79 Mon Sep 17 00:00:00 2001 From: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com> Date: Wed, 23 Sep 2026 18:26:52 +0000 Subject: [PATCH] fix(python): preserve blob v2 columns on table overwrite --- python/python/lancedb/table.py | 63 ++++++++++++++++++++++++++++-- python/python/tests/test_blob.py | 67 ++++++++++++++++++++++++++++++++ 2 files changed, 127 insertions(+), 3 deletions(-) diff --git a/python/python/lancedb/table.py b/python/python/lancedb/table.py index 297b4237f..3ca0a330c 100644 --- a/python/python/lancedb/table.py +++ b/python/python/lancedb/table.py @@ -448,6 +448,55 @@ def _cast_to_target_schema( return pa.RecordBatchReader.from_batches(reordered_schema, gen()) +def _preserve_blob_v2_fields( + inferred_schema: pa.Schema, existing_schema: pa.Schema +) -> pa.Schema: + """Keep existing blob fields when an overwrite infers a replacement schema.""" + existing_fields = {field.name: field for field in existing_schema} + fields = [ + _preserve_blob_v2_field(field, existing_fields[field.name]) + if field.name in existing_fields + else field + for field in inferred_schema + ] + return pa.schema(fields, metadata=inferred_schema.metadata) + + +def _preserve_blob_v2_field(inferred: pa.Field, existing: pa.Field) -> pa.Field: + if is_blob_v2_field(existing): + return existing + + inferred_type, existing_type = inferred.type, existing.type + if pa.types.is_struct(inferred_type) and pa.types.is_struct(existing_type): + existing_children = {field.name: field for field in existing_type} + children = [ + _preserve_blob_v2_field(field, existing_children[field.name]) + if field.name in existing_children + else field + for field in inferred_type + ] + return inferred.with_type(pa.struct(children)) + + if _is_list_like(inferred_type) and _is_list_like(existing_type): + value_field = _preserve_blob_v2_field( + inferred_type.value_field, existing_type.value_field + ) + if value_field == inferred_type.value_field: + return inferred + value_field = value_field.with_name(existing_type.value_field.name) + if pa.types.is_fixed_size_list(inferred_type): + value_type = pa.list_(value_field, inferred_type.list_size) + elif pa.types.is_large_list(inferred_type): + value_type = pa.large_list(value_field) + else: + value_type = pa.list_(value_field) + return inferred.with_type(value_type) + + if blob_v2_column_paths(pa.schema([existing])): + return existing + return inferred + + def _coerce_blob_write_columns( batch: pa.RecordBatch, target_schema: pa.Schema ) -> pa.RecordBatch: @@ -1604,7 +1653,8 @@ class Table(ABC): - pyarrow.Table or pyarrow.RecordBatch mode: str The mode to use when writing the data. Valid values are - "append" and "overwrite". + "append" and "overwrite". Overwrite infers a replacement schema + but preserves existing blob v2 fields present in the new data. on_bad_vectors: str, default "error" What to do if any of the vectors are not the same size or contains NaNs. One of "error", "drop", "fill". @@ -3821,7 +3871,8 @@ class LanceTable(Table): The data to insert into the table. mode: str The mode to use when writing the data. Valid values are - "append" and "overwrite". + "append" and "overwrite". Overwrite infers a replacement schema + but preserves existing blob v2 fields present in the new data. on_bad_vectors: str, default "error" What to do if any of the vectors are not the same size or contains NaNs. One of "error", "drop", "fill", "null". @@ -5804,7 +5855,8 @@ class AsyncTable: - pyarrow.Table or pyarrow.RecordBatch mode: str The mode to use when writing the data. Valid values are - "append" and "overwrite". + "append" and "overwrite". Overwrite infers a replacement schema + but preserves existing blob v2 fields present in the new data. on_bad_vectors: str, default "error" What to do if any of the vectors are not the same size or contains NaNs. One of "error", "drop", "fill", "null". @@ -5839,6 +5891,11 @@ class AsyncTable: data, _ = sanitize_create_table( data, None, on_bad_vectors=on_bad_vectors, fill_value=fill_value ) + if blob_v2_column_paths(schema): + # Plain bytes cannot be inferred as blob v2. Keep matching blob + # fields while allowing other columns to use the inferred schema. + target_schema = _preserve_blob_v2_fields(data.schema, schema) + data = _cast_to_target_schema(data, target_schema) elif on_bad_vectors != "error" or ( schema.metadata is not None and b"embedding_functions" in schema.metadata ): diff --git a/python/python/tests/test_blob.py b/python/python/tests/test_blob.py index 0d3a89c21..04b0b6f60 100644 --- a/python/python/tests/test_blob.py +++ b/python/python/tests/test_blob.py @@ -419,6 +419,73 @@ def test_fetch_blobs_round_trip(): assert [blobs[0].as_py(), blobs[1].as_py()] == [b"alpha", b"beta"] +def test_add_overwrite_preserves_blob_v2(tmp_path): + db = lancedb.connect(tmp_path) + schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")]) + table = db.create_table("overwrite_blob", schema=schema) + table.add([{"id": 1, "image": b"old"}]) + + payload = b"x" * 200_000 + table.add([{"id": 2, "image": payload}], mode="overwrite") + + assert table.count_rows() == 1 + assert blob_v2_column_paths(table.schema) == ["image"] + assert table.blob_columns() == ["image"] + assert table.fetch_blobs("image", [_row_ids_by_id(table)[2]]).to_pylist() == [ + payload + ] + assert list(tmp_path.rglob("*.blob")) + + +def test_add_overwrite_with_blob_still_infers_vector_schema(): + db = lancedb.connect("memory:///overwrite_blob_vector") + schema = pa.schema( + [ + pa.field("vector", pa.list_(pa.float32(), 2)), + lancedb.blob("image"), + ] + ) + table = db.create_table("overwrite_blob_vector", schema=schema) + + table.add([{"vector": [1.0, 2.0, 3.0], "image": b"data"}], mode="overwrite") + + assert table.schema.field("vector").type == pa.list_(pa.float32(), 3) + assert table.blob_columns() == ["image"] + + +@pytest.mark.asyncio +async def test_add_overwrite_preserves_blob_v2_async(): + db = await lancedb.connect_async("memory:///overwrite_blob_async") + schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")]) + table = await db.create_table("overwrite_blob_async", schema=schema) + + await table.add([{"id": 1, "image": b"updated"}], mode="overwrite") + + assert blob_v2_column_paths(await table.schema()) == ["image"] + assert await table.blob_columns() == ["image"] + hits = await table.query().with_row_id().to_arrow() + assert (await table.fetch_blobs("image", hits)).to_pylist() == [b"updated"] + + +def test_add_overwrite_preserves_nested_blob_v2(): + db = lancedb.connect("memory:///overwrite_nested_blob") + schema = pa.schema( + [ + pa.field("id", pa.int64()), + pa.field("info", pa.struct([lancedb.blob("image")])), + ] + ) + table = db.create_table("overwrite_nested_blob", schema=schema) + + table.add([{"id": 1, "info": {"image": b"nested"}}], mode="overwrite") + + assert blob_v2_column_paths(table.schema) == ["info.image"] + assert table.blob_columns() == ["info.image"] + assert table.fetch_blobs("info.image", [_row_ids_by_id(table)[1]]).to_pylist() == [ + b"nested" + ] + + def test_merge_insert_writes_python_bytes(): table = _blob_table("merge_bytes", [{"id": 1, "image": b"before"}]) result = (