chore: update lance dependency to v13.0.0-beta.12 (#4265)

Update the Rust workspace Lance dependencies and Java lance-core from
v13.0.0-beta.8 to
[v13.0.0-beta.12](https://github.com/lance-format/lance/releases/tag/v13.0.0-beta.12),
refreshing Cargo.lock.
No compatibility fixes were required; validation passed with `cargo
clippy --quiet --workspace --tests --all-features -- -D warnings`,
`cargo fmt --all --quiet`, and `git diff --check`.

---------

Co-authored-by: Lu Qiu <luqiujob@gmail.com>
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>

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

@@ -1,7 +1,6 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
-use std::collections::HashMap;
 use std::sync::{Arc, LazyLock};
 
 use arrow_array::StructArray;
@@ -73,8 +72,12 @@ fn build_field_exprs(
         let input_expr = get_input_expr(input_idx);
 
         // 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) {
+        // to see, so pass it straight through once its extension metadata matches the table's
+        // (see `arrow_json_field`); otherwise it takes the relabelling cast below.
+        if is_json_field(table_field)
+            && is_arrow_json_field(input_field)
+            && input_field.as_ref() == &arrow_json_field(table_field, input_field.data_type())
+        {
             result.push((input_expr, Arc::clone(input_field) as FieldRef));
             continue;
         }
@@ -299,11 +302,25 @@ fn arrow_json_storage_type(input: &DataType) -> Option<DataType> {
     }
 }
 
-fn arrow_json_field(name: &str, storage: DataType, nullable: bool) -> Field {
-    Field::new(name, storage, nullable).with_metadata(HashMap::from([(
+/// The arrow.json field lance-core should see for `table_field`, a lance.json leaf.
+///
+/// Lance-core converts the leaf to lance.json by swapping the extension name and keeping the
+/// rest of the field metadata, and the file writer then requires the result to match the stored
+/// field exactly, `ARROW:extension:metadata` included. PyArrow exports every extension field
+/// with that key (empty for JSON), while a schema built in Rust omits it, so the relabelled
+/// field mirrors the table field's metadata rather than carrying a fixed set of keys.
+fn arrow_json_field(table_field: &Field, storage: &DataType) -> Field {
+    let mut metadata = table_field.metadata().clone();
+    metadata.insert(
         ARROW_EXT_NAME_KEY.to_string(),
         ARROW_JSON_EXT_NAME.to_string(),
-    )]))
+    );
+    Field::new(
+        table_field.name(),
+        storage.clone(),
+        table_field.is_nullable(),
+    )
+    .with_metadata(metadata)
 }
 
 /// Rewrite `table_field` so that every lance.json leaf the input supplies as text becomes an
@@ -320,11 +337,7 @@ fn json_write_target(input_field: &Field, table_field: &Field) -> Option<Field>
         } else {
             arrow_json_storage_type(input_field.data_type())?
         };
-        return Some(arrow_json_field(
-            table_field.name(),
-            storage,
-            table_field.is_nullable(),
-        ));
+        return Some(arrow_json_field(table_field, &storage));
     }
 
     if !has_json_fields(table_field) {
@@ -401,6 +414,7 @@ fn null_literal(field: &FieldRef) -> Result<Arc<dyn PhysicalExpr>> {
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
     use std::sync::Arc;
 
     use arrow::buffer::OffsetBuffer;
@@ -408,7 +422,7 @@ mod tests {
         Array, Float32Array, Float64Array, Int32Array, Int64Array, ListArray, RecordBatch,
         StringArray, StructArray, UInt32Array, UInt64Array,
     };
-    use arrow_schema::{DataType, Field, Fields, Schema};
+    use arrow_schema::{DataType, Field, FieldRef, Fields, Schema};
     use datafusion::prelude::SessionContext;
     use datafusion_catalog::MemTable;
     use futures::TryStreamExt;
@@ -1007,7 +1021,7 @@ mod tests {
         )]);
 
         let input_item = if input_labelled {
-            Arc::new(arrow_json_field("item", item_type, true))
+            Arc::new(arrow_json_field(&json_field("item", true), &item_type))
         } else {
             Arc::new(Field::new("item", item_type, true))
         };
@@ -1318,4 +1332,150 @@ mod tests {
         assert_eq!(result.num_rows(), 2);
         assert_eq!(result.column(0).null_count(), 2);
     }
+
+    /// PyArrow records `ARROW:extension:metadata` (empty) alongside the extension name when it
+    /// exports a `pa.json_()` column, so a table created from one stores that key. Lance-core
+    /// swaps the name of a relabelled leaf back to lance.json but keeps the other metadata, and
+    /// the file writer then requires the field to match the stored one key for key. The
+    /// relabelled field must therefore mirror the table field's metadata.
+    #[rstest::rstest]
+    #[case::unlabelled_text(false)]
+    #[case::arrow_json_without_metadata_key(true)]
+    #[tokio::test]
+    async fn test_relabelled_json_mirrors_table_extension_metadata(#[case] input_labelled: bool) {
+        use lance_arrow::json::{ARROW_JSON_EXT_NAME, JSON_EXT_NAME};
+        use lance_arrow::{ARROW_EXT_META_KEY, ARROW_EXT_NAME_KEY};
+
+        let table_schema = Schema::new(vec![
+            Field::new("data", DataType::LargeBinary, true).with_metadata(HashMap::from([
+                (ARROW_EXT_NAME_KEY.to_string(), JSON_EXT_NAME.to_string()),
+                (ARROW_EXT_META_KEY.to_string(), String::new()),
+            ])),
+        ]);
+
+        let mut input_field = Field::new("data", DataType::Utf8, true);
+        if input_labelled {
+            input_field = input_field.with_metadata(HashMap::from([(
+                ARROW_EXT_NAME_KEY.to_string(),
+                ARROW_JSON_EXT_NAME.to_string(),
+            )]));
+        }
+        let input_schema = Arc::new(Schema::new(vec![input_field]));
+        let values = StringArray::from(vec![Some(r#"{"x": 1}"#), None]);
+        let input_batch = RecordBatch::try_new(input_schema, vec![Arc::new(values)]).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::Utf8);
+        assert_eq!(
+            out_field.metadata(),
+            &HashMap::from([
+                (
+                    ARROW_EXT_NAME_KEY.to_string(),
+                    ARROW_JSON_EXT_NAME.to_string()
+                ),
+                (ARROW_EXT_META_KEY.to_string(), String::new()),
+            ]),
+            "the arrow.json label must carry the table field's other extension metadata"
+        );
+
+        let result = collect(projected).await;
+        assert_eq!(result.num_rows(), 2);
+        assert_eq!(result.column(0).null_count(), 1);
+    }
+
+    /// The inverse: a table built in Rust records only the extension name, so an input that
+    /// carries the empty metadata key (what pyarrow exports) must drop it, or lance-core sees a
+    /// field the file schema does not describe.
+    #[tokio::test]
+    async fn test_arrow_json_input_drops_metadata_key_the_table_lacks() {
+        use lance_arrow::json::{ARROW_JSON_EXT_NAME, json_field};
+        use lance_arrow::{ARROW_EXT_META_KEY, ARROW_EXT_NAME_KEY};
+
+        let table_schema = Schema::new(vec![json_field("data", true)]);
+
+        let input_field = Field::new("data", DataType::Utf8, true).with_metadata(HashMap::from([
+            (
+                ARROW_EXT_NAME_KEY.to_string(),
+                ARROW_JSON_EXT_NAME.to_string(),
+            ),
+            (ARROW_EXT_META_KEY.to_string(), String::new()),
+        ]));
+        let input_schema = Arc::new(Schema::new(vec![input_field]));
+        let values = StringArray::from(vec![Some(r#"{"x": 1}"#)]);
+        let input_batch = RecordBatch::try_new(input_schema, vec![Arc::new(values)]).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::Utf8);
+        assert_eq!(
+            out_field.metadata(),
+            &HashMap::from([(
+                ARROW_EXT_NAME_KEY.to_string(),
+                ARROW_JSON_EXT_NAME.to_string()
+            )])
+        );
+        assert_eq!(collect(projected).await.num_rows(), 1);
+    }
+
+    /// Nested leaves mirror their table leaf's metadata too, since lance-core checks each
+    /// child of a list, map or struct against the stored child.
+    #[tokio::test]
+    async fn test_nested_relabelled_json_mirrors_table_extension_metadata() {
+        use lance_arrow::json::{ARROW_JSON_EXT_NAME, JSON_EXT_NAME};
+        use lance_arrow::{ARROW_EXT_META_KEY, ARROW_EXT_NAME_KEY};
+
+        let table_item =
+            Field::new("item", DataType::LargeBinary, true).with_metadata(HashMap::from([
+                (ARROW_EXT_NAME_KEY.to_string(), JSON_EXT_NAME.to_string()),
+                (ARROW_EXT_META_KEY.to_string(), String::new()),
+            ]));
+        let table_schema = Schema::new(vec![Field::new(
+            "docs",
+            DataType::List(Arc::new(table_item)),
+            true,
+        )]);
+
+        let input_item: FieldRef = Arc::new(Field::new("item", DataType::Utf8, 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_eq!(out_item.data_type(), &DataType::Utf8);
+        assert_eq!(
+            out_item.metadata(),
+            &HashMap::from([
+                (
+                    ARROW_EXT_NAME_KEY.to_string(),
+                    ARROW_JSON_EXT_NAME.to_string()
+                ),
+                (ARROW_EXT_META_KEY.to_string(), String::new()),
+            ])
+        );
+        assert_eq!(collect(projected).await.num_rows(), 2);
+    }
 }