From e2d899ac2561c168f48dd63e04d55b8c4a917cf1 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 26 Oct 2026 17:52:05 -0700 Subject: [PATCH] test/vectors: fix the S3 Vectors e2e harness (#12112) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * test/vectors: give the vector server static storage credentials The e2e harness ran weed-vector-server without credentials, so lance's AWS object store fell back to IMDS and CreateIndex died on the runner's 411 from the token endpoint. Point it at the test credentials the s3 gateway now reads from s3.json. * vector: take the list item field from the schema when batching vectors Lance stores the fixed-size list item as nullable even when the write schema declared it non-null, so PutVectors built batches that no longer matched: expected FixedSizeList(4 x Float32) but found FixedSizeList(4 x non-null Float32). * 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. --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- seaweed-vector/Cargo.lock | 1 + seaweed-vector/crates/server/Cargo.toml | 1 + seaweed-vector/crates/server/src/service.rs | 49 ++++++++++++++------- test/vectors/run_test.sh | 3 ++ test/vectors/s3.json | 20 +++++++++ weed/s3api/s3vectors/handlers.go | 2 +- 6 files changed, 60 insertions(+), 16 deletions(-) create mode 100644 test/vectors/s3.json 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{}