Stop Badger GC before closing datastore

Co-authored-by: Pijus Kamandulis <pikami@users.noreply.github.com>

api/handlers/documents.go

@@ -47,6 +47,9 @@ func (h *Handlers) GetDocument(c *gin.Context) {
 
 	document, status := h.dataStore.GetDocument(databaseId, collectionId, documentId)
 	if status == datastore.StatusOk {
+		if etag, ok := document["_etag"].(string); ok {
+			c.Header(headers.ETag, etag)
+		}
 		c.IndentedJSON(http.StatusOK, document)
 		return
 	}
@@ -90,7 +93,25 @@ func (h *Handlers) ReplaceDocument(c *gin.Context) {
 		return
 	}
 
-	status := h.dataStore.DeleteDocument(databaseId, collectionId, documentId)
+	existingDocument, status := h.dataStore.GetDocument(databaseId, collectionId, documentId)
+	if status == datastore.StatusNotFound {
+		c.IndentedJSON(http.StatusNotFound, constants.NotFoundResponse)
+		return
+	}
+	if status != datastore.StatusOk {
+		c.IndentedJSON(http.StatusInternalServerError, constants.UnknownErrorResponse)
+		return
+	}
+
+	if ifMatch := c.GetHeader(headers.IfMatch); ifMatch != "" {
+		if existingDocument["_etag"] != ifMatch {
+			c.Header(headers.ErrorCode, "PreconditionFailed")
+			c.JSON(http.StatusPreconditionFailed, constants.PreconditionFailedResponse)
+			return
+		}
+	}
+
+	status = h.dataStore.DeleteDocument(databaseId, collectionId, documentId)
 	if status == datastore.StatusNotFound {
 		c.IndentedJSON(http.StatusNotFound, constants.NotFoundResponse)
 		return

api/headers/headers.go

@@ -4,8 +4,10 @@ const (
 	AIM                = "A-Im"
 	Authorization      = "authorization"
 	CosmosLsn          = "x-ms-cosmos-llsn"
+	ErrorCode          = "x-ms-error-code"
 	ETag               = "etag"
 	GlobalCommittedLsn = "x-ms-global-committed-lsn"
+	IfMatch            = "if-match"
 	IfNoneMatch        = "if-none-match"
 	IsBatchRequest     = "x-ms-cosmos-is-batch-request"
 	IsQueryPlanRequest = "x-ms-cosmos-is-query-plan-request"

api/tests/documents_array_functions_test.go

@@ -0,0 +1,47 @@
+package tests_test
+
+import (
+	"testing"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos"
+)
+
+func Test_Documents_ArrayContains(t *testing.T) {
+	presets := []testPreset{PresetJsonStore, PresetBadgerStore}
+
+	runTestsWithPresets(t, "Test_Documents_ArrayContains", presets, func(t *testing.T, ts *TestServer, client *azcosmos.Client) {
+		collectionClient := documents_InitializeDb(t, ts)
+
+		t.Run("Should execute ARRAY_CONTAINS() without partial match argument", func(t *testing.T) {
+			testCosmosQuery(t, collectionClient,
+				`SELECT VALUE ARRAY_CONTAINS(["apple", "banana", "cherry"], "banana") FROM c ORDER BY c.id`,
+				nil,
+				[]interface{}{true, true},
+			)
+		})
+
+		t.Run("Should execute ARRAY_CONTAINS() returning false for missing item", func(t *testing.T) {
+			testCosmosQuery(t, collectionClient,
+				`SELECT VALUE ARRAY_CONTAINS(["apple", "banana", "cherry"], "grape") FROM c ORDER BY c.id`,
+				nil,
+				[]interface{}{false, false},
+			)
+		})
+
+		t.Run("Should execute ARRAY_CONTAINS() with object full match", func(t *testing.T) {
+			testCosmosQuery(t, collectionClient,
+				`SELECT VALUE ARRAY_CONTAINS([{"name": "apple", "color": "red"}], {"name": "apple"}) FROM c ORDER BY c.id`,
+				nil,
+				[]interface{}{false, false},
+			)
+		})
+
+		t.Run("Should execute ARRAY_CONTAINS() with object partial match", func(t *testing.T) {
+			testCosmosQuery(t, collectionClient,
+				`SELECT VALUE ARRAY_CONTAINS([{"name": "apple", "color": "red"}], {"name": "apple"}, true) FROM c ORDER BY c.id`,
+				nil,
+				[]interface{}{true, true},
+			)
+		})
+	})
+}

api/tests/documents_string_functions_test.go

@@ -0,0 +1,73 @@
+package tests_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/Azure/azure-sdk-for-go/sdk/data/azcosmos"
+	"github.com/pikami/cosmium/api/config"
+	"github.com/pikami/cosmium/internal/datastore"
+	"github.com/stretchr/testify/assert"
+)
+
+func documents_InitializeSingleDocumentDb(t *testing.T, ts *TestServer) *azcosmos.ContainerClient {
+	ts.DataStore.CreateDatabase(datastore.Database{ID: testDatabaseName})
+	ts.DataStore.CreateCollection(testDatabaseName, datastore.Collection{
+		ID: testCollectionName,
+		PartitionKey: struct {
+			Paths   []string "json:\"paths\""
+			Kind    string   "json:\"kind\""
+			Version int      "json:\"Version\""
+		}{
+			Paths: []string{"/pk"},
+		},
+	})
+	ts.DataStore.CreateDocument(testDatabaseName, testCollectionName, map[string]interface{}{"id": "regexmatch-test", "pk": "regexmatch-test"})
+
+	client, err := azcosmos.NewClientFromConnectionString(
+		fmt.Sprintf("AccountEndpoint=%s;AccountKey=%s", ts.URL, config.DefaultAccountKey),
+		&azcosmos.ClientOptions{},
+	)
+	assert.Nil(t, err)
+
+	collectionClient, err := client.NewContainer(testDatabaseName, testCollectionName)
+	assert.Nil(t, err)
+
+	return collectionClient
+}
+
+func Test_Documents_RegexMatch(t *testing.T) {
+	presets := []testPreset{PresetJsonStore, PresetBadgerStore}
+
+	runTestsWithPresets(t, "Test_Documents_RegexMatch", presets, func(t *testing.T, ts *TestServer, client *azcosmos.Client) {
+		collectionClient := documents_InitializeSingleDocumentDb(t, ts)
+
+		t.Run("Should execute REGEXMATCH()", func(t *testing.T) {
+			testCosmosQuery(t, collectionClient,
+				`SELECT VALUE {
+					noModifiers: REGEXMATCH("abcd", "ABC"),
+					caseInsensitive: REGEXMATCH("abcd", "ABC", "i"),
+					wildcardCharacter: REGEXMATCH("abcd", "ab.", ""),
+					ignoreWhiteSpace: REGEXMATCH("abcd", "ab c", "x"),
+					caseInsensitiveAndIgnoreWhiteSpace: REGEXMATCH("abcd", "aB c", "ix"),
+					containNumberBetweenZeroAndNine: REGEXMATCH("03a", "[0-9]"),
+					containPrefix: REGEXMATCH("salt3824908", "salt{1}"),
+					containsFiveLetterWordStartingWithS: REGEXMATCH("shame", "s....", "i")
+				}`,
+				nil,
+				[]interface{}{
+					map[string]interface{}{
+						"noModifiers":                         false,
+						"caseInsensitive":                     true,
+						"wildcardCharacter":                   true,
+						"ignoreWhiteSpace":                    true,
+						"caseInsensitiveAndIgnoreWhiteSpace":  true,
+						"containNumberBetweenZeroAndNine":     true,
+						"containPrefix":                       true,
+						"containsFiveLetterWordStartingWithS": true,
+					},
+				},
+			)
+		})
+	})
+}

api/tests/documents_test.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"reflect"
 	"sync"
@@ -379,6 +380,80 @@ func Test_Documents(t *testing.T) {
 		})
 	})
 
