diff --git a/seaweed-vector/Cargo.lock b/seaweed-vector/Cargo.lock index 8b9da51181..e0b6355053 100644 --- a/seaweed-vector/Cargo.lock +++ b/seaweed-vector/Cargo.lock @@ -6486,6 +6486,7 @@ dependencies = [ "clap", "futures", "lance", + "lance-arrow", "lance-linalg", "prost", "prost-types", diff --git a/seaweed-vector/crates/server/Cargo.toml b/seaweed-vector/crates/server/Cargo.toml index 7a8018aee1..71f7f449a2 100644 --- a/seaweed-vector/crates/server/Cargo.toml +++ b/seaweed-vector/crates/server/Cargo.toml @@ -16,6 +16,7 @@ path = "src/main.rs" weed-vector-core = { path = "../core" } weed-lance-worker = { path = "../../../seaweed-worker/crates/lance" } lance = { version = "10", default-features = false, features = ["aws"] } +lance-arrow = "10" lance-linalg = "10" arrow-array = "58" arrow-cast = "58" diff --git a/seaweed-vector/crates/server/src/service.rs b/seaweed-vector/crates/server/src/service.rs index e49a357ff2..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, @@ -110,8 +112,18 @@ fn batch_of(vectors: &[pb::Vector], dimension: usize, schema: Arc) -> Re .flat_map(|v| v.data.iter().copied()) .collect::>(); let values: ArrayRef = Arc::new(Float32Array::from(flat)); + // Lance stores the list item as nullable even when the write schema + // declared it non-null, so the child field must come from the batch's + // schema rather than a fixed declaration. + let DataType::FixedSizeList(item_field, _) = schema + .field_with_name(VECTOR_COLUMN) + .map_err(anyhow::Error::from)? + .data_type() + else { + bail!("vector column is not a fixed-size list"); + }; let vectors_array = FixedSizeListArray::try_new( - Field::new("item", DataType::Float32, false).into(), + item_field.clone(), dimension as i32, values, None, @@ -125,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) } diff --git a/test/vectors/run_test.sh b/test/vectors/run_test.sh index 5de199f032..cb2d72c7b3 100755 --- a/test/vectors/run_test.sh +++ b/test/vectors/run_test.sh @@ -19,6 +19,7 @@ PIDS+=($!) ./weed s3 \ -filer 127.0.0.1:8888 \ + -config s3.json \ -ip 127.0.0.1 \ -ip.bind 127.0.0.1 \ -port 8334 \ @@ -29,6 +30,8 @@ PIDS+=($!) ./weed-vector-server \ --namespace http://127.0.0.1:9102 \ + --storage-option aws_access_key_id=test \ + --storage-option aws_secret_access_key=test \ --listen 127.0.0.1:9301 & PIDS+=($!) diff --git a/test/vectors/s3.json b/test/vectors/s3.json new file mode 100644 index 0000000000..a95df73362 --- /dev/null +++ b/test/vectors/s3.json @@ -0,0 +1,20 @@ +{ + "identities": [ + { + "name": "admin", + "credentials": [ + { + "accessKey": "test", + "secretKey": "test" + } + ], + "actions": [ + "Admin", + "Read", + "List", + "Tagging", + "Write" + ] + } + ] +} diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index 3f0e8ef0b5..583f51cb90 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -788,7 +788,7 @@ func filterExpr(filter map[string]interface{}) (string, error) { } func fieldClause(field string, cond interface{}) (string, error) { - col := fmt.Sprintf("json_get_str(`metadata`, '%s')", strings.ReplaceAll(field, "'", "''")) + col := fmt.Sprintf("json_get_string(`metadata`, '%s')", strings.ReplaceAll(field, "'", "''")) switch c := cond.(type) { case map[string]interface{}: parts := []string{}