From 48f5492ddfaacd885d7c3946a8c8e341ac51f9d6 Mon Sep 17 00:00:00 2001 From: gabisonia Date: Sun, 15 Feb 2026 22:34:32 +0400 Subject: [PATCH] Prepare v0.2.2 release --- .github/workflows/release.yml | 2 +- Makefile | 26 ++++- README.md | 6 +- docs/architecture.md | 5 +- docs/stores-implementation.md | 13 ++- stores/mssql/collection.go | 166 ++++++++++++++++++++------------ stores/mssql/collection_test.go | 69 +++++++++++++ 7 files changed, 212 insertions(+), 75 deletions(-) create mode 100644 stores/mssql/collection_test.go diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 2929378..7799f9d 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -7,7 +7,7 @@ on: workflow_dispatch: inputs: version: - description: "Semver version (e.g. 0.2.1 or v0.2.1)" + description: "Semver version (e.g. 0.2.2 or v0.2.2)" required: true type: string diff --git a/Makefile b/Makefile index 97fe8ba..16920a1 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: test-integration-all test-integration-stores test-integration-postgres test-integration-mssql test-integration-msql +.PHONY: test test-no-cache test-stores test-stores-no-cache test-integration-all test-integration-all-no-cache test-integration-stores test-integration-stores-no-cache test-integration-postgres test-integration-postgres-no-cache test-integration-mssql test-integration-mssql-no-cache test-integration-msql GO ?= go GOTOOLCHAIN ?= local @@ -8,16 +8,40 @@ TEST_TIMEOUT ?= 30m TEST_FLAGS ?= GO_ENV ?= env -u GOROOT GOTOOLCHAIN=$(GOTOOLCHAIN) +test: + $(GO_ENV) $(GO) test -mod=$(MOD_MODE) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./... + +test-no-cache: + $(GO_ENV) $(GO) test -count=1 -mod=$(MOD_MODE) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./... + +test-stores: + $(GO_ENV) $(GO) test -mod=$(MOD_MODE) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./stores/... + +test-stores-no-cache: + $(GO_ENV) $(GO) test -count=1 -mod=$(MOD_MODE) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./stores/... + test-integration-all: $(GO_ENV) $(GO) test -mod=$(MOD_MODE) -tags=$(INTEGRATION_TAG) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./... +test-integration-all-no-cache: + $(GO_ENV) $(GO) test -count=1 -mod=$(MOD_MODE) -tags=$(INTEGRATION_TAG) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./... + test-integration-stores: $(GO_ENV) $(GO) test -mod=$(MOD_MODE) -tags=$(INTEGRATION_TAG) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./stores/... +test-integration-stores-no-cache: + $(GO_ENV) $(GO) test -count=1 -mod=$(MOD_MODE) -tags=$(INTEGRATION_TAG) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./stores/... + test-integration-postgres: $(GO_ENV) $(GO) test -mod=$(MOD_MODE) -tags=$(INTEGRATION_TAG) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./stores/postgres +test-integration-postgres-no-cache: + $(GO_ENV) $(GO) test -count=1 -mod=$(MOD_MODE) -tags=$(INTEGRATION_TAG) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./stores/postgres + test-integration-mssql: $(GO_ENV) $(GO) test -mod=$(MOD_MODE) -tags=$(INTEGRATION_TAG) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./stores/mssql +test-integration-mssql-no-cache: + $(GO_ENV) $(GO) test -count=1 -mod=$(MOD_MODE) -tags=$(INTEGRATION_TAG) -timeout=$(TEST_TIMEOUT) $(TEST_FLAGS) ./stores/mssql + test-integration-msql: test-integration-mssql diff --git a/README.md b/README.md index 467fb0f..c508557 100644 --- a/README.md +++ b/README.md @@ -229,12 +229,12 @@ GitHub Actions workflows are configured for: Release options: -1. Manual (recommended): run `Release` workflow via GitHub UI with `version` input (`0.2.1` or `v0.2.1`). +1. Manual (recommended): run `Release` workflow via GitHub UI with `version` input (`0.2.2` or `v0.2.2`). 2. Tag-driven: push a semver tag and the workflow publishes release notes automatically: ```bash -git tag v0.2.1 -git push origin v0.2.1 +git tag v0.2.2 +git push origin v0.2.2 ``` ## Samples diff --git a/docs/architecture.md b/docs/architecture.md index 43128bc..fdbfa0b 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -138,10 +138,11 @@ The pgvector operators are documented in the [pgvector README](https://github.co `MSSQLCollection.SearchByVector`: -1. Loads records from SQL Server +1. Streams records from SQL Server 2. Evaluates filters against records in-process 3. Computes distance in-process (cosine/l2/inner product) -4. Applies threshold, sorts by distance, returns top-k +4. Applies threshold and keeps a bounded in-memory top-k heap +5. Returns top-k sorted by distance ## 6) Filter System (AST -> SQL) diff --git a/docs/stores-implementation.md b/docs/stores-implementation.md index 85e6d4f..912014b 100644 --- a/docs/stores-implementation.md +++ b/docs/stores-implementation.md @@ -171,18 +171,18 @@ Write behavior: - Chunk in batches of 500 - Use transaction per chunk - Upsert strategy: - - attempt `UPDATE` - - if `RowsAffected == 0`, execute `INSERT` + - execute single-statement `UPDATE ... WITH (UPDLOCK, SERIALIZABLE)` + conditional `INSERT` + - avoids race conditions under concurrent upserts for the same ID Search behavior (MVP): 1. Validate `topK` and vector dimension -2. Load records from SQL table into memory +2. Stream records row-by-row from SQL 3. Evaluate filter AST in-process (`matchesFilter`) 4. Compute distance in-process (`distanceBetween`) 5. Apply optional threshold -6. Sort by distance asc (stable tie-break by `ID`) -7. Apply projection and cut to top-k +6. Maintain bounded top-k max-heap in memory (`O(k)` memory) +7. Return results sorted by distance asc (tie-break by `ID`) `EnsureIndexes` is intentionally a no-op in MSSQL backend for this MVP. @@ -197,7 +197,7 @@ Filter AST supports: Execution strategy by backend: - Postgres: compile AST -> parameterized SQL via `CompileFilterSQL` -- MSSQL: evaluate AST in Go against materialized records +- MSSQL: evaluate AST in Go against streamed records Important behavior: @@ -257,4 +257,3 @@ To add a new store backend, follow the same contract shape: 4. Support filter handling (compiled or in-process) 5. Reuse `vectordata` errors and scoring semantics 6. Add integration tests matching existing backend test coverage style - diff --git a/stores/mssql/collection.go b/stores/mssql/collection.go index d0f8b76..f67988e 100644 --- a/stores/mssql/collection.go +++ b/stores/mssql/collection.go @@ -1,6 +1,7 @@ package mssql import ( + "container/heap" "context" "database/sql" "errors" @@ -126,20 +127,18 @@ func (c *MSSQLCollection) Count(ctx context.Context, filter vectordata.Filter) ( return count, nil } - records, err := c.loadRecords(ctx, false) - if err != nil { - return 0, err - } - count := int64(0) - for _, record := range records { + if err := c.streamRecords(ctx, false, func(record vectordata.Record) error { matches, err := matchesFilter(filter, record) if err != nil { - return 0, err + return err } if matches { count++ } + return nil + }); err != nil { + return 0, err } return count, nil } @@ -154,50 +153,54 @@ func (c *MSSQLCollection) SearchByVector(ctx context.Context, vector []float32, projection := resolveProjection(opts.Projection) metric := defaultMetric(c.metric) - records, err := c.loadRecords(ctx, true) - if err != nil { - return nil, err - } - - results := make([]vectordata.SearchResult, 0, len(records)) - for _, record := range records { + topKHeap := make(searchResultMaxHeap, 0, topK) + heap.Init(&topKHeap) + if err := c.streamRecords(ctx, true, func(record vectordata.Record) error { if err := c.validateVectorDimension(record.Vector); err != nil { - return nil, fmt.Errorf("invalid stored vector for record %q: %w", record.ID, err) + return fmt.Errorf("invalid stored vector for record %q: %w", record.ID, err) } matches, err := matchesFilter(opts.Filter, record) if err != nil { - return nil, err + return err } if !matches { - continue + return nil } distance, err := distanceBetween(metric, vector, record.Vector) if err != nil { - return nil, err + return err } if opts.Threshold != nil && distance > *opts.Threshold { - continue + return nil } - results = append(results, vectordata.SearchResult{ + candidate := vectordata.SearchResult{ Record: projectRecord(record, projection), Distance: distance, Score: vectordata.ScoreFromDistance(metric, distance), - }) - } + } - sort.Slice(results, func(i, j int) bool { - if results[i].Distance == results[j].Distance { - return results[i].Record.ID < results[j].Record.ID + if topKHeap.Len() < topK { + heap.Push(&topKHeap, candidate) + return nil + } + worst := topKHeap[0] + if isBetterResult(candidate, worst) { + heap.Pop(&topKHeap) + heap.Push(&topKHeap, candidate) } - return results[i].Distance < results[j].Distance - }) + return nil + }); err != nil { + return nil, err + } - if len(results) > topK { - results = results[:topK] + results := make([]vectordata.SearchResult, 0, topKHeap.Len()) + for topKHeap.Len() > 0 { + results = append(results, heap.Pop(&topKHeap).(vectordata.SearchResult)) } + sort.Slice(results, func(i, j int) bool { return isBetterResult(results[i], results[j]) }) return results, nil } @@ -221,13 +224,7 @@ func (c *MSSQLCollection) writeRecords(ctx context.Context, records []vectordata quoteIdent(metadataColumn), quoteIdent(contentColumn), ) - updateQuery := fmt.Sprintf("UPDATE %s SET %s = @p2, %s = @p3, %s = @p4 WHERE %s = @p1", - c.tableName(), - quoteIdent(vectorColumn), - quoteIdent(metadataColumn), - quoteIdent(contentColumn), - quoteIdent(idColumn), - ) + upsertQuery := buildUpsertQuery(c.tableName()) for start := 0; start < len(records); start += maxRowsPerStatement { end := start + maxRowsPerStatement @@ -239,7 +236,7 @@ func (c *MSSQLCollection) writeRecords(ctx context.Context, records []vectordata if err != nil { return err } - if err := c.writeBatch(ctx, tx, records[start:end], mode, insertQuery, updateQuery); err != nil { + if err := c.writeBatch(ctx, tx, records[start:end], mode, insertQuery, upsertQuery); err != nil { _ = tx.Rollback() return err } @@ -258,7 +255,7 @@ func (c *MSSQLCollection) writeBatch( records []vectordata.Record, mode writeMode, insertQuery string, - updateQuery string, + upsertQuery string, ) error { for _, record := range records { if strings.TrimSpace(record.ID) == "" { @@ -288,20 +285,9 @@ func (c *MSSQLCollection) writeBatch( return err } case writeModeUpsert: - result, err := tx.ExecContext(ctx, updateQuery, record.ID, vectorPayload, metadataPayload, contentArg) - if err != nil { - return err - } - - rowsAffected, err := result.RowsAffected() - if err != nil { + if _, err := tx.ExecContext(ctx, upsertQuery, record.ID, vectorPayload, metadataPayload, contentArg); err != nil { return err } - if rowsAffected == 0 { - if _, err := tx.ExecContext(ctx, insertQuery, record.ID, vectorPayload, metadataPayload, contentArg); err != nil { - return err - } - } default: return fmt.Errorf("unsupported write mode %d", mode) } @@ -310,7 +296,7 @@ func (c *MSSQLCollection) writeBatch( return nil } -func (c *MSSQLCollection) loadRecords(ctx context.Context, includeVector bool) ([]vectordata.Record, error) { +func (c *MSSQLCollection) streamRecords(ctx context.Context, includeVector bool, yield func(vectordata.Record) error) error { selectColumns := []string{quoteIdent(idColumn)} if includeVector { selectColumns = append(selectColumns, quoteIdent(vectorColumn)) @@ -320,11 +306,10 @@ func (c *MSSQLCollection) loadRecords(ctx context.Context, includeVector bool) ( query := fmt.Sprintf("SELECT %s FROM %s", strings.Join(selectColumns, ", "), c.tableName()) rows, err := c.store.db.QueryContext(ctx, query) if err != nil { - return nil, err + return err } defer rows.Close() - records := make([]vectordata.Record, 0) for rows.Next() { var record vectordata.Record var vectorRaw string @@ -333,22 +318,22 @@ func (c *MSSQLCollection) loadRecords(ctx context.Context, includeVector bool) ( if includeVector { if err := rows.Scan(&record.ID, &vectorRaw, &metadataRaw, &content); err != nil { - return nil, err + return err } parsedVector, err := parseVectorJSON(vectorRaw) if err != nil { - return nil, fmt.Errorf("decode vector: %w", err) + return fmt.Errorf("decode vector: %w", err) } record.Vector = parsedVector } else { if err := rows.Scan(&record.ID, &metadataRaw, &content); err != nil { - return nil, err + return err } } metadata, err := parseMetadataJSON(metadataRaw) if err != nil { - return nil, fmt.Errorf("decode metadata: %w", err) + return fmt.Errorf("decode metadata: %w", err) } record.Metadata = metadata @@ -357,13 +342,14 @@ func (c *MSSQLCollection) loadRecords(ctx context.Context, includeVector bool) ( record.Content = &value } - records = append(records, record) + if err := yield(record); err != nil { + return err + } } if err := rows.Err(); err != nil { - return nil, err + return err } - - return records, nil + return nil } func (c *MSSQLCollection) validateVectorDimension(vector []float32) error { @@ -395,3 +381,61 @@ func projectRecord(record vectordata.Record, projection vectordata.Projection) v } return projected } + +// buildUpsertQuery uses key-range locks so concurrent upserts on the same ID remain atomic. +func buildUpsertQuery(tableName string) string { + return fmt.Sprintf(`UPDATE %s WITH (UPDLOCK, SERIALIZABLE) +SET %s = @p2, %s = @p3, %s = @p4 +WHERE %s = @p1; +IF @@ROWCOUNT = 0 +BEGIN + INSERT INTO %s (%s, %s, %s, %s) VALUES (@p1, @p2, @p3, @p4); +END`, + tableName, + quoteIdent(vectorColumn), + quoteIdent(metadataColumn), + quoteIdent(contentColumn), + quoteIdent(idColumn), + tableName, + quoteIdent(idColumn), + quoteIdent(vectorColumn), + quoteIdent(metadataColumn), + quoteIdent(contentColumn), + ) +} + +type searchResultMaxHeap []vectordata.SearchResult + +func (h *searchResultMaxHeap) Len() int { return len(*h) } + +func (h *searchResultMaxHeap) Less(i, j int) bool { + return isWorseResult((*h)[i], (*h)[j]) +} + +func (h *searchResultMaxHeap) Swap(i, j int) { (*h)[i], (*h)[j] = (*h)[j], (*h)[i] } + +func (h *searchResultMaxHeap) Push(x any) { + *h = append(*h, x.(vectordata.SearchResult)) +} + +func (h *searchResultMaxHeap) Pop() any { + old := *h + n := len(old) + item := old[n-1] + *h = old[:n-1] + return item +} + +func isBetterResult(left, right vectordata.SearchResult) bool { + if left.Distance == right.Distance { + return left.Record.ID < right.Record.ID + } + return left.Distance < right.Distance +} + +func isWorseResult(left, right vectordata.SearchResult) bool { + if left.Distance == right.Distance { + return left.Record.ID > right.Record.ID + } + return left.Distance > right.Distance +} diff --git a/stores/mssql/collection_test.go b/stores/mssql/collection_test.go new file mode 100644 index 0000000..bf8d1dd --- /dev/null +++ b/stores/mssql/collection_test.go @@ -0,0 +1,69 @@ +package mssql + +import ( + "container/heap" + "sort" + "strings" + "testing" + + "github.com/gabisonia/go-vectorstore/vectordata" +) + +func TestBuildUpsertQueryUsesLockingPattern(t *testing.T) { + query := buildUpsertQuery("[dbo].[docs]") + + if !strings.Contains(query, "WITH (UPDLOCK, SERIALIZABLE)") { + t.Fatalf("expected upsert query to use locking hint, got: %s", query) + } + if !strings.Contains(query, "IF @@ROWCOUNT = 0") { + t.Fatalf("expected upsert query to insert on miss, got: %s", query) + } + if !strings.Contains(query, "INSERT INTO [dbo].[docs]") { + t.Fatalf("expected upsert query to target provided table, got: %s", query) + } +} + +func TestSearchResultHeapKeepsBestTopK(t *testing.T) { + const topK = 3 + + h := make(searchResultMaxHeap, 0, topK) + heap.Init(&h) + + pushTopK := func(id string, distance float64) { + candidate := vectordata.SearchResult{ + Record: vectordata.Record{ID: id}, + Distance: distance, + } + if h.Len() < topK { + heap.Push(&h, candidate) + return + } + if isBetterResult(candidate, h[0]) { + heap.Pop(&h) + heap.Push(&h, candidate) + } + } + + pushTopK("x", 0.40) + pushTopK("a", 0.10) + pushTopK("b", 0.20) + pushTopK("c", 0.20) // Same distance as b; tie-break should keep b. + pushTopK("d", 0.05) + + results := make([]vectordata.SearchResult, 0, topK) + for h.Len() > 0 { + results = append(results, heap.Pop(&h).(vectordata.SearchResult)) + } + sort.Slice(results, func(i, j int) bool { return isBetterResult(results[i], results[j]) }) + + if len(results) != topK { + t.Fatalf("expected %d results, got %d", topK, len(results)) + } + + expected := []string{"d", "a", "b"} + for i := range expected { + if results[i].Record.ID != expected[i] { + t.Fatalf("unexpected result order at %d: expected %q got %q", i, expected[i], results[i].Record.ID) + } + } +}