+	runTestsWithPresets(t, "Test_Documents_ETag_OptimisticConcurrency", presets, func(t *testing.T, ts *TestServer, client *azcosmos.Client) {
+		collectionClient := documents_InitializeDb(t, ts)
+
+		t.Run("Should fail replace with incorrect etag", func(t *testing.T) {
+			context := context.TODO()
+
+			item := map[string]interface{}{"id": "12345", "pk": "123", "isCool": true}
+			bytes, err := json.Marshal(item)
+			assert.Nil(t, err)
+
+			wrongETag := azcore.ETag("\"incorrect-etag\"")
+			_, err = collectionClient.ReplaceItem(
+				context,
+				azcosmos.PartitionKey{},
+				"12345",
+				bytes,
+				&azcosmos.ItemOptions{IfMatchEtag: &wrongETag},
+			)
+			assert.NotNil(t, err)
+
+			var respErr *azcore.ResponseError
+			if errors.As(err, &respErr) {
+				assert.Equal(t, http.StatusPreconditionFailed, respErr.StatusCode)
+				assert.Equal(t, "PreconditionFailed", respErr.RawResponse.Header.Get("x-ms-error-code"))
+
+				responseBody, readErr := io.ReadAll(respErr.RawResponse.Body)
+				assert.Nil(t, readErr)
+				assert.JSONEq(t,
+					`{"code":"PreconditionFailed","message":"Operation cannot be performed because one of the specified precondition is not met."}`,
+					string(responseBody),
+				)
+			} else {
+				panic(err)
+			}
+
+			document, status := ts.DataStore.GetDocument(testDatabaseName, testCollectionName, "12345")
+			assert.Equal(t, datastore.StatusOk, status)
+			assert.Equal(t, false, document["isCool"])
+		})
+
+		t.Run("Should replace with correct etag", func(t *testing.T) {
+			context := context.TODO()
+
+			readResponse, err := collectionClient.ReadItem(context, azcosmos.PartitionKey{}, "12345", nil)
+			assert.Nil(t, err)
+			assert.NotEmpty(t, readResponse.ETag)
+
+			var item map[string]interface{}
+			err = json.Unmarshal(readResponse.Value, &item)
+			assert.Nil(t, err)
+			assert.Equal(t, string(readResponse.ETag), item["_etag"])
+
+			item["pk"] = "999"
+			item["isCool"] = true
+			bytes, err := json.Marshal(item)
+			assert.Nil(t, err)
+
+			etag := readResponse.ETag
+			_, err = collectionClient.ReplaceItem(
+				context,
+				azcosmos.PartitionKey{},
+				"12345",
+				bytes,
+				&azcosmos.ItemOptions{IfMatchEtag: &etag},
+			)
+			assert.Nil(t, err)
+
+			document, status := ts.DataStore.GetDocument(testDatabaseName, testCollectionName, "12345")
+			assert.Equal(t, datastore.StatusOk, status)
+			assert.Equal(t, "999", document["pk"])
+			assert.Equal(t, true, document["isCool"])
+		})
+	})
+
 	runTestsWithPresets(t, "Test_Documents_TransactionalBatch", presets, func(t *testing.T, ts *TestServer, client *azcosmos.Client) {
 		collectionClient := documents_InitializeDb(t, ts)
 

internal/constants/responses.go

@@ -35,3 +35,7 @@ var UnknownErrorResponse = gin.H{"message": "Unknown error"}
 var NotFoundResponse = gin.H{"message": "NotFound"}
 var ConflictResponse = gin.H{"message": "Conflict"}
 var BadRequestResponse = gin.H{"message": "BadRequest"}
+var PreconditionFailedResponse = gin.H{
+	"code":    "PreconditionFailed",
+	"message": "Operation cannot be performed because one of the specified precondition is not met.",
+}

internal/datastore/badger_datastore/badger_datastore.go

@@ -12,8 +12,10 @@ import (
 )
 
 type BadgerDataStore struct {
-	db       *badger.DB
-	gcTicker *time.Ticker
+	db        *badger.DB
+	gcTicker  *time.Ticker
+	gcDone    chan struct{}
+	gcStopped chan struct{}
 }
 
 type BadgerDataStoreOptions struct {
@@ -36,8 +38,10 @@ func NewBadgerDataStore(options BadgerDataStoreOptions) *BadgerDataStore {
 	gcTicker := time.NewTicker(5 * time.Minute)
 
 	ds := &BadgerDataStore{
-		db:       db,
-		gcTicker: gcTicker,
+		db:        db,
+		gcTicker:  gcTicker,
+		gcDone:    make(chan struct{}),
+		gcStopped: make(chan struct{}),
 	}
 
 	ds.initializeDataStore(options.InitialDataFilePath)
@@ -50,7 +54,8 @@ func NewBadgerDataStore(options BadgerDataStoreOptions) *BadgerDataStore {
 func (r *BadgerDataStore) Close() {
 	if r.gcTicker != nil {
 		r.gcTicker.Stop()
-		r.gcTicker = nil
+		close(r.gcDone)
+		<-r.gcStopped
 	}
 
 	r.db.Close()
@@ -63,11 +68,19 @@ func (r *BadgerDataStore) DumpToJson() (string, error) {
 }
 
 func (r *BadgerDataStore) runGarbageCollector() {
-	for range r.gcTicker.C {
-	again:
-		err := r.db.RunValueLogGC(0.7)
-		if err == nil {
-			goto again
+	defer close(r.gcStopped)
+
+	for {
+		select {
+		case <-r.gcTicker.C:
+			for {
+				err := r.db.RunValueLogGC(0.7)
+				if err != nil {
+					break
+				}
+			}
+		case <-r.gcDone:
+			return
 		}
 	}
 }

parsers/models.go

@@ -107,6 +107,7 @@ const (
 	FunctionCallContains     FunctionCallType = "Contains"
 	FunctionCallEndsWith     FunctionCallType = "EndsWith"
 	FunctionCallStartsWith   FunctionCallType = "StartsWith"
+	FunctionCallRegexMatch   FunctionCallType = "RegexMatch"
 	FunctionCallIndexOf      FunctionCallType = "IndexOf"
 	FunctionCallToString     FunctionCallType = "ToString"
 	FunctionCallUpper        FunctionCallType = "Upper"

parsers/nosql/nosql.peg

@@ -681,6 +681,8 @@ ThreeArgumentStringFunctionExpression <- function:ThreeArgumentStringFunction ws
         functionType = parsers.FunctionCallEndsWith
     case "STARTSWITH":
         functionType = parsers.FunctionCallStartsWith
+    case "REGEXMATCH":
+        functionType = parsers.FunctionCallRegexMatch
     case "INDEX_OF":
         functionType = parsers.FunctionCallIndexOf
     }
@@ -688,7 +690,7 @@ ThreeArgumentStringFunctionExpression <- function:ThreeArgumentStringFunction ws
     return createFunctionCall(functionType, []interface{}{ex1, ex2, ignoreCase})
 }
 
-ThreeArgumentStringFunction <- ("CONTAINS"i / "ENDSWITH"i / "STARTSWITH"i / "INDEX_OF"i) {
+ThreeArgumentStringFunction <- ("CONTAINS"i / "ENDSWITH"i / "STARTSWITH"i / "REGEXMATCH"i / "INDEX_OF"i) {
     return string(c.text), nil
 }
 

parsers/nosql/string_functions_test.go

@@ -168,6 +168,32 @@ func Test_Execute_StringFunctions(t *testing.T) {
 		)
 	})
 
+	t.Run("Should parse function REGEXMATCH()", func(t *testing.T) {
+		testQueryParse(
+			t,
+			`SELECT REGEXMATCH(c.id, "aB c", "ix") FROM c`,
+			parsers.SelectStmt{
+				SelectItems: []parsers.SelectItem{
+					{
+						Type: parsers.SelectItemTypeFunctionCall,
+						Value: parsers.FunctionCall{
+							Type: parsers.FunctionCallRegexMatch,
+							Arguments: []interface{}{
+								parsers.SelectItem{
+									Path: []string{"c", "id"},
+									Type: parsers.SelectItemTypeField,
+								},
+								testutils.SelectItem_Constant_String("aB c"),
+								testutils.SelectItem_Constant_String("ix"),
+							},
+						},
+					},
+				},
+				Table: parsers.Table{SelectItem: testutils.SelectItem_Path("c")},
+			},
+		)
+	})
+
 	t.Run("Should parse function INDEX_OF()", func(t *testing.T) {
 		testQueryParse(
 			t,

query_executors/memory_executor/array_functions.go

@@ -25,7 +25,7 @@ func (r rowContext) array_Contains(arguments []interface{}) bool {
 	exprToSearch := r.resolveSelectItem(arguments[1].(parsers.SelectItem))
 
 	partialSearch := false
-	if len(arguments) > 2 {
+	if len(arguments) > 2 && arguments[2] != nil {
 		boolExpr := r.resolveSelectItem(arguments[2].(parsers.SelectItem))
 		if boolValue, ok := boolExpr.(bool); ok {
 			partialSearch = boolValue

query_executors/memory_executor/common.go

@@ -162,6 +162,8 @@ func (r rowContext) selectItem_SelectItemTypeFunctionCall(functionCall parsers.F
 		return r.strings_EndsWith(functionCall.Arguments)
 	case parsers.FunctionCallStartsWith:
 		return r.strings_StartsWith(functionCall.Arguments)
+	case parsers.FunctionCallRegexMatch:
+		return r.strings_RegexMatch(functionCall.Arguments)
 	case parsers.FunctionCallConcat:
 		return r.strings_Concat(functionCall.Arguments)
 	case parsers.FunctionCallIndexOf:

query_executors/memory_executor/string_functions.go

@@ -2,6 +2,7 @@ package memoryexecutor
 
 import (
 	"fmt"
+	"regexp"
 	"strings"
 
 	"github.com/pikami/cosmium/internal/logger"
@@ -75,6 +76,46 @@ func (r rowContext) strings_StartsWith(arguments []interface{}) bool {
 	return strings.HasPrefix(str1, str2)
 }
 
+func (r rowContext) strings_RegexMatch(arguments []interface{}) bool {
+	value, valueOk := r.parseString(arguments[0])
+	pattern, patternOk := r.parseString(arguments[1])
+	if !valueOk || !patternOk {
+		return false
+	}
+
+	modifiers, ok := r.getStringFlag(arguments)
+	if !ok {
+		return false
+	}
+
+	regexPattern := pattern
+	if strings.Contains(modifiers, "x") {
+		regexPattern = stripRegexIgnoredWhitespace(regexPattern)
+	}
+
+	var flags strings.Builder
+	if strings.Contains(modifiers, "i") {
+		flags.WriteByte('i')
+	}
+	if strings.Contains(modifiers, "m") {
+		flags.WriteByte('m')
+	}
+	if strings.Contains(modifiers, "s") {
+		flags.WriteByte('s')
+	}
+	if flags.Len() > 0 {
+		regexPattern = "(?" + flags.String() + ")" + regexPattern
+	}
+
+	matched, err := regexp.MatchString(regexPattern, value)
+	if err != nil {
+		logger.Errorf("strings_RegexMatch - invalid pattern %q: %v", pattern, err)
+		return false
+	}
+
+	return matched
+}
+
 func (r rowContext) strings_Concat(arguments []interface{}) string {
 	result := ""
 
@@ -318,6 +359,20 @@ func (r rowContext) getBoolFlag(arguments []interface{}) bool {
 	return ignoreCase
 }
 
+func (r rowContext) getStringFlag(arguments []interface{}) (string, bool) {
+	if len(arguments) <= 2 || arguments[2] == nil {
+		return "", true
+	}
+
+	flagItem := arguments[2].(parsers.SelectItem)
+	if value, ok := r.resolveSelectItem(flagItem).(string); ok {
+		return value, true
+	}
+
+	logger.ErrorLn("getStringFlag - got parameters of wrong type")
+	return "", false
+}
+
 func (r rowContext) parseString(argument interface{}) (value string, ok bool) {
 	exItem := argument.(parsers.SelectItem)
 	ex := r.resolveSelectItem(exItem)
@@ -329,6 +384,41 @@ func (r rowContext) parseString(argument interface{}) (value string, ok bool) {
 	return "", false
 }
 
+func stripRegexIgnoredWhitespace(pattern string) string {
+	var result strings.Builder
+	inCharClass := false
+	escaped := false
+
+	for _, r := range pattern {
+		if escaped {
+			result.WriteRune(r)
+			escaped = false
+			continue
+		}
+
+		if r == '\\' {
+			result.WriteRune(r)
+			escaped = true
+			continue
+		}
+
+		switch r {
+		case '[':
+			inCharClass = true
+		case ']':
+			inCharClass = false
+		}
+
+		if !inCharClass && (r == ' ' || r == '\t' || r == '\n' || r == '\r' || r == '\f') {
+			continue
+		}
+
+		result.WriteRune(r)
+	}
+
+	return result.String()
+}
+
 func convertToString(value interface{}) string {
 	switch v := value.(type) {
 	case string:

query_executors/memory_executor/string_functions_test.go

@@ -231,6 +231,42 @@ func Test_Execute_StringFunctions(t *testing.T) {
 		)
 	})
 
+	t.Run("Should execute function REGEXMATCH()", func(t *testing.T) {
+		testQueryExecute(
+			t,
+			parsers.SelectStmt{
+				SelectItems: []parsers.SelectItem{
+					{
+						Path: []string{"c", "id"},
+						Type: parsers.SelectItemTypeField,
+					},
+					{
+						Alias: "regexMatch",
+						Type:  parsers.SelectItemTypeFunctionCall,
+						Value: parsers.FunctionCall{
+							Type: parsers.FunctionCallRegexMatch,
+							Arguments: []interface{}{
+								parsers.SelectItem{
+									Path: []string{"c", "str"},
+									Type: parsers.SelectItemTypeField,
+								},
+								testutils.SelectItem_Constant_String("COOL WORLD"),
+								testutils.SelectItem_Constant_String("i"),
+							},
+						},
+					},
+				},
+				Table: parsers.Table{SelectItem: testutils.SelectItem_Path("c")},
+			},
+			mockData,
+			[]memoryexecutor.RowType{
+				map[string]interface{}{"id": "123", "regexMatch": false},
+				map[string]interface{}{"id": "456", "regexMatch": false},
+				map[string]interface{}{"id": "789", "regexMatch": true},
+			},
+		)
+	})
+
 	t.Run("Should execute function INDEX_OF()", func(t *testing.T) {
 		testQueryExecute(
 			t,