test: boto3 s3vectors conformance harness (P1) (#12105)

* vector: boto3 s3vectors conformance harness test/vectors spins a mini cluster, an s3 gateway with -port.vectors and -vectors.grpc, and a weed-vector-server, then exercises the AWS client end to end: bucket and index lifecycle, put/get/list/delete vectors, and a filtered query asserting distances and metadata. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ci: run the s3vectors conformance subset An e2e job in rust-vector-tests: builds weed and weed-vector-server, then runs test/vectors against a mini cluster plus the vectors gateway. The workflow now also watches vector.proto, its stubs, and test/vectors. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ci: fix e2e protoc and codespell findings The e2e job needed the same vendored-protoc step as the unit-test job — lance's build scripts locate protoc themselves. Codespell flags the AWS `$nin` operator literal and the `lits` variable: exempt `nin` in .codespellrc like the existing word exemptions, rename the variable, and fold the negation into the IN/NOT IN symbol directly. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * s3: classify catalog errors by code, not message text * s3: drop the implicit vector namespace on vector-bucket delete * test: point the vector engine at the gateway lance port * test: bind the vectors gateway listeners to loopback * test: make the filter and distance assertions load-bearing * test: rebuild the harness binaries on every run * ci: fix the vectors end-to-end job * ci: run the vector tests on gateway and command changes --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-10-27 07:20:44 +01:00 · 2026-10-26 15:53:50 -07:00
1 parent c55b7c5a29
commit a9d6bfc51f
@@ -8,6 +8,11 @@ on:
      - 'seaweed-worker/**'
      - 'seaweed-common/**'
      - 'weed/pb/plugin.proto'
+      - 'weed/pb/vector.proto'
+      - 'weed/pb/vector_pb/**'
+      - 'weed/s3api/**'
+      - 'weed/command/**'
+      - 'test/vectors/**'
      - '.github/workflows/rust-vector-tests.yml'
  push:
    branches: [ master, main ]
@@ -16,6 +21,11 @@ on:
      - 'seaweed-worker/**'
      - 'seaweed-common/**'
      - 'weed/pb/plugin.proto'
+      - 'weed/pb/vector.proto'
+      - 'weed/pb/vector_pb/**'
+      - 'weed/s3api/**'
+      - 'weed/command/**'
+      - 'test/vectors/**'
      - '.github/workflows/rust-vector-tests.yml'

 concurrency:
@@ -74,3 +84,41 @@ jobs:

      - name: Run unit tests
        run: cd seaweed-vector && cargo test --release --workspace
+
+
+  vectors-e2e:
+    name: S3 Vectors End-to-End
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          persist-credentials: false
+      - uses: actions/setup-go@v5
+        with:
+          go-version-file: go.mod
+      - name: Build weed
+        run: go build -o test/vectors/weed ./weed
+      - uses: dtolnay/rust-toolchain@stable
+      # Same as the unit-test job: lance's build scripts need a protoc; use
+      # the vendored one so no system package is required.
+      - name: Use the vendored protoc
+        run: |
+          cd seaweed-vector
+          cargo fetch
+          version=$(awk '/^name = "protoc-bin-vendored-linux-x86_64"$/{found=1; next} found && /^version = /{gsub(/"/,"",$3); print $3; exit}' Cargo.lock)
+          test -n "$version" || { echo "protoc-bin-vendored-linux-x86_64 is not in Cargo.lock" >&2; exit 1; }
+          protoc=$(find ~/.cargo/registry/src -path "*protoc-bin-vendored-linux-x86_64-$version/bin/protoc" | head -1)
+          test -x "$protoc" || { echo "no vendored protoc $version in the registry" >&2; exit 1; }
+          echo "PROTOC=$protoc" >> "$GITHUB_ENV"
+      - name: Build weed-vector-server
+        run: cargo build --release --manifest-path seaweed-vector/Cargo.toml -p weed-vector-server
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+      - name: Install test deps
+        run: pip install boto3 pytest
+      - name: Run the s3vectors conformance subset
+        working-directory: test/vectors
+        run: |
+          cp ../../seaweed-vector/target/release/weed-vector-server weed-vector-server
+          ./run_test.sh
@@ -0,0 +1,28 @@
+# End-to-end S3 Vectors conformance run:
+# weed mini (cluster + lance catalog) + weed s3 (s3vectors port) +
+# weed-vector-server (VectorService gRPC), then boto3 s3vectors calls.
+WEED := ./weed
+VEC := ./weed-vector-server
+
+.PHONY: all test build clean FORCE
+
+all: test
+
+# No prerequisites: go/cargo decide for themselves whether a rebuild is needed.
+$(WEED): FORCE
+	go build -o $(WEED) ../../weed
+
+$(VEC): FORCE
+	cargo build --release --manifest-path ../../seaweed-vector/Cargo.toml -p weed-vector-server
+	cp ../../seaweed-vector/target/release/weed-vector-server $(VEC)
+
+FORCE:
+
+build: $(WEED) $(VEC)
+
+test: build
+	./run_test.sh
+
+clean:
+	rm -f $(WEED) $(VEC)
+	rm -rf mini-data
@@ -0,0 +1,49 @@
+#!/usr/bin/env bash
+# Spin a mini cluster plus the vectors gateway and engine, run the boto3
+# conformance subset, tear everything down.
+set -euo pipefail
+cd "$(dirname "$0")"
+
+DATA=mini-data
+rm -rf "$DATA"
+PIDS=()
+cleanup() {
+  kill "${PIDS[@]}" 2>/dev/null || true
+  rm -rf "$DATA"
+}
+trap cleanup EXIT
+
+./weed mini -dir "$DATA" \
+  -s3=false -webdav=false -admin.ui=false &
+PIDS+=($!)
+
+./weed s3 \
+  -filer 127.0.0.1:8888 \
+  -ip 127.0.0.1 \
+  -ip.bind 127.0.0.1 \
+  -port 8334 \
+  -port.lance 9102 \
+  -port.vectors 9201 \
+  -vectors.grpc 127.0.0.1:9301 &
+PIDS+=($!)
+
+./weed-vector-server \
+  --namespace http://127.0.0.1:9102 \
+  --listen 127.0.0.1:9301 &
+PIDS+=($!)
+
+# Any HTTP response (even an auth rejection) means the lance catalog is up.
+for _ in $(seq 1 60); do
+  code=$(curl -s -o /dev/null -w '%{http_code}' http://127.0.0.1:9102/v1/namespace/root/list || true)
+  [ "$code" != "000" ] && break
+  sleep 1
+done
+[ "${code:-000}" != "000" ] || { echo "lance catalog on :9102 never came up" >&2; exit 1; }
+
+for _ in $(seq 1 60); do
+  (exec 3<>/dev/tcp/127.0.0.1/9301) 2>/dev/null && break
+  sleep 1
+done
+(exec 3<>/dev/tcp/127.0.0.1/9301) 2>/dev/null || { echo "vector server on :9301 never came up" >&2; exit 1; }
+
+python3 -m pytest -xvs s3vectors_test.py
@@ -0,0 +1,81 @@
+"""Boto3 conformance subset for the S3 Vectors API gateway.
+
+Run via `make test` — a mini cluster, an s3 gateway with -port.vectors, and a
+weed-vector-server are up by then.
+"""
+
+import boto3
+import pytest
+
+ENDPOINT = "http://127.0.0.1:9201"
+BUCKET = "vectors-bucket"
+INDEX = "idx"
+
+
+@pytest.fixture(scope="module")
+def vectors():
+    return boto3.client(
+        "s3vectors",
+        endpoint_url=ENDPOINT,
+        aws_access_key_id="test",
+        aws_secret_access_key="test",
+        region_name="us-east-1",
+    )
+
+
+def test_vector_lifecycle(vectors):
+    vectors.create_vector_bucket(vectorBucketName=BUCKET)
+    buckets = [b["vectorBucketName"] for b in vectors.list_vector_buckets()["vectorBuckets"]]
+    assert BUCKET in buckets
+
+    vectors.create_index(
+        vectorBucketName=BUCKET,
+        indexName=INDEX,
+        dataType="float32",
+        dimension=4,
+        distanceMetric="euclidean",
+    )
+    idx = vectors.get_index(vectorBucketName=BUCKET, indexName=INDEX)["index"]
+    assert idx["dimension"] == 4
+
+    vectors.put_vectors(
+        vectorBucketName=BUCKET,
+        indexName=INDEX,
+        vectors=[
+            {"key": "a", "data": {"float32": [1, 0, 0, 0]}, "metadata": {"color": "red"}},
+            {"key": "b", "data": {"float32": [0.95, 0.05, 0, 0]}, "metadata": {"color": "blue"}},
+            {"key": "c", "data": {"float32": [0.9, 0.1, 0, 0]}, "metadata": {"color": "red"}},
+        ],
+    )
+
+    got = vectors.get_vectors(
+        vectorBucketName=BUCKET, indexName=INDEX, keys=["a"], returnData=True, returnMetadata=True
+    )["vectors"]
+    assert got[0]["key"] == "a" and got[0]["metadata"]["color"] == "red"
+
+    listed = vectors.list_vectors(vectorBucketName=BUCKET, indexName=INDEX)["vectors"]
+    assert sorted(v["key"] for v in listed) == ["a", "b", "c"]
+
+    hits = vectors.query_vectors(
+        vectorBucketName=BUCKET,
+        indexName=INDEX,
+        queryVector={"float32": [1, 0, 0, 0]},
+        topK=2,
+        returnDistance=True,
+        returnMetadata=True,
+        filter={"color": {"$eq": "red"}},
+    )["vectors"]
+    # b is the second-nearest vector but blue, so an ignored filter would
+    # return ["a", "b"] here.
+    assert [h["key"] for h in hits] == ["a", "c"]
+    distances = {h["key"]: h["distance"] for h in hits}
+    assert distances["a"] == pytest.approx(0.0, abs=1e-6)
+    assert distances["c"] == pytest.approx(0.02, abs=1e-4)
+    assert all(h["metadata"]["color"] == "red" for h in hits)
+
+    vectors.delete_vectors(vectorBucketName=BUCKET, indexName=INDEX, keys=["b"])
+    listed = vectors.list_vectors(vectorBucketName=BUCKET, indexName=INDEX)["vectors"]
+    assert sorted(v["key"] for v in listed) == ["a", "c"]
+
+    vectors.delete_index(vectorBucketName=BUCKET, indexName=INDEX)
+    vectors.delete_vector_bucket(vectorBucketName=BUCKET)
@@ -3,6 +3,7 @@ package s3vectors
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"strconv"
@@ -119,21 +120,42 @@ func (s *Server) catalog(ctx context.Context, operation string, req, resp interf
 	return s.tables.Execute(ctx, s.filerClient, operation, req, resp, identityOf(r))
 }

+// catalogErrCode returns the error code the in-process catalog reports;
+// errors from other sources carry no code.
+func catalogErrCode(err error) string {
+	var e *s3tables.S3TablesError
+	if errors.As(err, &e) {
+		return e.Type
+	}
+	return ""
+}
+
 func isNotFound(err error) bool {
+	switch catalogErrCode(err) {
+	case s3tables.ErrCodeNoSuchBucket, s3tables.ErrCodeNoSuchTable, s3tables.ErrCodeNoSuchNamespace:
+		return true
+	}
 	return err != nil && (strings.Contains(err.Error(), "NoSuchTableBucket") ||
 		strings.Contains(err.Error(), "NoSuchTable") ||
 		strings.Contains(err.Error(), "NotFoundException") ||
 		strings.Contains(err.Error(), "NoSuchBucket"))
 }

+func isAlreadyExists(err error) bool {
+	code := catalogErrCode(err)
+	return strings.HasSuffix(code, "AlreadyExists") || code == s3tables.ErrCodeConflict ||
+		(err != nil && (strings.Contains(err.Error(), "AlreadyExists") || strings.Contains(err.Error(), "Conflict")))
+}
+
 func catalogErr(w http.ResponseWriter, err error) {
 	msg := err.Error()
 	switch {
 	case isNotFound(err):
 		writeError(w, http.StatusNotFound, "NotFoundException", msg)
-	case strings.Contains(msg, "AccessDenied") || strings.Contains(msg, "PermissionDenied"):
+	case catalogErrCode(err) == s3tables.ErrCodeAccessDenied ||
+		strings.Contains(msg, "AccessDenied") || strings.Contains(msg, "PermissionDenied"):
 		writeError(w, http.StatusForbidden, "AccessDeniedException", "access denied")
-	case strings.Contains(msg, "AlreadyExists") || strings.Contains(msg, "Conflict"):
+	case isAlreadyExists(err):
 		writeError(w, http.StatusConflict, "ConflictException", msg)
 	default:
 		writeError(w, http.StatusBadRequest, "ValidationException", msg)
@@ -238,6 +260,12 @@ func (s *Server) handleDeleteVectorBucket(w http.ResponseWriter, r *http.Request
 		return
 	}
 	var resp map[string]interface{}
+	// Indexes live in an implicit namespace; dropping it best-effort lets an
+	// emptied bucket delete, while surviving indexes keep it non-empty.
+	_ = s.catalog(r.Context(), "DeleteNamespace", &s3tables.DeleteNamespaceRequest{
+		TableBucketARN: s.s3tablesArn(arns.bucket),
+		Namespace:      []string{indexNamespace},
+	}, &resp, r)
 	err := s.catalog(r.Context(), "DeleteTableBucket",
 		&s3tables.DeleteTableBucketRequest{TableBucketARN: s.s3tablesArn(arns.bucket)}, &resp, r)
 	if err != nil {
@@ -303,7 +331,7 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) {
 	if err := s.catalog(r.Context(), "CreateNamespace", &s3tables.CreateNamespaceRequest{
 		TableBucketARN: s.s3tablesArn(arns.bucket),
 		Namespace:      []string{indexNamespace},
-	}, &nsResp, r); err != nil && !strings.Contains(err.Error(), "AlreadyExists") && !strings.Contains(err.Error(), "Conflict") {
+	}, &nsResp, r); err != nil && !isAlreadyExists(err) {
 		catalogErr(w, err)
 		return
 	}
@@ -787,19 +815,19 @@ func fieldClause(field string, cond interface{}) (string, error) {
 				if !ok {
 					return "", fmt.Errorf("%s expects a list", op)
 				}
-				terms := make([]string, 0, len(list))
+				vals := make([]string, 0, len(list))
 				for _, item := range list {
 					lit, err := sqlLiteral(item)
 					if err != nil {
 						return "", err
 					}
-					terms = append(terms, lit)
+					vals = append(vals, lit)
 				}
-				neg := ""
+				sym := "IN"
 				if op == "$nin" {
-					neg = "NOT "
+					sym = "NOT IN"
 				}
-				parts = append(parts, fmt.Sprintf("%s%s in (%s)", neg, col, strings.Join(terms, ", ")))
+				parts = append(parts, fmt.Sprintf("%s %s (%s)", col, sym, strings.Join(vals, ", ")))
 			case "$lt", "$lte", "$gt", "$gte":
 				lit, err := sqlLiteral(operand)
 				if err != nil {