fix: support blob v2 in fragment update_columns

python/python/tests/test_dataset.py

@@ -2656,6 +2656,47 @@ def test_merge_insert_subcols(tmp_path: Path):
     assert dataset.to_table().sort_by("a") == expected
 
 
+@pytest.mark.parametrize("container", ["struct", "list"])
+def test_merge_insert_subcols_preserves_nested_blob(tmp_path: Path, container: str):
+    blob_field = lance.blob_field("blob")
+    blob_values = lance.blob_array([b"one", b"two"])
+    if container == "struct":
+        nested_values = pa.StructArray.from_arrays(
+            [blob_values],
+            fields=[blob_field],
+        )
+        expected_nested = [{"blob": b"one"}, {"blob": b"two"}]
+    else:
+        nested_values = pa.ListArray.from_arrays(
+            pa.array([0, 1, 2], type=pa.int32()),
+            blob_values,
+            type=pa.list_(blob_field),
+        )
+        expected_nested = [[b"one"], [b"two"]]
+
+    dataset_uri = tmp_path / f"partial_nested_blob_{container}"
+    dataset = lance.write_dataset(
+        pa.table(
+            {
+                "id": pa.array([1, 2]),
+                "nested": nested_values,
+                "other": pa.array([10, 20]),
+            }
+        ),
+        dataset_uri,
+        data_storage_version="2.2",
+    )
+    source = pa.table({"id": pa.array([2]), "other": pa.array([200])})
+
+    dataset.merge_insert("id").when_matched_update_all().execute(source)
+
+    result = (
+        lance.dataset(dataset_uri).to_table(blob_handling="all_binary").sort_by("id")
+    )
+    assert result["other"].to_pylist() == [10, 200]
+    assert result["nested"].to_pylist() == expected_nested
+
+
 def test_merge_insert_full_fragment_rewrite_json_e2e(tmp_path: Path):
     """End-to-end test: merge_insert with JSON columns where ALL rows are updated.
 

python/python/tests/test_fragment.py

@@ -617,6 +617,139 @@ def test_fragment_update_columns_with_custom_join_key(tmp_path):
     assert result["name"][2] == "Chase"  # id=3 should have name Chase
 
 
+def test_fragment_update_columns_with_blob_v2(tmp_path):
+    data = pa.table(
+        {
+            "id": pa.array([1, 2, 3, 4]),
+            "payload": lance.blob_array([b"one", b"two", b"", None]),
+        }
+    )
+    dataset_uri = tmp_path / "test_dataset_update_columns_blob_v2"
+    dataset = lance.write_dataset(
+        data,
+        dataset_uri,
+        data_storage_version="2.2",
+    )
+
+    fragment = dataset.get_fragment(0)
+    updated_fragment, fields_modified = fragment.update_columns(
+        pa.table(
+            {
+                "id": pa.array([2]),
+                "payload": lance.blob_array([b"NEW"]),
+            }
+        ),
+        left_on="id",
+    )
+
+    operation = LanceOperation.Update(
+        updated_fragments=[updated_fragment],
+        fields_modified=fields_modified,
+    )
+    updated_dataset = LanceDataset.commit(
+        dataset_uri,
+        operation,
+        read_version=dataset.version,
+    )
+
+    result = updated_dataset.to_table(blob_handling="all_binary")
+    assert result["id"].to_pylist() == [1, 2, 3, 4]
+    assert result["payload"].to_pylist() == [b"one", b"NEW", b"", None]
+
+
+def test_fragment_update_columns_with_nested_blob_v2(tmp_path):
+    def info_array(names, payloads):
+        fields = [pa.field("name", pa.string()), lance.blob_field("blob")]
+        return pa.StructArray.from_arrays(
+            [pa.array(names), lance.blob_array(payloads)], fields=fields
+        )
+
+    dataset_uri = tmp_path / "test_dataset_update_columns_nested_blob_v2"
+    dataset = lance.write_dataset(
+        pa.table(
+            {
+                "id": pa.array([1, 2]),
+                "info": info_array(["a", "b"], [b"one", b"two"]),
+            }
+        ),
+        dataset_uri,
+        data_storage_version="2.2",
+    )
+
+    updated_fragment, fields_modified = dataset.get_fragment(0).update_columns(
+        pa.table(
+            {
+                "id": pa.array([2]),
+                "info": info_array(["B"], [b"NEW"]),
+            }
+        ),
+        left_on="id",
+    )
+    updated_dataset = LanceDataset.commit(
+        dataset_uri,
+        LanceOperation.Update(
+            updated_fragments=[updated_fragment],
+            fields_modified=fields_modified,
+        ),
+        read_version=dataset.version,
+    )
+
+    info = updated_dataset.to_table(blob_handling="all_binary")["info"].combine_chunks()
+    assert info.field("name").to_pylist() == ["a", "B"]
+    assert info.field("blob").to_pylist() == [b"one", b"NEW"]
+
+
+def test_fragment_update_columns_preserves_external_blob_v2(tmp_path):
+    dataset_uri = tmp_path / "test_dataset_update_columns_external_blob_v2"
+    external = tmp_path / "existing-payload.bin"
+    external.write_bytes(b"outside")
+    dataset = lance.write_dataset(
+        pa.table(
+            {
+                "id": pa.array([1, 2]),
+                "payload": lance.blob_array([external.as_uri(), b"two"]),
+            }
+        ),
+        dataset_uri,
+        data_storage_version="2.2",
+        allow_external_blob_outside_bases=True,
+    )
+
+    updated_fragment, fields_modified = dataset.get_fragment(0).update_columns(
+        pa.table(
+            {
+                "id": pa.array([2]),
+                "payload": lance.blob_array([b"NEW"]),
+            }
+        ),
+        left_on="id",
+    )
+    updated_dataset = LanceDataset.commit(
+        dataset_uri,
+        LanceOperation.Update(
+            updated_fragments=[updated_fragment],
+            fields_modified=fields_modified,
+        ),
+        read_version=dataset.version,
+    )
+
+    result = updated_dataset.to_table(blob_handling="all_binary")
+    assert result["payload"].to_pylist() == [b"outside", b"NEW"]
+
+    new_external = tmp_path / "new-payload.bin"
+    new_external.write_bytes(b"new outside")
+    with pytest.raises(ValueError, match="outside registered external bases"):
+        updated_dataset.get_fragment(0).update_columns(
+            pa.table(
+                {
+                    "id": pa.array([2]),
+                    "payload": lance.blob_array([new_external.as_uri()]),
+                }
+            ),
+            left_on="id",
+        )
+
+
 def test_fragment_update_columns_with_nulls(tmp_path):
     """Test fragment update columns with null values."""
     # Create initial dataset

python/python/tests/test_optimize.py

@@ -1,5 +1,6 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright The Lance Authors
+import json
 import pickle
 import random
 import re
@@ -88,6 +89,44 @@ def test_blob_compaction(tmp_path: Path):
     assert contents == blobs
 
 
+def test_blob_compaction_with_nested_json_sibling(tmp_path: Path):
+    dataset_uri = tmp_path / "nested_blob_json"
+    info_fields = [lance.blob_field("blob"), pa.field("meta", pa.json_())]
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("info", pa.struct(info_fields)),
+        ]
+    )
+    for index, row_id in enumerate([1, 2]):
+        info = pa.StructArray.from_arrays(
+            [
+                lance.blob_array([f"blob-{row_id}".encode()]),
+                pa.array([json.dumps({"row": row_id})], type=pa.json_()),
+            ],
+            fields=info_fields,
+        )
+        lance.write_dataset(
+            pa.Table.from_arrays([pa.array([row_id]), info], schema=schema),
+            dataset_uri,
+            mode="create" if index == 0 else "append",
+            data_storage_version="2.2",
+        )
+
+    dataset = lance.dataset(dataset_uri)
+    dataset.optimize.compact_files(num_threads=1)
+
+    assert len(dataset.get_fragments()) == 1
+    assert [data for _, data in dataset.read_blobs("info.blob", indices=[0, 1])] == [
+        b"blob-1",
+        b"blob-2",
+    ]
+    assert [
+        json.loads(value)
+        for value in dataset.to_table(columns=["info.meta"])["info.meta"].to_pylist()
+    ] == [{"row": 1}, {"row": 2}]
+
+
 @pytest.mark.parametrize("storage_version", ["2.0", "2.1", "2.2"])
 def test_blob_compaction_preserves_null_empty_and_read_parity(
     tmp_path: Path, storage_version: str