vector: store index metadata as lance jsonb so filters resolve

The metadata field was plain Utf8, but the filter functions lance's
planner registers (json_get_string, json_get_float, json_exists) take a
LargeBinary jsonb column, so QueryVectors with a metadata filter failed
to plan. Declare the field with the arrow.json extension — lance encodes
it to jsonb on write and back on read — and encode the batch's strings
when the stored column is LargeBinary. Also rename the filter's
json_get_str call to json_get_string, the name lance registers.

seaweed-vector/crates/server/src/service.rs

@@ -10,7 +10,7 @@ use lance::io::ObjectStoreParams;
 use lance::io::StorageOptionsAccessor;
 use lance_linalg::distance::DistanceType;
 use prost_types::{ListValue, Struct, Value, value::Kind};
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, HashMap};
 use std::sync::Arc;
 use tonic::{Request, Response, Status};
 use weed_lance_worker::catalog::NamespaceClient;
@@ -74,17 +74,19 @@ fn schema_for(
         .into_iter()
         .collect(),
     );
-    let mut metadata_field = Field::new(METADATA_COLUMN, DataType::Utf8, true);
+    // The arrow.json extension marks the Utf8 column as JSON: lance stores it
+    // as jsonb, which is what the json_get_* filter functions take.
+    let mut metadata_meta = HashMap::from_iter([(
+        "ARROW:extension:name".to_string(),
+        "arrow.json".to_string(),
+    )]);
     if !non_filterable.is_empty() {
-        metadata_field = metadata_field.with_metadata(
-            [(
-                "non_filterable_metadata_keys".to_string(),
-                serde_json::to_string(non_filterable).unwrap_or_default(),
-            )]
-            .into_iter()
-            .collect(),
+        metadata_meta.insert(
+            "non_filterable_metadata_keys".to_string(),
+            serde_json::to_string(non_filterable).unwrap_or_default(),
         );
     }
+    let metadata_field = Field::new(METADATA_COLUMN, DataType::Utf8, true).with_metadata(metadata_meta);
     Arc::new(Schema::new(vec![
         Field::new(KEY_COLUMN, DataType::Utf8, false),
         vector_field,
@@ -135,13 +137,20 @@ fn batch_of(vectors: &[pb::Vector], dimension: usize, schema: Arc<Schema>) -> Re
                 .transpose()
         })
         .collect::<Result<Vec<_>, _>>()?;
+    // The stored column is lance's jsonb (LargeBinary) once the field carries
+    // the arrow.json extension; encode the strings, else keep the Utf8 batch.
+    let metadata_array: ArrayRef = match schema
+        .field_with_name(METADATA_COLUMN)
+        .map(|f| f.data_type())
+    {
+        Ok(DataType::LargeBinary) => {
+            Arc::new(lance_arrow::json::JsonArray::try_from_iter(metas)?.into_inner())
+        }
+        _ => Arc::new(StringArray::from(metas)),
+    };
     RecordBatch::try_new(
         schema,
-        vec![
-            Arc::new(keys),
-            Arc::new(vectors_array),
-            Arc::new(StringArray::from(metas)),
-        ],
+        vec![Arc::new(keys), Arc::new(vectors_array), metadata_array],
     )
     .map_err(Into::into)
 }