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

@@ -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

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 {

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)
+	}
+}