From c55b7c5a296b67f533cd9a86472212bd407619cc Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 26 Oct 2026 15:40:25 -0700 Subject: [PATCH] s3: fix CreateIndex metadata panic and ListVectors segmented paging (#12106) * s3: nil-check metadataConfiguration before reading its keys * s3: honor maxResults/nextToken in segmented ListVectors * s3: exclude non-Lance buckets from ListVectorBuckets * s3: test segmented ListVectors paging token math * ci: silence codespell on the "$nin" operator and rename lits --- .codespellrc | 3 +- weed/s3api/s3vectors/handlers.go | 140 ++++++++++++++++------- weed/s3api/s3vectors/handlers_test.go | 154 ++++++++++++++++++++++++++ 3 files changed, 255 insertions(+), 42 deletions(-) create mode 100644 weed/s3api/s3vectors/handlers_test.go diff --git a/.codespellrc b/.codespellrc index 038fb65244..01e80413d6 100644 --- a/.codespellrc +++ b/.codespellrc @@ -20,4 +20,5 @@ ignore-regex = \b[a-z]+[A-Z]\w*\b|\b[A-Z][a-z]+[A-Z]\w*\b # unknwon: GitHub username / Go module path (`github.com/unknwon/goconfig`) # atleast: CLI mode literal string in test/benchmark/fuse_db/bin/sqlite_verify.py # sme: local variable for a *streamMutateError in mount tests -ignore-words-list = visibles,fo,te,ser,bject,unparseable,keep-alives,tread,anc,ue,auther,thirdparty,unknwon,atleast,sme +# nin: AWS S3 Vectors "$nin" filter operator literal (API-defined, cannot rename) +ignore-words-list = visibles,fo,te,ser,bject,unparseable,keep-alives,tread,anc,ue,auther,thirdparty,unknwon,atleast,sme,nin diff --git a/weed/s3api/s3vectors/handlers.go b/weed/s3api/s3vectors/handlers.go index f2189c4a44..4691e9bc8b 100644 --- a/weed/s3api/s3vectors/handlers.go +++ b/weed/s3api/s3vectors/handlers.go @@ -5,6 +5,7 @@ import ( "encoding/json" "fmt" "net/http" + "strconv" "strings" "time" @@ -196,6 +197,7 @@ func (s *Server) handleListVectorBuckets(w http.ResponseWriter, r *http.Request) Arn string `json:"arn"` Name string `json:"name"` CreatedAt string `json:"createdAt"` + Format string `json:"format"` } `json:"tableBuckets"` ContinuationToken string `json:"continuationToken"` } @@ -209,6 +211,11 @@ func (s *Server) handleListVectorBuckets(w http.ResponseWriter, r *http.Request) } out := &ListVectorBucketsResponse{NextToken: resp.ContinuationToken} for _, b := range resp.TableBuckets { + // Buckets pre-dating declared formats accept any format; only a bucket + // that declared another format is excluded here. + if b.Format != "" && b.Format != s3tables.FormatLance { + continue + } created, _ := time.Parse(time.RFC3339, b.CreatedAt) out.VectorBuckets = append(out.VectorBuckets, &VectorBucket{ VectorBucketArn: s.vectorBucketArn(b.Name), @@ -284,8 +291,10 @@ func (s *Server) handleCreateIndex(w http.ResponseWriter, r *http.Request) { "dimension": fmt.Sprint(req.Dimension), "distanceMetric": req.DistanceMetric, } - if keys, err := json.Marshal(req.MetadataConfiguration.NonFilterableMetadataKeys); req.MetadataConfiguration != nil && err == nil { - metadata["nonFilterableMetadataKeys"] = string(keys) + if req.MetadataConfiguration != nil { + if keys, err := json.Marshal(req.MetadataConfiguration.NonFilterableMetadataKeys); err == nil { + metadata["nonFilterableMetadataKeys"] = string(keys) + } } metadataJSON, _ := json.Marshal(metadata) var nsResp map[string]interface{} @@ -569,46 +578,40 @@ func (s *Server) handleListVectors(w http.ResponseWriter, r *http.Request) { catalogErr(w, err) return } - // AWS exposes a static split: segmentIndex/segmentCount divides the full - // listing. The engine service does not know AWS segments, so page through - // the whole index here and slice the caller's share — O(index) per call, - // acceptable for the MVP. - if req.SegmentCount > 0 { - var all []*vector_pb.Vector - token := "" - for { - page, err := s.vectors.ListVectors(r.Context(), &vector_pb.ListVectorsRequest{ - Table: tableID(arns.bucket, arns.index), - MaxResults: 1000, - NextToken: token, - ReturnData: req.ReturnData, - ReturnMetadata: req.ReturnMetadata, - }) - if err != nil { - writeError(w, http.StatusInternalServerError, "InternalFailure", err.Error()) - return - } - all = append(all, page.Vectors...) - if page.NextToken == "" { - break - } - token = page.NextToken - } - var out []*vector_pb.Vector - for i, v := range all { - if i%int(req.SegmentCount) == int(req.SegmentIndex) { - out = append(out, v) - } - } - writeJSON(w, http.StatusOK, &ListVectorsResponse{ - Vectors: toOutputVectors(out, req.ReturnData, req.ReturnMetadata), - }) - return - } if req.MaxResults < 0 { writeError(w, http.StatusBadRequest, "ValidationException", "maxResults must be non-negative") return } + // AWS exposes a static split: segmentIndex/segmentCount divides the full + // listing. The engine service does not know AWS segments, so page through + // the index here and keep only the caller's share — O(index) per call, + // acceptable for the MVP. The engine's next_token is a row offset, so the + // caller's token is the offset to resume scanning from. + if req.SegmentCount > 0 { + if req.SegmentIndex < 0 || req.SegmentIndex >= req.SegmentCount { + writeError(w, http.StatusBadRequest, "ValidationException", "segmentIndex must be in [0, segmentCount)") + return + } + var offset uint64 + if req.NextToken != "" { + var err error + if offset, err = strconv.ParseUint(req.NextToken, 10, 64); err != nil { + writeError(w, http.StatusBadRequest, "ValidationException", "invalid nextToken") + return + } + } + out, nextToken, err := s.listSegmentVectors(r.Context(), tableID(arns.bucket, arns.index), + offset, req.MaxResults, uint64(req.SegmentCount), uint64(req.SegmentIndex), req.ReturnData, req.ReturnMetadata) + if err != nil { + writeError(w, http.StatusInternalServerError, "InternalFailure", err.Error()) + return + } + writeJSON(w, http.StatusOK, &ListVectorsResponse{ + NextToken: nextToken, + Vectors: toOutputVectors(out, req.ReturnData, req.ReturnMetadata), + }) + return + } resp, err := s.vectors.ListVectors(r.Context(), &vector_pb.ListVectorsRequest{ Table: tableID(arns.bucket, arns.index), MaxResults: uint32(req.MaxResults), @@ -626,6 +629,61 @@ func (s *Server) handleListVectors(w http.ResponseWriter, r *http.Request) { }) } +// Mirrors the engine's page sizing so a segmented listing pages like a plain +// one does. +const ( + vectorsDefaultPage = 500 + vectorsMaxPage = 1000 +) + +// listSegmentVectors scans the index from offset, keeping the rows that fall +// in the caller's segment until limit are collected or the index runs out. +// The engine's next_token is a row offset, so the returned token is the offset +// scanning resumes from. +func (s *Server) listSegmentVectors(ctx context.Context, table *vector_pb.Table, offset uint64, maxResults int, segmentCount, segmentIndex uint64, returnData, returnMetadata bool) ([]*vector_pb.Vector, string, error) { + limit := maxResults + if limit <= 0 { + limit = vectorsDefaultPage + } else if limit > vectorsMaxPage { + limit = vectorsMaxPage + } + var out []*vector_pb.Vector + nextToken := "" + token := strconv.FormatUint(offset, 10) + for { + page, err := s.vectors.ListVectors(ctx, &vector_pb.ListVectorsRequest{ + Table: table, + MaxResults: vectorsMaxPage, + NextToken: token, + ReturnData: returnData, + ReturnMetadata: returnMetadata, + }) + if err != nil { + return nil, "", err + } + base := offset + offset += uint64(len(page.Vectors)) + full := false + for j, v := range page.Vectors { + if (base+uint64(j))%segmentCount == segmentIndex { + out = append(out, v) + if len(out) == limit { + if j+1 < len(page.Vectors) || page.NextToken != "" { + nextToken = strconv.FormatUint(base+uint64(j)+1, 10) + } + full = true + break + } + } + } + if full || len(page.Vectors) == 0 || page.NextToken == "" { + break + } + token = page.NextToken + } + return out, nextToken, nil +} + func (s *Server) handleDeleteVectors(w http.ResponseWriter, r *http.Request) { var req DeleteVectorsRequest if err := readJSON(r, &req); err != nil { @@ -729,19 +787,19 @@ func fieldClause(field string, cond interface{}) (string, error) { if !ok { return "", fmt.Errorf("%s expects a list", op) } - lits := make([]string, 0, len(list)) + terms := make([]string, 0, len(list)) for _, item := range list { lit, err := sqlLiteral(item) if err != nil { return "", err } - lits = append(lits, lit) + terms = append(terms, lit) } neg := "" if op == "$nin" { neg = "NOT " } - parts = append(parts, fmt.Sprintf("%s%s in (%s)", neg, col, strings.Join(lits, ", "))) + parts = append(parts, fmt.Sprintf("%s%s in (%s)", neg, col, strings.Join(terms, ", "))) case "$lt", "$lte", "$gt", "$gte": lit, err := sqlLiteral(operand) if err != nil { diff --git a/weed/s3api/s3vectors/handlers_test.go b/weed/s3api/s3vectors/handlers_test.go new file mode 100644 index 0000000000..648ec9c591 --- /dev/null +++ b/weed/s3api/s3vectors/handlers_test.go @@ -0,0 +1,154 @@ +package s3vectors + +import ( + "context" + "fmt" + "strconv" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/vector_pb" + "google.golang.org/grpc" +) + +// fakeVectors serves a fixed row set with the engine's offset-token paging. +type fakeVectors struct { + vector_pb.VectorServiceClient + rows []*vector_pb.Vector +} + +func newFakeVectors(n int) *fakeVectors { + f := &fakeVectors{} + for i := 0; i < n; i++ { + f.rows = append(f.rows, &vector_pb.Vector{Key: fmt.Sprintf("k%d", i)}) + } + return f +} + +func (f *fakeVectors) ListVectors(ctx context.Context, req *vector_pb.ListVectorsRequest, opts ...grpc.CallOption) (*vector_pb.ListVectorsResponse, error) { + offset := 0 + if req.NextToken != "" { + n, err := strconv.Atoi(req.NextToken) + if err != nil { + return nil, fmt.Errorf("invalid next_token %q", req.NextToken) + } + offset = n + } + page := int(req.MaxResults) + if page <= 0 { + page = vectorsDefaultPage + } + if page > vectorsMaxPage { + page = vectorsMaxPage + } + if offset > len(f.rows) { + offset = len(f.rows) + } + end := offset + page + if end > len(f.rows) { + end = len(f.rows) + } + resp := &vector_pb.ListVectorsResponse{Vectors: f.rows[offset:end]} + if rows := end - offset; rows == page && rows > 0 { + resp.NextToken = strconv.Itoa(end) + } + return resp, nil +} + +// Following every segment's tokens must visit each vector exactly once. +func TestListSegmentVectorsCoversEveryRowOnce(t *testing.T) { + f := newFakeVectors(2500) + s := &Server{vectors: f} + seen := map[string]int{} + for segmentIndex := uint64(0); segmentIndex < 3; segmentIndex++ { + token := "" + for { + offset, err := strconv.ParseUint(token, 10, 64) + if token == "" { + offset = 0 + } else if err != nil { + t.Fatalf("bad token %q: %v", token, err) + } + out, next, err := s.listSegmentVectors(context.Background(), nil, offset, 7, 3, segmentIndex, false, false) + if err != nil { + t.Fatal(err) + } + for _, v := range out { + seen[v.Key]++ + } + if next == "" { + break + } + token = next + } + } + if len(seen) != len(f.rows) { + t.Fatalf("saw %d distinct rows, want %d", len(seen), len(f.rows)) + } + for k, count := range seen { + if count != 1 { + t.Fatalf("row %s seen %d times", k, count) + } + } +} + +// A limit reached mid-page must resume from the row after the last match, not +// the next engine page. +func TestListSegmentVectorsStopsInsideEnginePage(t *testing.T) { + f := newFakeVectors(100) + s := &Server{vectors: f} + out, next, err := s.listSegmentVectors(context.Background(), nil, 0, 5, 2, 0, false, false) + if err != nil { + t.Fatal(err) + } + if len(out) != 5 { + t.Fatalf("got %d rows, want 5", len(out)) + } + // Matches are rows 0,2,4,6,8 — resume scanning at row 9. + if next != "9" { + t.Fatalf("token %q, want 9", next) + } + offset, _ := strconv.ParseUint(next, 10, 64) + out, next, err = s.listSegmentVectors(context.Background(), nil, offset, 5, 2, 0, false, false) + if err != nil { + t.Fatal(err) + } + if len(out) != 5 || out[0].Key != "k10" { + t.Fatalf("resumed at %v rows, first %v", len(out), out[0]) + } +} + +// The last row of the index filling the page exactly must not emit a token. +func TestListSegmentVectorsEndsOnLastRow(t *testing.T) { + f := newFakeVectors(10) + s := &Server{vectors: f} + out, next, err := s.listSegmentVectors(context.Background(), nil, 0, 10, 1, 0, false, false) + if err != nil { + t.Fatal(err) + } + if len(out) != 10 { + t.Fatalf("got %d rows, want 10", len(out)) + } + if next != "" { + t.Fatalf("token %q, want empty", next) + } +} + +// maxResults 0 defaults to the engine page; values past the engine cap clamp. +func TestListSegmentVectorsPageLimits(t *testing.T) { + f := newFakeVectors(1500) + s := &Server{vectors: f} + out, next, err := s.listSegmentVectors(context.Background(), nil, 0, 0, 1, 0, false, false) + if err != nil { + t.Fatal(err) + } + if len(out) != vectorsDefaultPage || next != strconv.Itoa(vectorsDefaultPage) { + t.Fatalf("default page: %d rows token %q", len(out), next) + } + out, next, err = s.listSegmentVectors(context.Background(), nil, 0, 5000, 1, 0, false, false) + if err != nil { + t.Fatal(err) + } + if len(out) != vectorsMaxPage || next != strconv.Itoa(vectorsMaxPage) { + t.Fatalf("capped page: %d rows token %q", len(out), next) + } +}