diff --git a/seaweed-vector/crates/server/src/service.rs b/seaweed-vector/crates/server/src/service.rs index e39882e0c8..094eb932f8 100644 --- a/seaweed-vector/crates/server/src/service.rs +++ b/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) -> Re .transpose() }) .collect::, _>>()?; + // 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) }