fix(python): preserve blob v2 columns on table overwrite

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
         ):

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 = (