fix: accept all-null batches and plain JSON strings for json columns (#4067)

Two ways of writing to a `json` column failed or silently corrupted
data.

**All-null batches were rejected.** `add()` refused a batch whose values
for a `json` column were all null, while every plain Arrow type accepted
the same batch. This bites row-at-a-time inserts hardest: a one-row
batch with no value for an optional column is trivially all-null, so
most such writes failed. pyarrow infers `null` as the column's type, and
the write path had no handling for it — casting to the table's type
dropped the field metadata that identifies the column as `lance.json`,
so lance rejected the batch (`` `val` should have type json but type was
large_binary ``). A null-typed input column now becomes typed nulls
matching the table's field exactly, metadata included.

**Unlabelled JSON text was stored raw.** JSON supplied as plain strings
(what pyarrow infers for a column of `str`) was cast to the column's
`LargeBinary` storage type and relabelled `lance.json`, putting unparsed
text where JSONB was expected. Reads returned the text unnormalized and
`json_extract` failed with `InvalidJsonb`. Lance-core does the JSONB
encoding, but only for input labelled `arrow.json`, so string input is
now labelled rather than cast — at the top level and inside structs.

Both fixes are in the shared Rust write path, so they apply to any
binding, including hand-built Arrow tables that never pass through
Python's list-of-dicts type inference. `_align_field` gets the same
JSON-string fix for the legacy Python `_sanitize_data` path, which
`on_bad_vectors` and embedding functions still route through.

The blob v2 half of the issue landed separately in #4065, which added a
`DataType::Null` arm to blob coercion. This PR keeps that implementation
and adds end-to-end add-path coverage for it.

The tests from #4066 are included here and pass, so that PR's
Python-layer inference changes are no longer needed to close the issue.

Fixes #3759

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

python/python/lancedb/table.py

@@ -634,26 +634,43 @@ def _align_field_types(
     return new_fields
 
 
-def _align_list_value_field(
-    value_field: pa.Field, target_value_field: pa.Field
-) -> pa.Field:
-    # A list has exactly one child, so the inferred child name ("item") aligns
-    # positionally and adopts the table's child name; pa.Table.cast renames it.
-    return _align_field(value_field, target_value_field).with_name(
-        target_value_field.name
-    )
+def _align_container_child(child: pa.Field, target_child: pa.Field) -> pa.Field:
+    # A list has one child, a map one key and one item, so an inferred child name
+    # ("item") aligns positionally and adopts the table's; pa.Table.cast renames it.
+    return _align_field(child, target_child).with_name(target_child.name)
+
+
+def _arrow_json_storage_type(input_type: pa.DataType) -> Optional[pa.DataType]:
+    """The storage type arrow.json would use for ``input_type``.
+
+    Returns None if the type cannot hold JSON text.
+    """
+    if pa.types.is_string(input_type) or pa.types.is_string_view(input_type):
+        return pa.string()
+    if pa.types.is_large_string(input_type):
+        return pa.large_string()
+    return None
 
 
 def _align_field(field: pa.Field, target_field: pa.Field) -> pa.Field:
-    # Preserve arrow.json input until it reaches Lance. LanceDB exposes stored
-    # JSON columns as lance.json (JSONB-backed LargeBinary), but casting the
-    # input to that storage type here merely relabels the raw JSON bytes as
+    # LanceDB exposes stored JSON columns as lance.json (JSONB-backed LargeBinary), but
+    # casting the input to that storage type here merely relabels the raw JSON bytes as
     # JSONB. Lance must see arrow.json so it can perform the JSONB encoding.
-    if (
-        _field_extension_name(field) == "arrow.json"
-        and _field_extension_name(target_field) == "lance.json"
-    ):
-        return field
+    if _field_extension_name(target_field) == "lance.json":
+        if _field_extension_name(field) == "arrow.json":
+            return field
+        # Plain JSON text, which is what pyarrow infers for a column of `str`, only
+        # needs the arrow.json label.
+        json_storage = _arrow_json_storage_type(field.type)
+        if json_storage is not None:
+            # Labelled through metadata rather than pa.json_(), which only exists on
+            # newer PyArrow; Lance reads the extension name off the field either way.
+            return pa.field(
+                field.name,
+                json_storage,
+                field.nullable,
+                {"ARROW:extension:name": "arrow.json"},
+            )
     if pa.types.is_struct(target_field.type):
         if pa.types.is_struct(field.type):
             new_type = pa.struct(
@@ -667,7 +684,7 @@ def _align_field(field: pa.Field, target_field: pa.Field) -> pa.Field:
     elif pa.types.is_list(target_field.type):
         if _is_list_like(field.type):
             new_type = pa.list_(
-                _align_list_value_field(
+                _align_container_child(
                     field.type.value_field, target_field.type.value_field
                 )
             )
@@ -676,7 +693,7 @@ def _align_field(field: pa.Field, target_field: pa.Field) -> pa.Field:
     elif pa.types.is_large_list(target_field.type):
         if _is_list_like(field.type):
             new_type = pa.large_list(
-                _align_list_value_field(
+                _align_container_child(
                     field.type.value_field, target_field.type.value_field
                 )
             )
@@ -685,13 +702,28 @@ def _align_field(field: pa.Field, target_field: pa.Field) -> pa.Field:
     elif pa.types.is_fixed_size_list(target_field.type):
         if _is_list_like(field.type):
             new_type = pa.list_(
-                _align_list_value_field(
+                _align_container_child(
                     field.type.value_field, target_field.type.value_field
                 ),
                 target_field.type.list_size,
             )
         else:
             new_type = target_field.type
+    elif pa.types.is_map(target_field.type):
+        if pa.types.is_map(field.type):
+            # A map has exactly one key and one item field, so like a list's child they
+            # align positionally and adopt the table's names.
+            new_type = pa.map_(
+                _align_container_child(
+                    field.type.key_field, target_field.type.key_field
+                ),
+                _align_container_child(
+                    field.type.item_field, target_field.type.item_field
+                ),
+                keys_sorted=target_field.type.keys_sorted,
+            )
+        else:
+            new_type = target_field.type
     else:
         new_type = target_field.type
     return pa.field(field.name, new_type, field.nullable, target_field.metadata)

python/python/tests/test_blob.py

@@ -751,6 +751,80 @@ def test_fetch_blobs_preserves_null_and_empty_values():
     assert blobs[3].as_py() == b"present"
 
 
+def test_add_all_null_list_to_blob_column():
+    table = _blob_table("all_null_add", [{"id": 1, "image": None}])
+
+    hits = table.search().to_arrow()
+    blobs = table.fetch_blobs("image", hits)
+    assert len(blobs) == 1
+    assert blobs[0].as_py() is None
+
+
+def test_add_all_null_list_to_blob_column_with_sanitizer():
+    db = lancedb.connect("memory:///")
+    schema = pa.schema([pa.field("id", pa.int64()), lancedb.blob("image")])
+    table = db.create_table("all_null_sanitized_add", schema=schema)
+
+    table.add([{"id": 1, "image": None}], on_bad_vectors="fill")
+
+    hits = table.search().to_arrow()
+    blobs = table.fetch_blobs("image", hits)
+    assert len(blobs) == 1
+    assert blobs[0].as_py() is None
+
+
+def test_add_all_null_list_to_nested_blob_column():
+    db = lancedb.connect("memory:///")
+    blob_field = lancedb.blob("image")
+    info_field = pa.field("info", pa.struct([blob_field]))
+    info = pa.StructArray.from_arrays(
+        [_blob_array("image", [b"seed"])], fields=[blob_field]
+    )
+    seed = pa.Table.from_arrays(
+        [pa.array([0], type=pa.int64()), info],
+        schema=pa.schema([pa.field("id", pa.int64()), info_field]),
+    )
+    table = db.create_table("nested_null_add", data=seed)
+
+    table.add([{"id": 1, "info": {"image": None}}])
+    table.add([{"id": 2, "info": {"image": None}}], on_bad_vectors="fill")
+
+    hits = table.search().where("id > 0").to_arrow()
+    blobs = table.fetch_blobs("info.image", hits)
+    assert len(blobs) == 2
+    assert all(blob.as_py() is None for blob in blobs)
+
+
+@pytest.mark.parametrize("large_list", [False, True], ids=["list", "large_list"])
+def test_add_list_of_dicts_to_blob_list_column(large_list):
+    db = lancedb.connect("memory:///")
+    blob_field = lancedb.blob("image")
+    blob_values = _blob_array("image", [b"seed"])
+    if large_list:
+        items_field = pa.field("items", pa.large_list(blob_field))
+        items = pa.LargeListArray.from_arrays(
+            pa.array([0, 1], type=pa.int64()), blob_values
+        )
+    else:
+        items_field = pa.field("items", pa.list_(blob_field))
+        items = pa.ListArray.from_arrays(pa.array([0, 1], type=pa.int32()), blob_values)
+    seed = pa.Table.from_arrays(
+        [pa.array([0], type=pa.int64()), items],
+        schema=pa.schema([pa.field("id", pa.int64()), items_field]),
+    )
+    table = db.create_table(f"blob_{large_list}_list_add", data=seed)
+
+    table.add([{"id": 1, "items": [None]}])
+    table.add(
+        [{"id": 2, "items": [b"a", None]}],
+        on_bad_vectors="fill",
+    )
+
+    ids = table.search().select(["id"]).to_arrow()["id"].to_pylist()
+    assert sorted(ids) == [0, 1, 2]
+    assert pa.types.is_large_list(table.schema.field("items").type) is large_list
+
+
 def test_fetch_blob_ranges_aligns_repeated_ranges_and_nulls():
     table = _blob_table(
         "range_alignment",

python/python/tests/test_table.py

@@ -2979,28 +2979,29 @@ async def test_merge_insert_async(mem_db_async: AsyncConnection):
     assert (await table.to_arrow()).sort_by("a") == expected
 
 
+def _json_arrow_table(schema, rows):
+    json_type = schema.field("j").type
+    json_values = pa.ExtensionArray.from_storage(
+        json_type,
+        pa.array([value for _, value in rows], type=json_type.storage_type),
+    )
+    return pa.Table.from_arrays(
+        [pa.array([row_id for row_id, _ in rows]), json_values], schema=schema
+    )
+
+
 @pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
 @pytest.mark.asyncio
 async def test_merge_insert_encodes_json(mem_db_async: AsyncConnection):
-    json_type = pa.json_()
-    schema = pa.schema([pa.field("id", pa.string()), pa.field("j", json_type)])
-
-    def json_table(rows):
-        json_values = pa.ExtensionArray.from_storage(
-            json_type,
-            pa.array([value for _, value in rows], type=json_type.storage_type),
-        )
-        return pa.Table.from_arrays(
-            [pa.array([row_id for row_id, _ in rows]), json_values], schema=schema
-        )
+    schema = pa.schema([pa.field("id", pa.string()), pa.field("j", pa.json_())])
 
     table = await mem_db_async.create_table("json_merge", schema=schema)
-    await table.add(json_table([("a", '{"k": 1}'), ("b", '{"k": 9}')]))
+    await table.add(_json_arrow_table(schema, [("a", '{"k": 1}'), ("b", '{"k": 9}')]))
 
     await (
         table.merge_insert("id")
         .when_matched_update_all()
-        .execute(json_table([("a", '{"k": 2}')]))
+        .execute(_json_arrow_table(schema, [("a", '{"k": 2}')]))
     )
 
     rows = sorted(await table.query().to_list(), key=lambda row: row["id"])
@@ -3015,20 +3016,176 @@ async def test_merge_insert_encodes_json(mem_db_async: AsyncConnection):
 @pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
 @pytest.mark.asyncio
 async def test_add_sanitization_encodes_json(mem_db_async: AsyncConnection):
-    json_type = pa.json_()
-    schema = pa.schema([pa.field("id", pa.string()), pa.field("j", json_type)])
-    json_values = pa.ExtensionArray.from_storage(
-        json_type, pa.array(['{"k": 3}'], type=json_type.storage_type)
-    )
-    data = pa.Table.from_arrays([pa.array(["c"]), json_values], schema=schema)
+    schema = pa.schema([pa.field("id", pa.string()), pa.field("j", pa.json_())])
 
     table = await mem_db_async.create_table("json_add", schema=schema)
-    await table.add(data, on_bad_vectors="fill")
+    await table.add(
+        _json_arrow_table(schema, [("c", '{"k": 3}')]), on_bad_vectors="fill"
+    )
 
     rows = await table.query().where("json_extract(j, '$.k') = '3'").to_list()
     assert rows == [{"id": "c", "j": '{"k":3}'}]
 
 
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+@pytest.mark.asyncio
+async def test_add_all_null_json_batch(mem_db_async: AsyncConnection):
+    """A batch of dicts whose json values are all None infers as pa.null(), which used
+    to fail with a `json` vs `large_binary` schema mismatch. A row-at-a-time insert of
+    an optional json column always looks like this."""
+    schema = pa.schema([pa.field("id", pa.string()), pa.field("j", pa.json_())])
+    table = await mem_db_async.create_table("json_nulls", schema=schema)
+
+    await table.add([{"id": "a", "j": None}])
+    assert await table.count_rows() == 1
+
+    # ... and again once real JSON has been written.
+    await table.add(_json_arrow_table(schema, [("b", '{"k": 9}')]))
+    await table.add([{"id": "c", "j": None}])
+
+    rows = sorted(await table.query().to_list(), key=lambda row: row["id"])
+    assert rows == [
+        {"id": "a", "j": None},
+        {"id": "b", "j": '{"k":9}'},
+        {"id": "c", "j": None},
+    ]
+
+    # The nulls must not disturb reads of the column.
+    filtered = await table.query().where("json_extract(j, '$.k') = '9'").to_list()
+    assert filtered == [{"id": "b", "j": '{"k":9}'}]
+    assert len(await table.query().where("j IS NULL").to_list()) == 2
+
+
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+def test_add_all_null_json_batch_sync(mem_db: DBConnection):
+    schema = pa.schema([pa.field("id", pa.string()), pa.field("j", pa.json_())])
+    table = mem_db.create_table("json_nulls_sync", schema=schema)
+
+    table.add([{"id": "a", "j": None}])
+
+    assert table.count_rows() == 1
+    assert table.to_arrow()["j"].to_pylist() == [None]
+
+
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("values", "expected"),
+    [
+        ([None], [None]),
+        ([None, '{"k": 1}'], [None, '{"k":1}']),
+        (['{"k": 2}'], ['{"k":2}']),
+    ],
+)
+async def test_add_list_of_dicts_to_json_column(
+    mem_db_async: AsyncConnection, values, expected
+):
+    schema = pa.schema([pa.field("id", pa.int64()), pa.field("value", pa.json_())])
+    table = await mem_db_async.create_table("json_list_add", schema=schema)
+
+    await table.add([{"id": idx, "value": value} for idx, value in enumerate(values)])
+
+    rows = (await table.to_arrow()).sort_by("id").to_pylist()
+    assert [row["value"] for row in rows] == expected
+
+
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+@pytest.mark.asyncio
+async def test_add_list_of_dicts_to_nested_json_column(
+    mem_db_async: AsyncConnection,
+):
+    json_field = pa.field("value", pa.json_())
+    info_field = pa.field("info", pa.struct([json_field]))
+    info = pa.StructArray.from_arrays(
+        [pa.array(['{"seed": 0}'], type=pa.json_())], fields=[json_field]
+    )
+    seed = pa.Table.from_arrays(
+        [pa.array([0], type=pa.int64()), info],
+        schema=pa.schema([pa.field("id", pa.int64()), info_field]),
+    )
+    table = await mem_db_async.create_table("nested_json_list_add", data=seed)
+
+    await table.add([{"id": 1, "info": {"value": '{"k": 1}'}}])
+    await table.add([{"id": 2, "info": {"value": '{"k": 2}'}}], on_bad_vectors="fill")
+
+    rows = (await table.to_arrow()).sort_by("id").to_pylist()
+    assert rows == [
+        {"id": 0, "info": {"value": '{"seed":0}'}},
+        {"id": 1, "info": {"value": '{"k":1}'}},
+        {"id": 2, "info": {"value": '{"k":2}'}},
+    ]
+
+
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+@pytest.mark.asyncio
+async def test_add_list_of_dicts_to_json_list_column(mem_db_async: AsyncConnection):
+    """JSON inside a list must be JSONB-encoded, not stored as the raw text.
+
+    Storing raw text appends without error but leaves the column unreadable, so the
+    round trip is checked with a filter as well as by value.
+    """
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("docs", pa.list_(pa.field("item", pa.json_()))),
+        ]
+    )
+    table = await mem_db_async.create_table("json_list_add", schema=schema)
+
+    await table.add([{"id": 1, "docs": ['{"k": 1}', '{"k": 2}']}])
+    await table.add([{"id": 2, "docs": ['{"k": 3}']}], on_bad_vectors="fill")
+
+    rows = (await table.to_arrow()).sort_by("id").to_pylist()
+    assert rows == [
+        {"id": 1, "docs": ['{"k":1}', '{"k":2}']},
+        {"id": 2, "docs": ['{"k":3}']},
+    ]
+
+    matched = await table.query().where("json_extract(docs[1], '$.k') = 3").to_arrow()
+    assert matched.column("id").to_pylist() == [2]
+
+
+@pytest.mark.skipif(not hasattr(pa, "json_"), reason="requires PyArrow JSON type")
+@pytest.mark.asyncio
+async def test_add_map_of_json_values(mem_db_async: AsyncConnection):
+    """JSON in a map's values needs the same encoding a list's items do."""
+    schema = pa.schema(
+        [
+            pa.field("id", pa.int64()),
+            pa.field("m", pa.map_(pa.string(), pa.json_())),
+        ]
+    )
+    table = await mem_db_async.create_table(
+        "json_map_add",
+        schema=schema,
+        storage_options={"new_table_data_storage_version": "2.2"},
+    )
+
+    def batch(row_id: int, text: str) -> pa.Table:
+        return pa.table(
+            {
+                "id": pa.array([row_id], type=pa.int64()),
+                "m": pa.array([[("k", text)]], type=pa.map_(pa.string(), pa.string())),
+            }
+        )
+
+    await table.add(batch(1, '{"x": 1}'))
+    await table.add(batch(2, '{"x": 2}'), on_bad_vectors="fill")
+
+    rows = (await table.to_arrow()).sort_by("id").to_pylist()
+    assert rows == [
+        {"id": 1, "m": [("k", '{"x":1}')]},
+        {"id": 2, "m": [("k", '{"x":2}')]},
+    ]
+
+    matched = (
+        await table.query()
+        .where("json_extract(element_at(m, 'k')[1], '$.x') = '2'")
+        .to_arrow()
+    )
+    assert matched.column("id").to_pylist() == [2]
+
+
 def test_create_with_embedding_function(mem_db: DBConnection):
     class MyTable(LanceModel):
         text: str

rust/lancedb/src/table/add_data.rs

@@ -1104,4 +1104,220 @@ mod tests {
             }
         }
     }
+
+    /// A batch whose json values are all null infers as `DataType::Null` (this is what
+    /// pyarrow produces for a one-row insert with no value). The column's lance.json
+    /// identity lives in the field metadata, so dropping it while casting used to make
+    /// lance-core reject the batch as a schema mismatch.
+    #[tokio::test]
+    async fn test_add_all_null_json_column() {
+        use arrow_array::{Array, cast::AsArray, new_null_array};
+
+        let table_schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            lance_arrow::json::json_field("data", true),
+        ]));
+
+        let db = connect("memory://").execute().await.unwrap();
+        let table = db
+            .create_empty_table("json_nulls", table_schema)
+            .execute()
+            .await
+            .unwrap();
+
+        let null_batch = |ids: Vec<i64>| {
+            let len = ids.len();
+            RecordBatch::try_new(
+                Arc::new(Schema::new(vec![
+                    Field::new("id", DataType::Int64, false),
+                    Field::new("data", DataType::Null, true),
+                ])),
+                vec![
+                    Arc::new(arrow_array::Int64Array::from(ids)),
+                    new_null_array(&DataType::Null, len),
+                ],
+            )
+            .unwrap()
+        };
+
+        // A single all-null row as the very first write, then again after real JSON has
+        // been written - both scenarios from the bug report.
+        table.add(null_batch(vec![1])).execute().await.unwrap();
+
+        let arrow_json_field = Field::new("data", DataType::Utf8, true).with_metadata(
+            std::collections::HashMap::from([(
+                lance_arrow::ARROW_EXT_NAME_KEY.to_string(),
+                lance_arrow::json::ARROW_JSON_EXT_NAME.to_string(),
+            )]),
+        );
+        let populated = RecordBatch::try_new(
+            Arc::new(Schema::new(vec![
+                Field::new("id", DataType::Int64, false),
+                arrow_json_field,
+            ])),
+            vec![
+                Arc::new(arrow_array::Int64Array::from(vec![2])),
+                Arc::new(arrow_array::StringArray::from(vec![Some(r#"{"a": 1}"#)])),
+            ],
+        )
+        .unwrap();
+        table.add(populated).execute().await.unwrap();
+        table.add(null_batch(vec![3])).execute().await.unwrap();
+
+        assert_eq!(table.count_rows(None).await.unwrap(), 3);
+
+        let results: Vec<RecordBatch> = table
+            .query()
+            .execute()
+            .await
+            .unwrap()
+            .try_collect()
+            .await
+            .unwrap();
+        let batch = arrow_select::concat::concat_batches(&results[0].schema(), &results).unwrap();
+        let ids = batch
+            .column_by_name("id")
+            .unwrap()
+            .as_primitive::<arrow::datatypes::Int64Type>();
+        let json_strs = batch.column_by_name("data").unwrap().as_string::<i32>();
+        for row in 0..batch.num_rows() {
+            match ids.value(row) {
+                2 => assert_eq!(json_strs.value(row), r#"{"a":1}"#),
+                _ => assert!(json_strs.is_null(row), "row {row} expected null"),
+            }
+        }
+    }
+
+    /// JSON text with no arrow.json label - what pyarrow infers for a column of `str` - is
+    /// encoded as JSONB rather than stored verbatim, at the top level and inside a struct.
+    #[tokio::test]
+    async fn test_add_unlabelled_json_strings() {
+        use arrow_array::{Array, cast::AsArray};
+        use arrow_schema::Fields;
+
+        let table_schema = Arc::new(Schema::new(vec![
+            lance_arrow::json::json_field("data", true),
+            Field::new(
+                "info",
+                DataType::Struct(vec![lance_arrow::json::json_field("value", true)].into()),
+                true,
+            ),
+        ]));
+
+        let db = connect("memory://").execute().await.unwrap();
+        let table = db
+            .create_empty_table("json_strings", table_schema)
+            .execute()
+            .await
+            .unwrap();
+
+        let nested_children: Fields = vec![Field::new("value", DataType::Utf8, true)].into();
+        let input_schema = Arc::new(Schema::new(vec![
+            Field::new("data", DataType::Utf8, true),
+            Field::new("info", DataType::Struct(nested_children.clone()), true),
+        ]));
+        let batch = RecordBatch::try_new(
+            input_schema,
+            vec![
+                Arc::new(arrow_array::StringArray::from(vec![
+                    Some(r#"{"a": 1}"#),
+                    None,
+                ])),
+                Arc::new(arrow_array::StructArray::new(
+                    nested_children,
+                    vec![Arc::new(arrow_array::StringArray::from(vec![
+                        Some(r#"{"b": 2}"#),
+                        None,
+                    ]))],
+                    None,
+                )),
+            ],
+        )
+        .unwrap();
+        table.add(batch).execute().await.unwrap();
+
+        let results: Vec<RecordBatch> = table
+            .query()
+            .execute()
+            .await
+            .unwrap()
+            .try_collect()
+            .await
+            .unwrap();
+        let batch = arrow_select::concat::concat_batches(&results[0].schema(), &results).unwrap();
+        assert_eq!(batch.num_rows(), 2);
+
+        let data = batch.column_by_name("data").unwrap().as_string::<i32>();
+        assert_eq!(data.value(0), r#"{"a":1}"#);
+        assert!(data.is_null(1));
+
+        let nested = batch
+            .column_by_name("info")
+            .unwrap()
+            .as_struct()
+            .column_by_name("value")
+            .unwrap()
+            .as_string::<i32>();
+        assert_eq!(nested.value(0), r#"{"b":2}"#);
+        assert!(nested.is_null(1));
+    }
+
+    /// A null struct row survives a cast of one of its children, even when that child is
+    /// non-nullable. Lance checks a non-nullable child for nulls without applying the
+    /// parent's validity, so rebuilding the struct must leave the children untouched.
+    #[tokio::test]
+    async fn test_add_null_struct_with_non_nullable_child() {
+        use arrow_array::{Array, cast::AsArray};
+        use arrow_schema::Fields;
+
+        let table_schema = Arc::new(Schema::new(vec![Field::new(
+            "s",
+            DataType::Struct(vec![Field::new("x", DataType::Int64, false)].into()),
+            true,
+        )]));
+
+        let db = connect("memory://").execute().await.unwrap();
+        let table = db
+            .create_empty_table("null_struct", table_schema)
+            .execute()
+            .await
+            .unwrap();
+
+        // Int32 rather than the table's Int64, so the struct goes through reconstruction.
+        let input_children: Fields = vec![Field::new("x", DataType::Int32, false)].into();
+        let input_schema = Arc::new(Schema::new(vec![Field::new(
+            "s",
+            DataType::Struct(input_children.clone()),
+            true,
+        )]));
+        let batch = RecordBatch::try_new(
+            input_schema,
+            vec![Arc::new(arrow_array::StructArray::new(
+                input_children,
+                vec![Arc::new(arrow_array::Int32Array::from(vec![0, 6]))],
+                Some(arrow::buffer::NullBuffer::from(vec![false, true])),
+            ))],
+        )
+        .unwrap();
+        table.add(batch).execute().await.unwrap();
+
+        let results: Vec<RecordBatch> = table
+            .query()
+            .execute()
+            .await
+            .unwrap()
+            .try_collect()
+            .await
+            .unwrap();
+        let batch = arrow_select::concat::concat_batches(&results[0].schema(), &results).unwrap();
+        let s = batch.column_by_name("s").unwrap().as_struct();
+        assert!(s.is_null(0));
+        assert_eq!(
+            s.column_by_name("x")
+                .unwrap()
+                .as_primitive::<arrow::datatypes::Int64Type>()
+                .value(1),
+            6
+        );
+    }
 }

rust/lancedb/src/table/datafusion/cast.rs

@@ -1,20 +1,25 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
-use std::sync::Arc;
+use std::collections::HashMap;
+use std::sync::{Arc, LazyLock};
 
+use arrow_array::StructArray;
+use arrow_array::cast::AsArray;
 use arrow_cast::can_cast_types;
 use arrow_schema::{DataType, Field, FieldRef, Fields, Schema};
 use datafusion::functions::core::{get_field, named_struct};
-use datafusion_common::ScalarValue;
 use datafusion_common::config::ConfigOptions;
+use datafusion_common::metadata::FieldMetadata;
+use datafusion_common::{DataFusionError, Result as DFResult, ScalarValue};
+use datafusion_expr::{ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility};
 use datafusion_physical_expr::ScalarFunctionExpr;
 use datafusion_physical_expr::expressions::{CastExpr, Literal};
 use datafusion_physical_plan::expressions::Column;
 use datafusion_physical_plan::projection::ProjectionExec;
 use datafusion_physical_plan::{ExecutionPlan, PhysicalExpr};
-use lance_arrow::FieldExt;
-use lance_arrow::json::{is_arrow_json_field, is_json_field};
+use lance_arrow::json::{ARROW_JSON_EXT_NAME, has_json_fields, is_arrow_json_field, is_json_field};
+use lance_arrow::{ARROW_EXT_NAME_KEY, FieldExt};
 
 use super::blob_coerce::coerce_blob_expr;
 use crate::{Error, Result};
@@ -67,18 +72,33 @@ fn build_field_exprs(
         let input_field = &input_fields[input_idx];
         let input_expr = get_input_expr(input_idx);
 
-        // Special case: input is arrow.json (PyArrow pa.json_() extension type backed by
-        // Utf8/LargeUtf8) and the table field is lance.json (backed by LargeBinary).
-        // Lance-core's write path already handles the arrow.json → lance.json conversion
-        // (including JSONB encoding), so we pass the expression through unchanged and let
-        // lance-core deal with it. Attempting to cast Utf8 → LargeBinary here would
-        // produce a field whose metadata still identifies it as arrow.json, which then
-        // causes a schema-mismatch error inside lance-core.
-        if is_arrow_json_field(input_field) && is_json_field(table_field) {
+        // PyArrow's pa.json_() is already labelled arrow.json, which is what lance-core wants
+        // to see, so pass it straight through.
+        if is_json_field(table_field) && is_arrow_json_field(input_field) {
             result.push((input_expr, Arc::clone(input_field) as FieldRef));
             continue;
         }
 
+        // Anything else destined for a json column needs its JSON leaves labelled; see
+        // `json_write_target`. Structs are excluded because the recursion below rebuilds them
+        // field by field, which also handles reordered and partial input.
+        if !matches!(table_field.data_type(), DataType::Struct(_))
+            && let Some(target) = json_write_target(input_field, table_field)
+            && can_cast_types(input_field.data_type(), target.data_type())
+        {
+            // The label goes on the cast's target field rather than the field returned
+            // alongside it, because DataFusion derives the projection's output schema from
+            // `PhysicalExpr::return_field`.
+            let target: FieldRef = Arc::new(target);
+            let expr = Arc::new(CastExpr::new_with_target_field(
+                input_expr,
+                target.clone(),
+                None,
+            ));
+            result.push((expr, target));
+            continue;
+        }
+
         // Blob columns accept raw binary on write; exact matches pass through below.
         if table_field.is_blob_v2() && input_field.as_ref() != table_field.as_ref() {
             result.push(coerce_blob_expr(
@@ -90,6 +110,18 @@ fn build_field_exprs(
             continue;
         }
 
+        // A column whose values are all null infers as `Null` (pyarrow does this for a list of
+        // dicts), so there is no input type to cast from. Emit typed nulls carrying the table
+        // field verbatim: a plain cast would drop the field metadata, and extension columns
+        // such as lance.json are identified by that metadata alone, so lance-core would then
+        // reject the batch as a schema mismatch.
+        if matches!(input_field.data_type(), DataType::Null)
+            && !matches!(table_field.data_type(), DataType::Null)
+        {
+            result.push((null_literal(table_field)?, table_field.clone()));
+            continue;
+        }
+
         let expr = match (input_field.data_type(), table_field.data_type()) {
             // Both are structs: recurse into sub-fields to handle subschemas and casts.
             (DataType::Struct(in_children), DataType::Struct(tbl_children))
@@ -137,7 +169,16 @@ fn build_field_exprs(
                     config.clone(),
                 ));
 
-                result.push((ns_expr, output_field));
+                result.push((
+                    restore_struct_validity(
+                        ns_expr,
+                        input_expr,
+                        input_field,
+                        &output_field,
+                        config.clone(),
+                    ),
+                    output_field,
+                ));
                 continue;
             }
             // Types match: pass through.
@@ -171,6 +212,193 @@ fn build_field_exprs(
     Ok(result)
 }
 
+// `named_struct` returns a struct with no null bitmap, and the `get_field` calls feeding it
+// read each child without applying the parent's validity, so a null input struct would come
+// back non-null with its masked children exposed. Move the input's null buffer onto the
+// rebuilt struct.
+fn restore_struct_validity(
+    rebuilt: Arc<dyn PhysicalExpr>,
+    input_expr: Arc<dyn PhysicalExpr>,
+    input_field: &FieldRef,
+    output_field: &FieldRef,
+    config: Arc<ConfigOptions>,
+) -> Arc<dyn PhysicalExpr> {
+    if !input_field.is_nullable() {
+        return rebuilt;
+    }
+
+    Arc::new(ScalarFunctionExpr::new(
+        &format!("restore_validity({})", output_field.name()),
+        RESTORE_VALIDITY_UDF.clone(),
+        vec![rebuilt, input_expr],
+        output_field.clone(),
+        config,
+    ))
+}
+
+static RESTORE_VALIDITY_UDF: LazyLock<Arc<datafusion_expr::ScalarUDF>> =
+    LazyLock::new(|| Arc::new(datafusion_expr::ScalarUDF::from(RestoreValidityUdf::new())));
+
+/// Returns its first argument, a struct, carrying the null buffer of its second.
+///
+/// Selecting a typed null for the null rows instead would nullify their children too, which
+/// Lance rejects outright for a non-nullable child, even where the parent masks it. Children
+/// therefore have to survive the round trip byte for byte.
+#[derive(Debug, Hash, PartialEq, Eq)]
+struct RestoreValidityUdf {
+    signature: Signature,
+}
+
+impl RestoreValidityUdf {
+    fn new() -> Self {
+        Self {
+            signature: Signature::any(2, Volatility::Immutable),
+        }
+    }
+}
+
+impl ScalarUDFImpl for RestoreValidityUdf {
+    fn name(&self) -> &str {
+        "restore_validity"
+    }
+
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn return_type(&self, arg_types: &[DataType]) -> DFResult<DataType> {
+        Ok(arg_types[0].clone())
+    }
+
+    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> DFResult<ColumnarValue> {
+        let rows = args.number_rows;
+        let rebuilt = args.args[0].to_array(rows)?;
+        let nulls_from = args.args[1].to_array(rows)?;
+
+        let rebuilt = rebuilt.as_struct_opt().ok_or_else(|| {
+            DataFusionError::Internal(format!(
+                "restore_validity expects a struct, got {}",
+                rebuilt.data_type()
+            ))
+        })?;
+
+        let nulls = nulls_from.logical_nulls();
+        let (fields, columns, _) = rebuilt.clone().into_parts();
+        let restored = StructArray::try_new_with_length(fields, columns, nulls, rows)?;
+        Ok(ColumnarValue::Array(Arc::new(restored)))
+    }
+}
+
+// The storage type arrow.json would use for `input`, or None if it cannot hold JSON text.
+fn arrow_json_storage_type(input: &DataType) -> Option<DataType> {
+    match input {
+        // arrow.json only recognises Utf8 and LargeUtf8 storage, so a view has to be cast.
+        DataType::Utf8 | DataType::Utf8View => Some(DataType::Utf8),
+        DataType::LargeUtf8 => Some(DataType::LargeUtf8),
+        _ => None,
+    }
+}
+
+fn arrow_json_field(name: &str, storage: DataType, nullable: bool) -> Field {
+    Field::new(name, storage, nullable).with_metadata(HashMap::from([(
+        ARROW_EXT_NAME_KEY.to_string(),
+        ARROW_JSON_EXT_NAME.to_string(),
+    )]))
+}
+
+/// Rewrite `table_field` so that every lance.json leaf the input supplies as text becomes an
+/// arrow.json leaf, leaving the rest of the shape untouched.
+///
+/// Lance-core encodes JSON text into JSONB on write, but only for leaves labelled arrow.json.
+/// Casting to the lance.json storage type instead relabels raw text as JSONB, which appends
+/// successfully but leaves the column unreadable, so the label has to reach every leaf however
+/// deeply it is nested. Returns `None` when there is nothing to relabel.
+fn json_write_target(input_field: &Field, table_field: &Field) -> Option<Field> {
+    if is_json_field(table_field) {
+        let storage = if is_arrow_json_field(input_field) {
+            input_field.data_type().clone()
+        } else {
+            arrow_json_storage_type(input_field.data_type())?
+        };
+        return Some(arrow_json_field(
+            table_field.name(),
+            storage,
+            table_field.is_nullable(),
+        ));
+    }
+
+    if !has_json_fields(table_field) {
+        return None;
+    }
+
+    let relabelled = match (input_field.data_type(), table_field.data_type()) {
+        (
+            DataType::List(input_item)
+            | DataType::LargeList(input_item)
+            | DataType::FixedSizeList(input_item, _),
+            DataType::List(table_item)
+            | DataType::LargeList(table_item)
+            | DataType::FixedSizeList(table_item, _),
+        ) => {
+            let item: FieldRef = Arc::new(json_write_target(input_item, table_item)?);
+            match table_field.data_type() {
+                DataType::List(_) => DataType::List(item),
+                DataType::LargeList(_) => DataType::LargeList(item),
+                DataType::FixedSizeList(_, len) => DataType::FixedSizeList(item, *len),
+                _ => unreachable!("matched a list type above"),
+            }
+        }
+        (DataType::Map(input_entries, _), DataType::Map(table_entries, sorted)) => {
+            let entries = json_write_target(input_entries, table_entries)?;
+            DataType::Map(Arc::new(entries), *sorted)
+        }
+        (DataType::Struct(input_children), DataType::Struct(table_children)) => {
+            let mut children = Vec::with_capacity(table_children.len());
+            let mut relabelled_any = false;
+            for table_child in table_children {
+                let relabelled_child = input_children
+                    .iter()
+                    .find(|f| f.name() == table_child.name())
+                    .and_then(|input_child| json_write_target(input_child, table_child));
+                match relabelled_child {
+                    Some(child) => {
+                        relabelled_any = true;
+                        children.push(Arc::new(child));
+                    }
+                    None => children.push(table_child.clone()),
+                }
+            }
+            if !relabelled_any {
+                return None;
+            }
+            DataType::Struct(children.into())
+        }
+        _ => return None,
+    };
+
+    Some(
+        Field::new(table_field.name(), relabelled, table_field.is_nullable())
+            .with_metadata(table_field.metadata().clone()),
+    )
+}
+
+// The field's metadata is attached to the literal itself, because DataFusion derives the
+// projection's output schema from `PhysicalExpr::return_field` rather than from the field we
+// return alongside the expression.
+fn null_literal(field: &FieldRef) -> Result<Arc<dyn PhysicalExpr>> {
+    let scalar = ScalarValue::try_new_null(field.data_type()).map_err(|e| Error::InvalidInput {
+        message: format!(
+            "cannot build null literal for column '{}' of type {}: {e}",
+            field.name(),
+            field.data_type()
+        ),
+    })?;
+    Ok(Arc::new(Literal::new_with_metadata(
+        scalar,
+        Some(FieldMetadata::from(field.as_ref())),
+    )))
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
@@ -715,4 +943,379 @@ mod tests {
         assert!(result.column(0).is_null(1));
         assert_eq!(v2, r#"{"y": 2}"#);
     }
+
+    /// Plain JSON text (what pyarrow infers for a column of `str`, and what a caller writing
+    /// JSON by hand supplies) has to be labelled arrow.json so lance-core encodes it as JSONB.
+    /// Casting it to the table field's LargeBinary storage type would store the raw text.
+    #[rstest::rstest]
+    #[case::utf8(DataType::Utf8, DataType::Utf8)]
+    #[case::large_utf8(DataType::LargeUtf8, DataType::LargeUtf8)]
+    #[case::utf8_view(DataType::Utf8View, DataType::Utf8)]
+    #[tokio::test]
+    async fn test_unlabelled_string_into_lance_json_gets_arrow_json_label(
+        #[case] input_type: DataType,
+        #[case] expected_type: DataType,
+    ) {
+        use lance_arrow::json::{is_arrow_json_field, json_field};
+
+        let table_schema = Schema::new(vec![json_field("data", true)]);
+
+        let input_schema = Arc::new(Schema::new(vec![Field::new("data", input_type, true)]));
+        let values = vec![Some(r#"{"x": 1}"#), None];
+        let input_array = arrow_cast::cast(
+            &StringArray::from(values) as &dyn arrow_array::Array,
+            input_schema.field(0).data_type(),
+        )
+        .unwrap();
+        let input_batch = RecordBatch::try_new(input_schema, vec![input_array]).unwrap();
+
+        let plan = plan_from_batch(input_batch).await;
+        let projected = cast_to_table_schema(plan, &table_schema).unwrap();
+
+        let out_field = projected.schema().field_with_name("data").unwrap().clone();
+        assert_eq!(out_field.data_type(), &expected_type);
+        assert!(
+            is_arrow_json_field(&out_field),
+            "output field must be labelled arrow.json, got {:?}",
+            out_field.metadata()
+        );
+
+        let result = collect(projected).await;
+        assert_eq!(result.num_rows(), 2);
+        assert_eq!(result.column(0).null_count(), 1);
+    }
+
+    /// A json leaf inside a list is relabelled too. The outer field is a list, so without
+    /// recursing into it the generic container cast would turn the text into LargeBinary
+    /// labelled lance.json - an append that succeeds but stores unreadable JSON.
+    #[rstest::rstest]
+    #[case::unlabelled(DataType::Utf8, false)]
+    #[case::already_labelled(DataType::Utf8, true)]
+    #[tokio::test]
+    async fn test_unlabelled_list_item_into_lance_json_gets_arrow_json_label(
+        #[case] item_type: DataType,
+        #[case] input_labelled: bool,
+    ) {
+        use lance_arrow::json::{is_arrow_json_field, json_field};
+
+        use super::arrow_json_field;
+
+        let table_schema = Schema::new(vec![Field::new(
+            "docs",
+            DataType::List(Arc::new(json_field("item", true))),
+            true,
+        )]);
+
+        let input_item = if input_labelled {
+            Arc::new(arrow_json_field("item", item_type, true))
+        } else {
+            Arc::new(Field::new("item", item_type, true))
+        };
+        let input_schema = Arc::new(Schema::new(vec![Field::new(
+            "docs",
+            DataType::List(input_item.clone()),
+            true,
+        )]));
+        let values = StringArray::from(vec![Some(r#"{"k": 1}"#), Some(r#"{"k": 2}"#)]);
+        let input_batch = RecordBatch::try_new(
+            input_schema,
+            vec![Arc::new(ListArray::new(
+                input_item,
+                OffsetBuffer::new(vec![0, 1, 2].into()),
+                Arc::new(values),
+                None,
+            ))],
+        )
+        .unwrap();
+
+        let plan = plan_from_batch(input_batch).await;
+        let projected = cast_to_table_schema(plan, &table_schema).unwrap();
+
+        let out_field = projected.schema().field_with_name("docs").unwrap().clone();
+        let DataType::List(out_item) = out_field.data_type() else {
+            panic!("expected a list, got {}", out_field.data_type());
+        };
+        assert!(
+            is_arrow_json_field(out_item),
+            "the list item must be labelled arrow.json, got {out_item:?}"
+        );
+
+        let result = collect(projected).await;
+        assert_eq!(result.num_rows(), 2);
+    }
+
+    /// The same, for a json column nested inside a struct: the struct is rebuilt from its
+    /// children, so the label has to travel on the child field.
+    #[tokio::test]
+    async fn test_unlabelled_struct_child_into_lance_json_gets_arrow_json_label() {
+        use lance_arrow::json::{is_arrow_json_field, json_field};
+
+        let table_schema = Schema::new(vec![Field::new(
+            "info",
+            DataType::Struct(
+                vec![
+                    Field::new("id", DataType::Int64, true),
+                    json_field("value", true),
+                ]
+                .into(),
+            ),
+            true,
+        )]);
+
+        let input_children: Fields = vec![
+            Field::new("id", DataType::Int64, true),
+            Field::new("value", DataType::Utf8, true),
+        ]
+        .into();
+        let input_schema = Arc::new(Schema::new(vec![Field::new(
+            "info",
+            DataType::Struct(input_children.clone()),
+            true,
+        )]));
+        let input_batch = RecordBatch::try_new(
+            input_schema,
+            vec![Arc::new(StructArray::new(
+                input_children,
+                vec![
+                    Arc::new(Int64Array::from(vec![1, 2])),
+                    Arc::new(StringArray::from(vec![Some(r#"{"a": 1}"#), None])),
+                ],
+                None,
+            ))],
+        )
+        .unwrap();
+
+        let plan = plan_from_batch(input_batch).await;
+        let projected = cast_to_table_schema(plan, &table_schema).unwrap();
+
+        let out_field = projected.schema().field_with_name("info").unwrap().clone();
+        let DataType::Struct(out_children) = out_field.data_type() else {
+            panic!("expected a struct, got {}", out_field.data_type());
+        };
+        let value = out_children.iter().find(|f| f.name() == "value").unwrap();
+        assert!(
+            is_arrow_json_field(value),
+            "nested field must be labelled arrow.json, got {:?}",
+            value.metadata()
+        );
+
+        let result = collect(projected).await;
+        let info: &StructArray = result.column(0).as_any().downcast_ref().unwrap();
+        assert_eq!(info.column_by_name("value").unwrap().null_count(), 1);
+    }
+
+    /// An all-null column comes through as `DataType::Null` (pyarrow infers that for a batch
+    /// of dicts whose values are all `None`). The lance.json extension metadata lives on the
+    /// field alone, so it has to be carried into the output schema or lance-core rejects the
+    /// batch with a "json vs large_binary" schema mismatch.
+    #[tokio::test]
+    async fn test_null_column_into_lance_json_keeps_extension_metadata() {
+        use lance_arrow::ARROW_EXT_NAME_KEY;
+        use lance_arrow::json::{JSON_EXT_NAME, json_field};
+
+        let table_schema = Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            json_field("data", true),
+        ]);
+
+        let input_schema = Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            Field::new("data", DataType::Null, true),
+        ]));
+        let input_batch = RecordBatch::try_new(
+            input_schema,
+            vec![
+                Arc::new(Int64Array::from(vec![0, 1, 2])),
+                arrow_array::new_null_array(&DataType::Null, 3),
+            ],
+        )
+        .unwrap();
+
+        let plan = plan_from_batch(input_batch).await;
+        let projected = cast_to_table_schema(plan, &table_schema).unwrap();
+
+        let out_field = projected.schema().field_with_name("data").unwrap().clone();
+        assert_eq!(out_field.data_type(), &DataType::LargeBinary);
+        assert_eq!(
+            out_field
+                .metadata()
+                .get(ARROW_EXT_NAME_KEY)
+                .map(|s| s.as_str()),
+            Some(JSON_EXT_NAME),
+            "output field must still identify itself as lance.json"
+        );
+
+        let result = collect(projected).await;
+        assert_eq!(result.num_rows(), 3);
+        assert_eq!(result.column_by_name("data").unwrap().null_count(), 3);
+    }
+
+    /// The same, for a lance.json column nested inside a struct: the struct is rebuilt from
+    /// its children, so each child field must keep its own metadata, and a null struct must
+    /// stay null even though it is rebuilt child by child.
+    #[tokio::test]
+    async fn test_null_struct_child_into_lance_json_keeps_extension_metadata() {
+        use lance_arrow::ARROW_EXT_NAME_KEY;
+        use lance_arrow::json::{JSON_EXT_NAME, json_field};
+
+        let table_schema = Schema::new(vec![Field::new(
+            "meta",
+            DataType::Struct(
+                vec![
+                    Field::new("id", DataType::Int64, true),
+                    json_field("doc", true),
+                ]
+                .into(),
+            ),
+            true,
+        )]);
+
+        let input_children: Fields = vec![
+            Field::new("id", DataType::Int64, true),
+            Field::new("doc", DataType::Null, true),
+        ]
+        .into();
+        let input_schema = Arc::new(Schema::new(vec![Field::new(
+            "meta",
+            DataType::Struct(input_children.clone()),
+            true,
+        )]));
+        let input_batch = RecordBatch::try_new(
+            input_schema,
+            vec![Arc::new(StructArray::new(
+                input_children,
+                vec![
+                    Arc::new(Int64Array::from(vec![7, 8])),
+                    arrow_array::new_null_array(&DataType::Null, 2),
+                ],
+                Some(arrow::buffer::NullBuffer::from(vec![false, true])),
+            ))],
+        )
+        .unwrap();
+
+        let plan = plan_from_batch(input_batch).await;
+        let projected = cast_to_table_schema(plan, &table_schema).unwrap();
+
+        let out_field = projected.schema().field_with_name("meta").unwrap().clone();
+        let DataType::Struct(out_children) = out_field.data_type() else {
+            panic!("expected a struct, got {}", out_field.data_type());
+        };
+        let doc = out_children.iter().find(|f| f.name() == "doc").unwrap();
+        assert_eq!(doc.data_type(), &DataType::LargeBinary);
+        assert_eq!(
+            doc.metadata().get(ARROW_EXT_NAME_KEY).map(|s| s.as_str()),
+            Some(JSON_EXT_NAME)
+        );
+
+        let result = collect(projected).await;
+        let meta: &StructArray = result.column(0).as_any().downcast_ref().unwrap();
+        assert!(meta.is_null(0), "a null struct must stay null once rebuilt");
+        assert!(meta.is_valid(1));
+        assert_eq!(meta.column_by_name("doc").unwrap().null_count(), 2);
+    }
+
+    /// Any struct whose children need adjusting is rebuilt child by child, so the parent's
+    /// nulls have to be restored afterwards - not only for the extension-column cases.
+    #[tokio::test]
+    async fn test_null_struct_stays_null_when_child_is_cast() {
+        let input_children: Fields = vec![Field::new("x", DataType::Int32, true)].into();
+        let table_schema = Schema::new(vec![Field::new(
+            "s",
+            DataType::Struct(vec![Field::new("x", DataType::Int64, true)].into()),
+            true,
+        )]);
+
+        let input_schema = Arc::new(Schema::new(vec![Field::new(
+            "s",
+            DataType::Struct(input_children.clone()),
+            true,
+        )]));
+        let input_batch = RecordBatch::try_new(
+            input_schema,
+            vec![Arc::new(StructArray::new(
+                input_children,
+                vec![Arc::new(Int32Array::from(vec![5, 6]))],
+                Some(arrow::buffer::NullBuffer::from(vec![false, true])),
+            ))],
+        )
+        .unwrap();
+
+        let plan = plan_from_batch(input_batch).await;
+        let projected = cast_to_table_schema(plan, &table_schema).unwrap();
+
+        let result = collect(projected).await;
+        let s: &StructArray = result.column(0).as_any().downcast_ref().unwrap();
+        assert!(s.is_null(0));
+        assert!(s.is_valid(1));
+        let x: &Int64Array = s.column(0).as_any().downcast_ref().unwrap();
+        assert_eq!(x.value(1), 6);
+    }
+
+    /// Lance rejects a non-nullable child that carries nulls even where the parent masks
+    /// them, so the null rows have to keep the placeholder children the input gave them.
+    #[tokio::test]
+    async fn test_null_struct_keeps_children_of_a_non_nullable_child() {
+        let input_children: Fields = vec![Field::new("x", DataType::Int32, false)].into();
+        let table_schema = Schema::new(vec![Field::new(
+            "s",
+            DataType::Struct(vec![Field::new("x", DataType::Int64, false)].into()),
+            true,
+        )]);
+
+        let input_schema = Arc::new(Schema::new(vec![Field::new(
+            "s",
+            DataType::Struct(input_children.clone()),
+            true,
+        )]));
+        let input_batch = RecordBatch::try_new(
+            input_schema,
+            vec![Arc::new(StructArray::new(
+                input_children,
+                vec![Arc::new(Int32Array::from(vec![0, 6]))],
+                Some(arrow::buffer::NullBuffer::from(vec![false, true])),
+            ))],
+        )
+        .unwrap();
+
+        let plan = plan_from_batch(input_batch).await;
+        let projected = cast_to_table_schema(plan, &table_schema).unwrap();
+
+        let result = collect(projected).await;
+        let s: &StructArray = result.column(0).as_any().downcast_ref().unwrap();
+        assert!(s.is_null(0));
+        assert!(s.is_valid(1));
+        let x: &Int64Array = s.column(0).as_any().downcast_ref().unwrap();
+        assert_eq!(x.null_count(), 0);
+        assert_eq!(x.value(1), 6);
+    }
+
+    /// A `Null` input column against a plain table column writes nulls too, including for
+    /// target types that a DataFusion cast would not reach.
+    #[tokio::test]
+    async fn test_null_column_into_struct_column() {
+        let children: Fields = vec![Field::new("x", DataType::Int32, true)].into();
+        let table_schema = Schema::new(vec![Field::new(
+            "s",
+            DataType::Struct(children.clone()),
+            true,
+        )]);
+
+        let input_schema = Arc::new(Schema::new(vec![Field::new("s", DataType::Null, true)]));
+        let input_batch = RecordBatch::try_new(
+            input_schema,
+            vec![arrow_array::new_null_array(&DataType::Null, 2)],
+        )
+        .unwrap();
+
+        let plan = plan_from_batch(input_batch).await;
+        let projected = cast_to_table_schema(plan, &table_schema).unwrap();
+        assert_eq!(
+            projected.schema().field_with_name("s").unwrap().data_type(),
+            &DataType::Struct(children)
+        );
+
+        let result = collect(projected).await;
+        assert_eq!(result.num_rows(), 2);
+        assert_eq!(result.column(0).null_count(), 2);
+    }
 }

rust/lancedb/tests/blob_integration.rs

@@ -47,6 +47,21 @@ fn binary_input_batch(ids: &[i64], payloads: &[Option<&[u8]>]) -> RecordBatch {
     .unwrap()
 }
 
+/// What pyarrow infers for a batch of dicts whose blob values are all `None`.
+fn null_typed_input_batch(ids: &[i64]) -> RecordBatch {
+    RecordBatch::try_new(
+        Arc::new(Schema::new(vec![
+            Field::new("id", DataType::Int64, false),
+            Field::new("image", DataType::Null, true),
+        ])),
+        vec![
+            Arc::new(Int64Array::from(ids.to_vec())),
+            new_null_array(&DataType::Null, ids.len()),
+        ],
+    )
+    .unwrap()
+}
+
 async fn create_inline_blob_table(
     db: &Connection,
     name: &str,
@@ -255,6 +270,38 @@ async fn add_accepts_null_blob_rows() -> Result<()> {
     Ok(())
 }
 
+/// A batch whose blob values are all null carries no type information — pyarrow infers
+/// `DataType::Null` for it, which is what a row-at-a-time insert of an optional blob column
+/// looks like. Such a batch must be accepted, both as the first write and after bytes have
+/// already been written.
+#[tokio::test]
+async fn add_accepts_all_null_typed_blob_column() -> Result<()> {
+    let tmp = tempdir().unwrap();
+    let db = connect(tmp.path().to_str().unwrap()).execute().await?;
+    let table = db
+        .create_empty_table("t", blob_table_schema())
+        .execute()
+        .await?;
+
+    table.add(null_typed_input_batch(&[1])).execute().await?;
+    assert_eq!(table.count_rows(None).await?, 1);
+    assert!(query_image_struct(&table).await.is_null(0));
+
+    table
+        .add(binary_input_batch(&[2], &[Some(b"bytes".as_slice())]))
+        .execute()
+        .await?;
+    table.add(null_typed_input_batch(&[3])).execute().await?;
+    assert_eq!(table.count_rows(None).await?, 3);
+
+    let row_ids = collect_row_ids(&table).await?;
+    let bytes = table.fetch_blobs("image", &row_ids).await?;
+    assert!(bytes.is_null(0));
+    assert_eq!(bytes.value(1), b"bytes");
+    assert!(bytes.is_null(2));
+    Ok(())
+}
+
 #[tokio::test]
 async fn add_rejects_uncoercible_blob_input() -> Result<()> {
     let tmp = tempdir().unwrap();