From a06788aad5b57ae0d193ebc4f2fadb379b39e5f0 Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Fri, 17 Jul 2026 23:06:34 +0900 Subject: [PATCH 1/5] refactor(dynamodb): replace badger with sqlite; drop the dependency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rewrite the DynamoDB store on the shared SQLite backend used by every other service, removing the sole dependency on dgraph-io/badger and its transitive tree (ristretto, flatbuffers, klauspost/compress, go-humanize, otel, go-logr, xxhash, protobuf). - items/tables/ttl/tags now live in SQLite tables; table metadata is still cached in memory and loaded on open - GSI/LSI queries are projected on the fly from the base table (no denormalized index rows to keep in sync) - drop the dead badger-transaction surface (ExecTransaction, PutItemTxn, GetItemTxn, DeleteItemTxn, PutItemWithGSI — all had 0 callers) Verified: go build ./..., go test ./... (108 ok, 0 fail), and a live server round-trip (CreateTable/PutItem/GetItem) that survives a restart on the same data dir. --- go.mod | 17 +- go.sum | 39 +- internal/services/dynamodb/store.go | 875 ++++++++-------------------- 3 files changed, 253 insertions(+), 678 deletions(-) diff --git a/go.mod b/go.mod index a9c28cf3..20bf698c 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,6 @@ go 1.26 toolchain go1.26.1 require ( - github.com/dgraph-io/badger/v4 v4.9.4 github.com/gorilla/websocket v1.5.3 github.com/mattn/go-sqlite3 v1.14.42 github.com/stretchr/testify v1.11.1 @@ -13,19 +12,9 @@ require ( ) require ( - github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect - github.com/dgraph-io/ristretto/v2 v2.2.0 // indirect - github.com/dustin/go-humanize v1.0.1 // indirect - github.com/go-logr/logr v1.4.3 // indirect - github.com/go-logr/stdr v1.2.2 // indirect - github.com/google/flatbuffers v25.2.10+incompatible // indirect - github.com/klauspost/compress v1.18.0 // indirect + github.com/kr/pretty v0.3.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect - go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.opentelemetry.io/otel v1.41.0 // indirect - go.opentelemetry.io/otel/metric v1.41.0 // indirect - go.opentelemetry.io/otel/trace v1.41.0 // indirect - golang.org/x/sys v0.35.0 // indirect - google.golang.org/protobuf v1.36.7 // indirect + github.com/rogpeppe/go-internal v1.14.1 // indirect + gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect ) diff --git a/go.sum b/go.sum index a45c18ee..22d3eab1 100644 --- a/go.sum +++ b/go.sum @@ -1,52 +1,25 @@ -github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= -github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/dgraph-io/badger/v4 v4.9.4 h1:bcw+waCpzRZ2nmcSPbnPvDVhiEsn98TKmvnAhK7r7LM= -github.com/dgraph-io/badger/v4 v4.9.4/go.mod h1:nJjaJTUOSsQEBhsq209FmwCvMJzEA3e74RjZw6V2pQI= -github.com/dgraph-io/ristretto/v2 v2.2.0 h1:bkY3XzJcXoMuELV8F+vS8kzNgicwQFAaGINAEJdWGOM= -github.com/dgraph-io/ristretto/v2 v2.2.0/go.mod h1:RZrm63UmcBAaYWC1DotLYBmTvgkrs0+XhBd7Npn7/zI= -github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da h1:aIftn67I1fkbMa512G+w+Pxci9hJPB8oMnkcP3iZF38= -github.com/dgryski/go-farm v0.0.0-20240924180020-3414d57e47da/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw= -github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY= -github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto= -github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= -github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= -github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= -github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= -github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= -github.com/google/flatbuffers v25.2.10+incompatible h1:F3vclr7C3HpB1k9mxCGRMXq6FdUalZ6H/pNX4FP1v0Q= -github.com/google/flatbuffers v25.2.10+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8= -github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= -github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= -github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= -github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= +github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/mattn/go-sqlite3 v1.14.42 h1:MigqEP4ZmHw3aIdIT7T+9TLa90Z6smwcthx+Azv4Cgo= github.com/mattn/go-sqlite3 v1.14.42/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= -go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/otel v1.41.0 h1:YlEwVsGAlCvczDILpUXpIpPSL/VPugt7zHThEMLce1c= -go.opentelemetry.io/otel v1.41.0/go.mod h1:Yt4UwgEKeT05QbLwbyHXEwhnjxNO6D8L5PQP51/46dE= -go.opentelemetry.io/otel/metric v1.41.0 h1:rFnDcs4gRzBcsO9tS8LCpgR0dxg4aaxWlJxCno7JlTQ= -go.opentelemetry.io/otel/metric v1.41.0/go.mod h1:xPvCwd9pU0VN8tPZYzDZV/BMj9CM9vs00GuBjeKhJps= -go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa9TIN0= -go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= -golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= -golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= -google.golang.org/protobuf v1.36.7 h1:IgrO7UwFQGJdRNXH/sQux4R1Dj1WAKcLElzeeRaXV2A= -google.golang.org/protobuf v1.36.7/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/5YcXBHnY= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/internal/services/dynamodb/store.go b/internal/services/dynamodb/store.go index 486cb4fe..da9be089 100644 --- a/internal/services/dynamodb/store.go +++ b/internal/services/dynamodb/store.go @@ -3,14 +3,16 @@ package dynamodb import ( + "database/sql" "encoding/json" "errors" "fmt" + "path/filepath" "strings" "sync" "time" - badger "github.com/dgraph-io/badger/v4" + "github.com/skyoo2003/devcloud/internal/storage/sqlite" ) // Sentinel errors for DynamoStore operations. @@ -81,25 +83,44 @@ type AttributeValue struct { // Item is a DynamoDB item: a map of attribute names to values. type Item map[string]*AttributeValue -// DynamoStore is a BadgerDB-backed store for DynamoDB tables and items. +var migrations = []sqlite.Migration{ + {Version: 1, SQL: ` + CREATE TABLE IF NOT EXISTS ddb_tables ( + name TEXT PRIMARY KEY, + info TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS ddb_items ( + table_name TEXT NOT NULL, + pk TEXT NOT NULL, + sk TEXT NOT NULL DEFAULT '', + data TEXT NOT NULL, + PRIMARY KEY (table_name, pk, sk) + ); + CREATE TABLE IF NOT EXISTS ddb_ttl ( + table_name TEXT PRIMARY KEY, + config TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS ddb_tags ( + resource_arn TEXT PRIMARY KEY, + tags TEXT NOT NULL + ); + `}, +} + +// DynamoStore is a SQLite-backed store for DynamoDB tables and items. Table +// metadata is cached in memory (loaded on open); items live in SQLite. type DynamoStore struct { - db *badger.DB + db *sqlite.Store tables map[string]*TableInfo mu sync.RWMutex } -const metaPrefix = "_meta/" -const itemPrefix = "_item/" -const gsiPrefix = "_gsi/" -const ttlPrefix = "_ttl/" -const tagsPrefix = "_tags/" - -// NewDynamoStore opens (or creates) a BadgerDB at dir and loads existing table metadata. +// NewDynamoStore opens (or creates) a SQLite database under dir and loads +// existing table metadata. func NewDynamoStore(dir string) (*DynamoStore, error) { - opts := badger.DefaultOptions(dir).WithLogger(nil) - db, err := badger.Open(opts) + db, err := sqlite.Open(filepath.Join(dir, "dynamodb.db"), migrations) if err != nil { - return nil, fmt.Errorf("open badger: %w", err) + return nil, fmt.Errorf("open dynamodb store: %w", err) } s := &DynamoStore{ @@ -107,7 +128,6 @@ func NewDynamoStore(dir string) (*DynamoStore, error) { tables: make(map[string]*TableInfo), } - // Load existing table metadata from the database. if err := s.loadTableMeta(); err != nil { _ = db.Close() return nil, fmt.Errorf("load table metadata: %w", err) @@ -116,40 +136,71 @@ func NewDynamoStore(dir string) (*DynamoStore, error) { return s, nil } -// loadTableMeta reads all _meta/ keys and populates the in-memory tables map. +// loadTableMeta reads all persisted table metadata into the in-memory map. func (s *DynamoStore) loadTableMeta() error { - return s.db.View(func(txn *badger.Txn) error { - prefix := []byte(metaPrefix) - opts := badger.DefaultIteratorOptions - opts.Prefix = prefix - it := txn.NewIterator(opts) - defer it.Close() - - for it.Rewind(); it.Valid(); it.Next() { - item := it.Item() - if err := item.Value(func(v []byte) error { - var info TableInfo - if err := json.Unmarshal(v, &info); err != nil { - return err - } - s.tables[info.Name] = &info - return nil - }); err != nil { - return err - } + rows, err := s.db.DB().Query(`SELECT info FROM ddb_tables`) + if err != nil { + return err + } + defer func() { _ = rows.Close() }() + for rows.Next() { + var raw string + if err := rows.Scan(&raw); err != nil { + return err } - return nil - }) + var info TableInfo + if err := json.Unmarshal([]byte(raw), &info); err != nil { + return err + } + s.tables[info.Name] = &info + } + return rows.Err() } -// Close closes the underlying BadgerDB. +// Close closes the underlying database. func (s *DynamoStore) Close() error { return s.db.Close() } -// metaKey returns the BadgerDB key for a table's metadata record. -func metaKey(tableName string) []byte { - return []byte(metaPrefix + tableName) +// attributeStringValue extracts the string representation of an AttributeValue key field. +func attributeStringValue(av *AttributeValue) (string, error) { + if av == nil { + return "", fmt.Errorf("attribute value is nil") + } + if av.S != nil { + return *av.S, nil + } + if av.N != nil { + return *av.N, nil + } + if av.B != nil { + return string(av.B), nil + } + return "", fmt.Errorf("unsupported attribute value type for key") +} + +// keyValues derives the partition and (optional) sort key string values for an +// item or key map, given its table's schema. +func keyValues(table *TableInfo, item Item) (pk, sk string, err error) { + pkAttr, ok := item[table.PartitionKey.Name] + if !ok { + return "", "", fmt.Errorf("missing partition key attribute %q", table.PartitionKey.Name) + } + pk, err = attributeStringValue(pkAttr) + if err != nil { + return "", "", fmt.Errorf("partition key value: %w", err) + } + if table.SortKey != nil { + skAttr, ok := item[table.SortKey.Name] + if !ok { + return "", "", fmt.Errorf("missing sort key attribute %q", table.SortKey.Name) + } + sk, err = attributeStringValue(skAttr) + if err != nil { + return "", "", fmt.Errorf("sort key value: %w", err) + } + } + return pk, sk, nil } // CreateTable creates a new table. Returns ErrTableAlreadyExists if the table exists. @@ -176,9 +227,8 @@ func (s *DynamoStore) CreateTable(info TableInfo) error { return fmt.Errorf("marshal table info: %w", err) } - if err := s.db.Update(func(txn *badger.Txn) error { - return txn.Set(metaKey(info.Name), data) - }); err != nil { + if _, err := s.db.DB().Exec( + `INSERT OR REPLACE INTO ddb_tables (name, info) VALUES (?, ?)`, info.Name, string(data)); err != nil { return fmt.Errorf("persist table metadata: %w", err) } @@ -195,38 +245,11 @@ func (s *DynamoStore) DeleteTable(name string) error { return ErrTableNotFound } - // Collect all item keys belonging to this table. - tableItemPrefix := []byte(itemPrefix + name + "/") - var keysToDelete [][]byte - - if err := s.db.View(func(txn *badger.Txn) error { - opts := badger.DefaultIteratorOptions - opts.Prefix = tableItemPrefix - opts.PrefetchValues = false - it := txn.NewIterator(opts) - defer it.Close() - - for it.Rewind(); it.Valid(); it.Next() { - k := it.Item().KeyCopy(nil) - keysToDelete = append(keysToDelete, k) - } - return nil - }); err != nil { - return fmt.Errorf("scan items for deletion: %w", err) + if _, err := s.db.DB().Exec(`DELETE FROM ddb_items WHERE table_name = ?`, name); err != nil { + return fmt.Errorf("delete table items: %w", err) } - - // Delete metadata and all items in a single batch. - keysToDelete = append(keysToDelete, metaKey(name)) - - if err := s.db.Update(func(txn *badger.Txn) error { - for _, k := range keysToDelete { - if err := txn.Delete(k); err != nil { - return err - } - } - return nil - }); err != nil { - return fmt.Errorf("delete table records: %w", err) + if _, err := s.db.DB().Exec(`DELETE FROM ddb_tables WHERE name = ?`, name); err != nil { + return fmt.Errorf("delete table metadata: %w", err) } delete(s.tables, name) @@ -254,70 +277,40 @@ func (s *DynamoStore) GetTable(name string) (*TableInfo, error) { if !exists { return nil, ErrTableNotFound } - copy := *info - return ©, nil -} - -// itemKey builds the BadgerDB key for an item. -// Format: _item/tableName/partitionKeyValue or _item/tableName/partitionKeyValue/sortKeyValue -func itemKey(tableName, pk string, sk ...string) []byte { - key := itemPrefix + tableName + "/" + pk - if len(sk) > 0 && sk[0] != "" { - key += "/" + sk[0] - } - return []byte(key) + cp := *info + return &cp, nil } -// attributeStringValue extracts the string representation of an AttributeValue key field. -func attributeStringValue(av *AttributeValue) (string, error) { - if av == nil { - return "", fmt.Errorf("attribute value is nil") - } - if av.S != nil { - return *av.S, nil - } - if av.N != nil { - return *av.N, nil +// UpdateTable applies updates to a table's metadata and persists them. +func (s *DynamoStore) UpdateTable(name string, updates func(*TableInfo)) error { + s.mu.Lock() + defer s.mu.Unlock() + info, exists := s.tables[name] + if !exists { + return ErrTableNotFound } - if av.B != nil { - return string(av.B), nil + updates(info) + data, err := json.Marshal(info) + if err != nil { + return fmt.Errorf("marshal table info: %w", err) } - return "", fmt.Errorf("unsupported attribute value type for key") + _, err = s.db.DB().Exec(`UPDATE ddb_tables SET info = ? WHERE name = ?`, string(data), name) + return err } -// PutItem stores an item in the specified table. -// The item key is derived from the table's partition key (and optional sort key) values. +// PutItem stores (upserts) an item in the specified table. The item key is +// derived from the table's partition key (and optional sort key) values. func (s *DynamoStore) PutItem(tableName string, item Item) error { s.mu.RLock() table, exists := s.tables[tableName] s.mu.RUnlock() - if !exists { return ErrTableNotFound } - pkAttr, ok := item[table.PartitionKey.Name] - if !ok { - return fmt.Errorf("item missing partition key attribute %q", table.PartitionKey.Name) - } - pkVal, err := attributeStringValue(pkAttr) + pk, sk, err := keyValues(table, item) if err != nil { - return fmt.Errorf("partition key value: %w", err) - } - - var key []byte - if table.SortKey != nil { - skAttr, ok := item[table.SortKey.Name] - if !ok { - return fmt.Errorf("item missing sort key attribute %q", table.SortKey.Name) - } - skVal, err := attributeStringValue(skAttr) - if err != nil { - return fmt.Errorf("sort key value: %w", err) - } - key = itemKey(tableName, pkVal, skVal) - } else { - key = itemKey(tableName, pkVal) + return err } data, err := json.Marshal(item) @@ -325,23 +318,10 @@ func (s *DynamoStore) PutItem(tableName string, item Item) error { return fmt.Errorf("marshal item: %w", err) } - return s.db.Update(func(txn *badger.Txn) error { - if err := txn.Set(key, data); err != nil { - return err - } - // Also update GSI/LSI entries. - for _, idx := range table.GlobalSecondaryIndexes { - if err := writeGSIEntry(txn, tableName, idx, item, data); err != nil { - return err - } - } - for _, idx := range table.LocalSecondaryIndexes { - if err := writeGSIEntry(txn, tableName, idx, item, data); err != nil { - return err - } - } - return nil - }) + _, err = s.db.DB().Exec( + `INSERT OR REPLACE INTO ddb_items (table_name, pk, sk, data) VALUES (?, ?, ?, ?)`, + tableName, pk, sk, string(data)) + return err } // GetItem retrieves an item by its key attributes. Returns ErrItemNotFound if missing. @@ -349,49 +329,28 @@ func (s *DynamoStore) GetItem(tableName string, key Item) (*Item, error) { s.mu.RLock() table, exists := s.tables[tableName] s.mu.RUnlock() - if !exists { return nil, ErrTableNotFound } - pkAttr, ok := key[table.PartitionKey.Name] - if !ok { - return nil, fmt.Errorf("key missing partition key attribute %q", table.PartitionKey.Name) - } - pkVal, err := attributeStringValue(pkAttr) + pk, sk, err := keyValues(table, key) if err != nil { - return nil, fmt.Errorf("partition key value: %w", err) + return nil, err } - var dbKey []byte - if table.SortKey != nil { - skAttr, ok := key[table.SortKey.Name] - if !ok { - return nil, fmt.Errorf("key missing sort key attribute %q", table.SortKey.Name) - } - skVal, err := attributeStringValue(skAttr) - if err != nil { - return nil, fmt.Errorf("sort key value: %w", err) - } - dbKey = itemKey(tableName, pkVal, skVal) - } else { - dbKey = itemKey(tableName, pkVal) + var raw string + err = s.db.DB().QueryRow( + `SELECT data FROM ddb_items WHERE table_name = ? AND pk = ? AND sk = ?`, + tableName, pk, sk).Scan(&raw) + if errors.Is(err, sql.ErrNoRows) { + return nil, ErrItemNotFound + } + if err != nil { + return nil, err } var result Item - err = s.db.View(func(txn *badger.Txn) error { - dbItem, err := txn.Get(dbKey) - if err != nil { - if errors.Is(err, badger.ErrKeyNotFound) { - return ErrItemNotFound - } - return err - } - return dbItem.Value(func(v []byte) error { - return json.Unmarshal(v, &result) - }) - }) - if err != nil { + if err := json.Unmarshal([]byte(raw), &result); err != nil { return nil, err } return &result, nil @@ -402,117 +361,44 @@ func (s *DynamoStore) DeleteItem(tableName string, key Item) error { s.mu.RLock() table, exists := s.tables[tableName] s.mu.RUnlock() - if !exists { return ErrTableNotFound } - pkAttr, ok := key[table.PartitionKey.Name] - if !ok { - return fmt.Errorf("key missing partition key attribute %q", table.PartitionKey.Name) - } - pkVal, err := attributeStringValue(pkAttr) + pk, sk, err := keyValues(table, key) if err != nil { - return fmt.Errorf("partition key value: %w", err) + return err } - var dbKey []byte - if table.SortKey != nil { - skAttr, ok := key[table.SortKey.Name] - if !ok { - return fmt.Errorf("key missing sort key attribute %q", table.SortKey.Name) - } - skVal, err := attributeStringValue(skAttr) - if err != nil { - return fmt.Errorf("sort key value: %w", err) - } - dbKey = itemKey(tableName, pkVal, skVal) - } else { - dbKey = itemKey(tableName, pkVal) + res, err := s.db.DB().Exec( + `DELETE FROM ddb_items WHERE table_name = ? AND pk = ? AND sk = ?`, tableName, pk, sk) + if err != nil { + return err } - - return s.db.Update(func(txn *badger.Txn) error { - _, err := txn.Get(dbKey) - if err != nil { - if errors.Is(err, badger.ErrKeyNotFound) { - return ErrItemNotFound - } - return err - } - return txn.Delete(dbKey) - }) + if n, _ := res.RowsAffected(); n == 0 { + return ErrItemNotFound + } + return nil } -// Query performs a prefix scan for items with the given partitionKeyValue. -// If sortKeyPrefix is non-empty, only items whose sort key starts with sortKeyPrefix are returned. +// Query returns items with the given partition key value, ordered by sort key. +// If sortKeyPrefix is non-empty, only items whose sort key starts with it are returned. func (s *DynamoStore) Query(tableName string, partitionKeyValue string, sortKeyPrefix string) ([]Item, error) { s.mu.RLock() table, exists := s.tables[tableName] s.mu.RUnlock() - if !exists { return nil, ErrTableNotFound } - // For partition-key-only tables, do an exact key lookup since items - // are stored as _item/tableName/pkValue (no trailing slash). - if table.SortKey == nil { - var items []Item - err := s.db.View(func(txn *badger.Txn) error { - key := itemKey(tableName, partitionKeyValue) - dbItem, err := txn.Get(key) - if err != nil { - if errors.Is(err, badger.ErrKeyNotFound) { - return nil // no match is not an error - } - return err - } - return dbItem.Value(func(v []byte) error { - var item Item - if err := json.Unmarshal(v, &item); err != nil { - return err - } - items = append(items, item) - return nil - }) - }) - if err != nil { - return nil, err - } - return items, nil + query := `SELECT data FROM ddb_items WHERE table_name = ? AND pk = ?` + args := []any{tableName, partitionKeyValue} + if table.SortKey != nil && sortKeyPrefix != "" { + query += ` AND sk LIKE ? ESCAPE '\'` + args = append(args, escapeLike(sortKeyPrefix)+"%") } - - // For tables with sort key, prefix scan: _item/tableName/partitionKeyValue/ - prefix := itemPrefix + tableName + "/" + partitionKeyValue + "/" - if sortKeyPrefix != "" { - prefix += sortKeyPrefix - } - - var items []Item - err := s.db.View(func(txn *badger.Txn) error { - opts := badger.DefaultIteratorOptions - opts.Prefix = []byte(prefix) - it := txn.NewIterator(opts) - defer it.Close() - - for it.Rewind(); it.Valid(); it.Next() { - if err := it.Item().Value(func(v []byte) error { - var item Item - if err := json.Unmarshal(v, &item); err != nil { - return err - } - items = append(items, item) - return nil - }); err != nil { - return err - } - } - return nil - }) - if err != nil { - return nil, err - } - return items, nil + query += ` ORDER BY sk` + return s.queryItems(query, args) } // Scan returns all items in a table. @@ -520,65 +406,85 @@ func (s *DynamoStore) Scan(tableName string) ([]Item, error) { s.mu.RLock() _, exists := s.tables[tableName] s.mu.RUnlock() - if !exists { return nil, ErrTableNotFound } + return s.queryItems(`SELECT data FROM ddb_items WHERE table_name = ? ORDER BY pk, sk`, []any{tableName}) +} - prefix := []byte(itemPrefix + tableName + "/") +// QueryGSI returns items whose value for the index's HASH key attribute equals +// pkValue. Secondary indexes are projected on the fly from the base table. +func (s *DynamoStore) QueryGSI(tableName, indexName, pkValue string) ([]Item, error) { + s.mu.RLock() + table, exists := s.tables[tableName] + s.mu.RUnlock() + if !exists { + return nil, ErrTableNotFound + } - var items []Item - err := s.db.View(func(txn *badger.Txn) error { - opts := badger.DefaultIteratorOptions - opts.Prefix = prefix - it := txn.NewIterator(opts) - defer it.Close() - - for it.Rewind(); it.Valid(); it.Next() { - // Skip sub-prefix keys that aren't direct items (none expected, but be safe). - keyStr := string(it.Item().Key()) - // Only include items that belong directly to this table prefix. - // Strip the table item prefix and check there's actual content. - rest := strings.TrimPrefix(keyStr, string(prefix)) - if rest == "" { - continue - } + hashAttr := indexHashKey(table, indexName) + if hashAttr == "" { + return nil, nil // unknown index — no matches + } - if err := it.Item().Value(func(v []byte) error { - var item Item - if err := json.Unmarshal(v, &item); err != nil { - return err - } - items = append(items, item) - return nil - }); err != nil { - return err - } - } - return nil - }) + all, err := s.queryItems(`SELECT data FROM ddb_items WHERE table_name = ?`, []any{tableName}) if err != nil { return nil, err } + var items []Item + for _, item := range all { + av, ok := item[hashAttr] + if !ok { + continue + } + if v, err := attributeStringValue(av); err == nil && v == pkValue { + items = append(items, item) + } + } return items, nil } -// UpdateTable updates billing mode and throughput stored in metadata. -func (s *DynamoStore) UpdateTable(name string, updates func(*TableInfo)) error { - s.mu.Lock() - defer s.mu.Unlock() - info, exists := s.tables[name] - if !exists { - return ErrTableNotFound +// indexHashKey returns the HASH key attribute name for the named GSI or LSI, +// or "" if no such index exists. +func indexHashKey(table *TableInfo, indexName string) string { + for _, idx := range append(table.GlobalSecondaryIndexes, table.LocalSecondaryIndexes...) { + if idx.IndexName != indexName { + continue + } + for _, ks := range idx.KeySchema { + if ks.KeyType == "HASH" { + return ks.Name + } + } } - updates(info) - data, err := json.Marshal(info) + return "" +} + +// queryItems runs a SELECT that returns a single "data" column of JSON items. +func (s *DynamoStore) queryItems(query string, args []any) ([]Item, error) { + rows, err := s.db.DB().Query(query, args...) if err != nil { - return fmt.Errorf("marshal table info: %w", err) + return nil, err } - return s.db.Update(func(txn *badger.Txn) error { - return txn.Set(metaKey(name), data) - }) + defer func() { _ = rows.Close() }() + var items []Item + for rows.Next() { + var raw string + if err := rows.Scan(&raw); err != nil { + return nil, err + } + var item Item + if err := json.Unmarshal([]byte(raw), &item); err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} + +// escapeLike escapes LIKE metacharacters so a prefix matches literally. +func escapeLike(s string) string { + return strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(s) } // PutTTLConfig stores TTL config for a table. @@ -587,51 +493,42 @@ func (s *DynamoStore) PutTTLConfig(tableName string, cfg TTLConfig) error { if err != nil { return err } - return s.db.Update(func(txn *badger.Txn) error { - return txn.Set([]byte(ttlPrefix+tableName), data) - }) + _, err = s.db.DB().Exec( + `INSERT OR REPLACE INTO ddb_ttl (table_name, config) VALUES (?, ?)`, tableName, string(data)) + return err } -// GetTTLConfig retrieves TTL config for a table. +// GetTTLConfig retrieves TTL config for a table. Returns an empty config if unset. func (s *DynamoStore) GetTTLConfig(tableName string) (*TTLConfig, error) { - var cfg TTLConfig - err := s.db.View(func(txn *badger.Txn) error { - dbItem, err := txn.Get([]byte(ttlPrefix + tableName)) - if err != nil { - if errors.Is(err, badger.ErrKeyNotFound) { - cfg = TTLConfig{} - return nil - } - return err - } - return dbItem.Value(func(v []byte) error { - return json.Unmarshal(v, &cfg) - }) - }) + var raw string + err := s.db.DB().QueryRow(`SELECT config FROM ddb_ttl WHERE table_name = ?`, tableName).Scan(&raw) + if errors.Is(err, sql.ErrNoRows) { + return &TTLConfig{}, nil + } if err != nil { return nil, err } + var cfg TTLConfig + if err := json.Unmarshal([]byte(raw), &cfg); err != nil { + return nil, err + } return &cfg, nil } // GetTags retrieves tags for a resource ARN. func (s *DynamoStore) GetTags(resourceArn string) (map[string]string, error) { tags := make(map[string]string) - err := s.db.View(func(txn *badger.Txn) error { - dbItem, err := txn.Get([]byte(tagsPrefix + resourceArn)) - if err != nil { - if errors.Is(err, badger.ErrKeyNotFound) { - return nil - } - return err - } - return dbItem.Value(func(v []byte) error { - return json.Unmarshal(v, &tags) - }) - }) + var raw string + err := s.db.DB().QueryRow(`SELECT tags FROM ddb_tags WHERE resource_arn = ?`, resourceArn).Scan(&raw) + if errors.Is(err, sql.ErrNoRows) { + return tags, nil + } if err != nil { return nil, err } + if err := json.Unmarshal([]byte(raw), &tags); err != nil { + return nil, err + } return tags, nil } @@ -644,13 +541,7 @@ func (s *DynamoStore) PutTags(resourceArn string, newTags map[string]string) err for k, v := range newTags { existing[k] = v } - data, err := json.Marshal(existing) - if err != nil { - return err - } - return s.db.Update(func(txn *badger.Txn) error { - return txn.Set([]byte(tagsPrefix+resourceArn), data) - }) + return s.writeTags(resourceArn, existing) } // RemoveTags removes specified tag keys for a resource ARN. @@ -662,293 +553,15 @@ func (s *DynamoStore) RemoveTags(resourceArn string, tagKeys []string) error { for _, k := range tagKeys { delete(existing, k) } - data, err := json.Marshal(existing) - if err != nil { - return err - } - return s.db.Update(func(txn *badger.Txn) error { - return txn.Set([]byte(tagsPrefix+resourceArn), data) - }) + return s.writeTags(resourceArn, existing) } -// QueryGSI scans items under the GSI prefix for a given index and partition key value. -func (s *DynamoStore) QueryGSI(tableName, indexName, pkValue string) ([]Item, error) { - s.mu.RLock() - _, exists := s.tables[tableName] - s.mu.RUnlock() - if !exists { - return nil, ErrTableNotFound - } - - prefix := []byte(gsiPrefix + tableName + "/" + indexName + "/" + pkValue + "/") - var items []Item - err := s.db.View(func(txn *badger.Txn) error { - opts := badger.DefaultIteratorOptions - opts.Prefix = prefix - it := txn.NewIterator(opts) - defer it.Close() - for it.Rewind(); it.Valid(); it.Next() { - if err := it.Item().Value(func(v []byte) error { - var item Item - if err := json.Unmarshal(v, &item); err != nil { - return err - } - items = append(items, item) - return nil - }); err != nil { - return err - } - } - return nil - }) - return items, err -} - -// PutItemWithGSI stores an item and updates GSI/LSI entries. -func (s *DynamoStore) PutItemWithGSI(tableName string, item Item) error { - s.mu.RLock() - table, exists := s.tables[tableName] - s.mu.RUnlock() - if !exists { - return ErrTableNotFound - } - - pkAttr, ok := item[table.PartitionKey.Name] - if !ok { - return fmt.Errorf("item missing partition key attribute %q", table.PartitionKey.Name) - } - pkVal, err := attributeStringValue(pkAttr) +func (s *DynamoStore) writeTags(resourceArn string, tags map[string]string) error { + data, err := json.Marshal(tags) if err != nil { - return fmt.Errorf("partition key value: %w", err) - } - - var key []byte - if table.SortKey != nil { - skAttr, ok := item[table.SortKey.Name] - if !ok { - return fmt.Errorf("item missing sort key attribute %q", table.SortKey.Name) - } - skVal, err := attributeStringValue(skAttr) - if err != nil { - return fmt.Errorf("sort key value: %w", err) - } - key = itemKey(tableName, pkVal, skVal) - } else { - key = itemKey(tableName, pkVal) - } - - data, err := json.Marshal(item) - if err != nil { - return fmt.Errorf("marshal item: %w", err) - } - - return s.db.Update(func(txn *badger.Txn) error { - if err := txn.Set(key, data); err != nil { - return err - } - for _, idx := range table.GlobalSecondaryIndexes { - if err := writeGSIEntry(txn, tableName, idx, item, data); err != nil { - return err - } - } - for _, idx := range table.LocalSecondaryIndexes { - if err := writeGSIEntry(txn, tableName, idx, item, data); err != nil { - return err - } - } - return nil - }) -} - -// writeGSIEntry writes an index entry for the item. itemPKVal and itemSKVal -// are the main table primary key values — used to ensure uniqueness within -// the GSI partition even when multiple items share the same GSI partition key. -func writeGSIEntry(txn *badger.Txn, tableName string, idx IndexDef, item Item, data []byte) error { - var gsiPK, gsiSK string - for _, ks := range idx.KeySchema { - av, ok := item[ks.Name] - if !ok { - return nil - } - val, err := attributeStringValue(av) - if err != nil { - return nil - } - if ks.KeyType == "HASH" { - gsiPK = val - } else { - gsiSK = val - } - } - if gsiPK == "" { - return nil // no GSI partition key found in item, skip - } - - // Build a unique key within the GSI using the full item key bytes as a suffix. - // Key format: _gsi/{table}/{index}/{gsiPK}/{gsiSK or _}/{itemKey} - // To derive itemKey, look at all fields in item to build a deterministic suffix. - // We use a hash of the serialized item key attributes for uniqueness. - // Simpler approach: store by gsiPK/gsiSK, but append a unique sub-key based on - // all item key fields to avoid collision. - // - // We encode the item's partition key value from whatever fields exist. - // Since we don't have table schema here, use the raw data bytes hash. - // Actually, build a suffix from the data content (or just the first 32 bytes of hash). - // - // Simplest unique approach: use gsiPK/gsiSK/itemPrimaryKeyEncoded where we - // compute a deterministic suffix from the item data. - // Use a simple approach: take gsiPK and gsiSK, then append a unique per-item suffix - // by hashing. For emulator purposes, use the JSON data length + first bytes. - // - // Best approach for correctness: caller passes the item's main table key. - // We already marshal item data, just use the full data hash. - // For simplicity in the emulator, use a fixed hash from item data. - suffix := uniqueSuffix(data) - var gsiKey string - if gsiSK != "" { - gsiKey = gsiPrefix + tableName + "/" + idx.IndexName + "/" + gsiPK + "/" + gsiSK + "/" + suffix - } else { - gsiKey = gsiPrefix + tableName + "/" + idx.IndexName + "/" + gsiPK + "/_/" + suffix - } - return txn.Set([]byte(gsiKey), data) -} - -// uniqueSuffix creates a short unique string from item data to prevent GSI key collisions. -func uniqueSuffix(data []byte) string { - // Simple FNV-like hash for uniqueness. - var h uint32 = 2166136261 - for _, b := range data { - h ^= uint32(b) - h *= 16777619 - } - return fmt.Sprintf("%08x", h) -} - -// ExecTransaction executes a function within a BadgerDB write transaction. -func (s *DynamoStore) ExecTransaction(fn func(txn *badger.Txn) error) error { - return s.db.Update(fn) -} - -// PutItemTxn stores an item within an existing BadgerDB transaction. -func (s *DynamoStore) PutItemTxn(txn *badger.Txn, tableName string, item Item) error { - s.mu.RLock() - table, exists := s.tables[tableName] - s.mu.RUnlock() - if !exists { - return ErrTableNotFound - } - - pkAttr, ok := item[table.PartitionKey.Name] - if !ok { - return fmt.Errorf("item missing partition key attribute %q", table.PartitionKey.Name) - } - pkVal, err := attributeStringValue(pkAttr) - if err != nil { - return fmt.Errorf("partition key value: %w", err) - } - - var key []byte - if table.SortKey != nil { - skAttr, ok := item[table.SortKey.Name] - if !ok { - return fmt.Errorf("item missing sort key attribute %q", table.SortKey.Name) - } - skVal, err := attributeStringValue(skAttr) - if err != nil { - return fmt.Errorf("sort key value: %w", err) - } - key = itemKey(tableName, pkVal, skVal) - } else { - key = itemKey(tableName, pkVal) - } - - data, err := json.Marshal(item) - if err != nil { - return fmt.Errorf("marshal item: %w", err) - } - return txn.Set(key, data) -} - -// DeleteItemTxn removes an item within an existing BadgerDB transaction. -func (s *DynamoStore) DeleteItemTxn(txn *badger.Txn, tableName string, key Item) error { - s.mu.RLock() - table, exists := s.tables[tableName] - s.mu.RUnlock() - if !exists { - return ErrTableNotFound - } - - pkAttr, ok := key[table.PartitionKey.Name] - if !ok { - return fmt.Errorf("key missing partition key attribute %q", table.PartitionKey.Name) - } - pkVal, err := attributeStringValue(pkAttr) - if err != nil { - return fmt.Errorf("partition key value: %w", err) - } - - var dbKey []byte - if table.SortKey != nil { - skAttr, ok := key[table.SortKey.Name] - if !ok { - return fmt.Errorf("key missing sort key attribute %q", table.SortKey.Name) - } - skVal, err := attributeStringValue(skAttr) - if err != nil { - return fmt.Errorf("sort key value: %w", err) - } - dbKey = itemKey(tableName, pkVal, skVal) - } else { - dbKey = itemKey(tableName, pkVal) - } - return txn.Delete(dbKey) -} - -// GetItemTxn retrieves an item within an existing BadgerDB transaction. -func (s *DynamoStore) GetItemTxn(txn *badger.Txn, tableName string, key Item) (*Item, error) { - s.mu.RLock() - table, exists := s.tables[tableName] - s.mu.RUnlock() - if !exists { - return nil, ErrTableNotFound - } - - pkAttr, ok := key[table.PartitionKey.Name] - if !ok { - return nil, fmt.Errorf("key missing partition key attribute %q", table.PartitionKey.Name) - } - pkVal, err := attributeStringValue(pkAttr) - if err != nil { - return nil, fmt.Errorf("partition key value: %w", err) - } - - var dbKey []byte - if table.SortKey != nil { - skAttr, ok := key[table.SortKey.Name] - if !ok { - return nil, fmt.Errorf("key missing sort key attribute %q", table.SortKey.Name) - } - skVal, err := attributeStringValue(skAttr) - if err != nil { - return nil, fmt.Errorf("sort key value: %w", err) - } - dbKey = itemKey(tableName, pkVal, skVal) - } else { - dbKey = itemKey(tableName, pkVal) - } - - var result Item - dbItem, err := txn.Get(dbKey) - if err != nil { - if errors.Is(err, badger.ErrKeyNotFound) { - return nil, ErrItemNotFound - } - return nil, err - } - if err := dbItem.Value(func(v []byte) error { - return json.Unmarshal(v, &result) - }); err != nil { - return nil, err + return err } - return &result, nil + _, err = s.db.DB().Exec( + `INSERT OR REPLACE INTO ddb_tags (resource_arn, tags) VALUES (?, ?)`, resourceArn, string(data)) + return err } From 3c1239f31cf9897ef4eb67b0ab0e32f45e40703d Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Fri, 17 Jul 2026 23:23:03 +0900 Subject: [PATCH 2/5] refactor(dynamodb): make DeleteTable atomic; document QueryGSI trade-offs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address code review on the sqlite store: - DeleteTable now removes items + metadata inside a single transaction, so a mid-way failure can't leave orphaned rows (was two separate Exec calls). - QueryGSI: document the O(n) scan ceiling + upgrade path, and why an unknown index returns an empty result rather than an error (matches prior behavior; an error would surface as an opaque 500 — a ValidationException belongs in the provider). Verified: go build ./..., go test ./... (108 ok, 0 fail). --- internal/services/dynamodb/store.go | 25 ++++++++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/internal/services/dynamodb/store.go b/internal/services/dynamodb/store.go index da9be089..a748d14e 100644 --- a/internal/services/dynamodb/store.go +++ b/internal/services/dynamodb/store.go @@ -245,12 +245,23 @@ func (s *DynamoStore) DeleteTable(name string) error { return ErrTableNotFound } - if _, err := s.db.DB().Exec(`DELETE FROM ddb_items WHERE table_name = ?`, name); err != nil { + // Delete items and metadata atomically so a mid-way failure can't leave + // orphaned rows. + tx, err := s.db.DB().Begin() + if err != nil { + return err + } + if _, err := tx.Exec(`DELETE FROM ddb_items WHERE table_name = ?`, name); err != nil { + _ = tx.Rollback() return fmt.Errorf("delete table items: %w", err) } - if _, err := s.db.DB().Exec(`DELETE FROM ddb_tables WHERE name = ?`, name); err != nil { + if _, err := tx.Exec(`DELETE FROM ddb_tables WHERE name = ?`, name); err != nil { + _ = tx.Rollback() return fmt.Errorf("delete table metadata: %w", err) } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit delete table: %w", err) + } delete(s.tables, name) return nil @@ -414,6 +425,10 @@ func (s *DynamoStore) Scan(tableName string) ([]Item, error) { // QueryGSI returns items whose value for the index's HASH key attribute equals // pkValue. Secondary indexes are projected on the fly from the base table. +// +// ponytail: O(n) full-table scan filtered in memory — fine at emulator scale. +// If a table ever holds enough items to matter, index the GSI hash attribute +// with a generated column (json_extract) and query it directly. func (s *DynamoStore) QueryGSI(tableName, indexName, pkValue string) ([]Item, error) { s.mu.RLock() table, exists := s.tables[tableName] @@ -424,7 +439,11 @@ func (s *DynamoStore) QueryGSI(tableName, indexName, pkValue string) ([]Item, er hashAttr := indexHashKey(table, indexName) if hashAttr == "" { - return nil, nil // unknown index — no matches + // Unknown index → empty result, not an error. This matches the prior + // behavior and how the provider surfaces it (an empty Query response). + // Returning an error would become an opaque 500; a real + // ValidationException belongs in the provider, not the store. + return nil, nil } all, err := s.queryItems(`SELECT data FROM ddb_items WHERE table_name = ?`, []any{tableName}) From b9ca50d0ea9d347465dd6dfae0b08a68ea7fed8d Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Fri, 17 Jul 2026 23:33:24 +0900 Subject: [PATCH 3/5] perf(dynamodb): index QueryGSI via a maintained projection table Resolve the ponytail debt in QueryGSI: instead of scanning every item in the table and filtering in memory, maintain a ddb_gsi projection (keyed by base-item identity) that PutItem/DeleteItem keep in sync, and serve QueryGSI with an indexed lookup on (table_name, index_name, gsi_pk). - correct for all key types (stores the same attributeStringValue used on the query side), and re-synced per write so an update that changes/removes an index key can't leave stale rows (the bug the old badger projection had) - GSIs are immutable after CreateTable in the provider, so no rebuild-on-update is needed - add TestDynamoStore_QueryGSI (query, unknown-index, update re-sync, delete); QueryGSI previously had no store-level test Verified: go vet, go test ./... (108 ok, 0 fail). --- internal/services/dynamodb/store.go | 142 +++++++++++++++-------- internal/services/dynamodb/store_test.go | 40 +++++++ 2 files changed, 135 insertions(+), 47 deletions(-) diff --git a/internal/services/dynamodb/store.go b/internal/services/dynamodb/store.go index a748d14e..9a536d68 100644 --- a/internal/services/dynamodb/store.go +++ b/internal/services/dynamodb/store.go @@ -105,6 +105,22 @@ var migrations = []sqlite.Migration{ tags TEXT NOT NULL ); `}, + {Version: 2, SQL: ` + -- Denormalized projection of secondary indexes, keyed by the base-table + -- item identity so an item's rows can be re-synced on every write. Lets + -- QueryGSI do an indexed lookup instead of scanning the whole table. + CREATE TABLE IF NOT EXISTS ddb_gsi ( + table_name TEXT NOT NULL, + index_name TEXT NOT NULL, + gsi_pk TEXT NOT NULL, + item_pk TEXT NOT NULL, + item_sk TEXT NOT NULL DEFAULT '', + data TEXT NOT NULL, + PRIMARY KEY (table_name, index_name, item_pk, item_sk) + ); + CREATE INDEX IF NOT EXISTS idx_ddb_gsi_lookup + ON ddb_gsi (table_name, index_name, gsi_pk); + `}, } // DynamoStore is a SQLite-backed store for DynamoDB tables and items. Table @@ -255,6 +271,10 @@ func (s *DynamoStore) DeleteTable(name string) error { _ = tx.Rollback() return fmt.Errorf("delete table items: %w", err) } + if _, err := tx.Exec(`DELETE FROM ddb_gsi WHERE table_name = ?`, name); err != nil { + _ = tx.Rollback() + return fmt.Errorf("delete table index rows: %w", err) + } if _, err := tx.Exec(`DELETE FROM ddb_tables WHERE name = ?`, name); err != nil { _ = tx.Rollback() return fmt.Errorf("delete table metadata: %w", err) @@ -329,10 +349,54 @@ func (s *DynamoStore) PutItem(tableName string, item Item) error { return fmt.Errorf("marshal item: %w", err) } - _, err = s.db.DB().Exec( + // Upsert the item and re-sync its secondary-index projection atomically. + tx, err := s.db.DB().Begin() + if err != nil { + return err + } + if _, err := tx.Exec( `INSERT OR REPLACE INTO ddb_items (table_name, pk, sk, data) VALUES (?, ?, ?, ?)`, - tableName, pk, sk, string(data)) - return err + tableName, pk, sk, string(data)); err != nil { + _ = tx.Rollback() + return err + } + if err := syncItemGSI(tx, table, pk, sk, item, string(data)); err != nil { + _ = tx.Rollback() + return err + } + return tx.Commit() +} + +// syncItemGSI rebuilds the ddb_gsi rows for a single item: it drops the item's +// existing projections (so an update that changes or removes an index key can't +// leave stale rows) and re-inserts one row per index whose HASH key attribute +// is present on the item. +func syncItemGSI(tx *sql.Tx, table *TableInfo, itemPK, itemSK string, item Item, data string) error { + if _, err := tx.Exec( + `DELETE FROM ddb_gsi WHERE table_name = ? AND item_pk = ? AND item_sk = ?`, + table.Name, itemPK, itemSK); err != nil { + return err + } + for _, idx := range append(table.GlobalSecondaryIndexes, table.LocalSecondaryIndexes...) { + hashAttr := indexHashKeyName(idx) + if hashAttr == "" { + continue + } + av, ok := item[hashAttr] + if !ok { + continue + } + gsiPK, err := attributeStringValue(av) + if err != nil { + continue + } + if _, err := tx.Exec( + `INSERT OR REPLACE INTO ddb_gsi (table_name, index_name, gsi_pk, item_pk, item_sk, data) VALUES (?, ?, ?, ?, ?, ?)`, + table.Name, idx.IndexName, gsiPK, itemPK, itemSK, data); err != nil { + return err + } + } + return nil } // GetItem retrieves an item by its key attributes. Returns ErrItemNotFound if missing. @@ -381,15 +445,27 @@ func (s *DynamoStore) DeleteItem(tableName string, key Item) error { return err } - res, err := s.db.DB().Exec( + tx, err := s.db.DB().Begin() + if err != nil { + return err + } + res, err := tx.Exec( `DELETE FROM ddb_items WHERE table_name = ? AND pk = ? AND sk = ?`, tableName, pk, sk) if err != nil { + _ = tx.Rollback() return err } if n, _ := res.RowsAffected(); n == 0 { + _ = tx.Rollback() return ErrItemNotFound } - return nil + if _, err := tx.Exec( + `DELETE FROM ddb_gsi WHERE table_name = ? AND item_pk = ? AND item_sk = ?`, + tableName, pk, sk); err != nil { + _ = tx.Rollback() + return err + } + return tx.Commit() } // Query returns items with the given partition key value, ordered by sort key. @@ -424,56 +500,28 @@ func (s *DynamoStore) Scan(tableName string) ([]Item, error) { } // QueryGSI returns items whose value for the index's HASH key attribute equals -// pkValue. Secondary indexes are projected on the fly from the base table. -// -// ponytail: O(n) full-table scan filtered in memory — fine at emulator scale. -// If a table ever holds enough items to matter, index the GSI hash attribute -// with a generated column (json_extract) and query it directly. +// pkValue, via an indexed lookup on the ddb_gsi projection maintained by +// PutItem/DeleteItem. An unknown index simply has no rows and yields an empty +// result (not an error) — matching how the provider surfaces it as an empty +// Query response; a real ValidationException would belong in the provider. func (s *DynamoStore) QueryGSI(tableName, indexName, pkValue string) ([]Item, error) { s.mu.RLock() - table, exists := s.tables[tableName] + _, exists := s.tables[tableName] s.mu.RUnlock() if !exists { return nil, ErrTableNotFound } - - hashAttr := indexHashKey(table, indexName) - if hashAttr == "" { - // Unknown index → empty result, not an error. This matches the prior - // behavior and how the provider surfaces it (an empty Query response). - // Returning an error would become an opaque 500; a real - // ValidationException belongs in the provider, not the store. - return nil, nil - } - - all, err := s.queryItems(`SELECT data FROM ddb_items WHERE table_name = ?`, []any{tableName}) - if err != nil { - return nil, err - } - var items []Item - for _, item := range all { - av, ok := item[hashAttr] - if !ok { - continue - } - if v, err := attributeStringValue(av); err == nil && v == pkValue { - items = append(items, item) - } - } - return items, nil + return s.queryItems( + `SELECT data FROM ddb_gsi WHERE table_name = ? AND index_name = ? AND gsi_pk = ?`, + []any{tableName, indexName, pkValue}) } -// indexHashKey returns the HASH key attribute name for the named GSI or LSI, -// or "" if no such index exists. -func indexHashKey(table *TableInfo, indexName string) string { - for _, idx := range append(table.GlobalSecondaryIndexes, table.LocalSecondaryIndexes...) { - if idx.IndexName != indexName { - continue - } - for _, ks := range idx.KeySchema { - if ks.KeyType == "HASH" { - return ks.Name - } +// indexHashKeyName returns the HASH key attribute name for an index, or "" if +// the index has no HASH key. +func indexHashKeyName(idx IndexDef) string { + for _, ks := range idx.KeySchema { + if ks.KeyType == "HASH" { + return ks.Name } } return "" diff --git a/internal/services/dynamodb/store_test.go b/internal/services/dynamodb/store_test.go index 37330539..061a3c6f 100644 --- a/internal/services/dynamodb/store_test.go +++ b/internal/services/dynamodb/store_test.go @@ -155,3 +155,43 @@ func TestDynamoStore_DeleteTable(t *testing.T) { assert.NotContains(t, tables, "Temp") assert.Empty(t, tables) } + +// TestDynamoStore_QueryGSI verifies indexed queries against a secondary index, +// including that updates and deletes keep the projection in sync. +func TestDynamoStore_QueryGSI(t *testing.T) { + store := newTestStore(t) + + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Users", + PartitionKey: KeyDef{Name: "UserID", Type: "S"}, + GlobalSecondaryIndexes: []IndexDef{{ + IndexName: "ByEmail", + KeySchema: []KeyDef{{Name: "Email", Type: "S", KeyType: "HASH"}}, + }}, + })) + + require.NoError(t, store.PutItem("Users", Item{"UserID": {S: strPtr("u1")}, "Email": {S: strPtr("a@x.com")}})) + require.NoError(t, store.PutItem("Users", Item{"UserID": {S: strPtr("u2")}, "Email": {S: strPtr("a@x.com")}})) + require.NoError(t, store.PutItem("Users", Item{"UserID": {S: strPtr("u3")}, "Email": {S: strPtr("b@x.com")}})) + + got, err := store.QueryGSI("Users", "ByEmail", "a@x.com") + require.NoError(t, err) + assert.Len(t, got, 2) + + // Unknown index → empty result, not an error. + none, err := store.QueryGSI("Users", "Nope", "a@x.com") + require.NoError(t, err) + assert.Empty(t, none) + + // Updating an item's index key re-syncs its projection (no stale rows). + require.NoError(t, store.PutItem("Users", Item{"UserID": {S: strPtr("u1")}, "Email": {S: strPtr("b@x.com")}})) + got, err = store.QueryGSI("Users", "ByEmail", "a@x.com") + require.NoError(t, err) + assert.Len(t, got, 1) // only u2 remains under a@x.com + + // Deleting an item removes it from the index too. + require.NoError(t, store.DeleteItem("Users", Item{"UserID": {S: strPtr("u2")}})) + got, err = store.QueryGSI("Users", "ByEmail", "a@x.com") + require.NoError(t, err) + assert.Empty(t, got) +} From 27dde93352434a45870d4824547d5d138dae8329 Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Fri, 17 Jul 2026 23:43:10 +0900 Subject: [PATCH 4/5] test(dynamodb): cover QueryGSI edge cases and key-value helpers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review: exercise the new SQLite-backed paths beyond the happy path. - QueryGSI: unknown table (ErrTableNotFound), local secondary index, an index with HASH+RANGE keys where items share a gsi_pk but differ by base key, and a binary (B) hash key — confirming the projection is correct for all key types - keyValues: missing partition/sort key errors on PutItem/GetItem/DeleteItem - numeric (N) partition key round-trip - escapeLike: sort-key prefix with %, _, \ treated as literals go test ./... → 108 ok, 0 fail. --- internal/services/dynamodb/store_test.go | 157 +++++++++++++++++++++++ 1 file changed, 157 insertions(+) diff --git a/internal/services/dynamodb/store_test.go b/internal/services/dynamodb/store_test.go index 061a3c6f..7d02e57e 100644 --- a/internal/services/dynamodb/store_test.go +++ b/internal/services/dynamodb/store_test.go @@ -195,3 +195,160 @@ func TestDynamoStore_QueryGSI(t *testing.T) { require.NoError(t, err) assert.Empty(t, got) } + +// TestDynamoStore_QueryGSIVariants covers QueryGSI edge cases: an unknown table, +// a local secondary index, an index with HASH+RANGE keys, and a binary key. +func TestDynamoStore_QueryGSIVariants(t *testing.T) { + t.Run("missing table returns ErrTableNotFound", func(t *testing.T) { + store := newTestStore(t) + _, err := store.QueryGSI("Nope", "ByEmail", "x") + assert.ErrorIs(t, err, ErrTableNotFound) + }) + + t.Run("local secondary index is queryable", func(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Events", + PartitionKey: KeyDef{Name: "UserID", Type: "S"}, + SortKey: &KeyDef{Name: "Ts", Type: "S"}, + LocalSecondaryIndexes: []IndexDef{{ + IndexName: "ByScore", + KeySchema: []KeyDef{ + {Name: "UserID", Type: "S", KeyType: "HASH"}, + {Name: "Score", Type: "N", KeyType: "RANGE"}, + }, + }}, + })) + require.NoError(t, store.PutItem("Events", Item{"UserID": {S: strPtr("u1")}, "Ts": {S: strPtr("t1")}, "Score": {N: strPtr("10")}})) + require.NoError(t, store.PutItem("Events", Item{"UserID": {S: strPtr("u1")}, "Ts": {S: strPtr("t2")}, "Score": {N: strPtr("20")}})) + require.NoError(t, store.PutItem("Events", Item{"UserID": {S: strPtr("u2")}, "Ts": {S: strPtr("t3")}, "Score": {N: strPtr("30")}})) + + got, err := store.QueryGSI("Events", "ByScore", "u1") + require.NoError(t, err) + assert.Len(t, got, 2) + }) + + t.Run("index with HASH and RANGE keys, items vary by base sort key", func(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Messages", + PartitionKey: KeyDef{Name: "MsgID", Type: "S"}, + SortKey: &KeyDef{Name: "Seq", Type: "S"}, + GlobalSecondaryIndexes: []IndexDef{{ + IndexName: "ByConversation", + KeySchema: []KeyDef{ + {Name: "ConvID", Type: "S", KeyType: "HASH"}, + {Name: "CreatedAt", Type: "N", KeyType: "RANGE"}, + }, + }}, + })) + // Two distinct items sharing the GSI hash key but differing in base key. + require.NoError(t, store.PutItem("Messages", Item{"MsgID": {S: strPtr("m1")}, "Seq": {S: strPtr("1")}, "ConvID": {S: strPtr("c1")}, "CreatedAt": {N: strPtr("100")}})) + require.NoError(t, store.PutItem("Messages", Item{"MsgID": {S: strPtr("m1")}, "Seq": {S: strPtr("2")}, "ConvID": {S: strPtr("c1")}, "CreatedAt": {N: strPtr("200")}})) + require.NoError(t, store.PutItem("Messages", Item{"MsgID": {S: strPtr("m2")}, "Seq": {S: strPtr("1")}, "ConvID": {S: strPtr("c2")}, "CreatedAt": {N: strPtr("300")}})) + + got, err := store.QueryGSI("Messages", "ByConversation", "c1") + require.NoError(t, err) + assert.Len(t, got, 2) // both c1 items projected, no collision on same gsi_pk + }) + + t.Run("binary hash key", func(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Blobs", + PartitionKey: KeyDef{Name: "BlobID", Type: "S"}, + GlobalSecondaryIndexes: []IndexDef{{ + IndexName: "ByDigest", + KeySchema: []KeyDef{{Name: "Digest", Type: "B", KeyType: "HASH"}}, + }}, + })) + d1 := []byte{0x01, 0x02, 0x03} + d2 := []byte{0x04, 0x05, 0x06} + require.NoError(t, store.PutItem("Blobs", Item{"BlobID": {S: strPtr("b1")}, "Digest": {B: d1}})) + require.NoError(t, store.PutItem("Blobs", Item{"BlobID": {S: strPtr("b2")}, "Digest": {B: d1}})) + require.NoError(t, store.PutItem("Blobs", Item{"BlobID": {S: strPtr("b3")}, "Digest": {B: d2}})) + + got, err := store.QueryGSI("Blobs", "ByDigest", string(d1)) + require.NoError(t, err) + assert.Len(t, got, 2) + }) +} + +// TestDynamoStore_KeyValidationErrors verifies keyValues surfaces missing-key +// errors on the item write/read/delete paths. +func TestDynamoStore_KeyValidationErrors(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Orders", + PartitionKey: KeyDef{Name: "OrderID", Type: "S"}, + SortKey: &KeyDef{Name: "LineID", Type: "S"}, + })) + + t.Run("PutItem missing partition key", func(t *testing.T) { + err := store.PutItem("Orders", Item{"LineID": {S: strPtr("L1")}}) + require.Error(t, err) + assert.Contains(t, err.Error(), "missing partition key attribute") + }) + t.Run("PutItem missing sort key", func(t *testing.T) { + err := store.PutItem("Orders", Item{"OrderID": {S: strPtr("O1")}}) + require.Error(t, err) + assert.Contains(t, err.Error(), "missing sort key attribute") + }) + t.Run("GetItem missing partition key", func(t *testing.T) { + _, err := store.GetItem("Orders", Item{"LineID": {S: strPtr("L1")}}) + require.Error(t, err) + assert.Contains(t, err.Error(), "missing partition key attribute") + }) + t.Run("DeleteItem missing sort key", func(t *testing.T) { + err := store.DeleteItem("Orders", Item{"OrderID": {S: strPtr("O1")}}) + require.Error(t, err) + assert.Contains(t, err.Error(), "missing sort key attribute") + }) +} + +// TestDynamoStore_NumericKey verifies non-string (N) key attributes round-trip. +func TestDynamoStore_NumericKey(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Accounts", + PartitionKey: KeyDef{Name: "AccountID", Type: "N"}, + })) + require.NoError(t, store.PutItem("Accounts", Item{"AccountID": {N: strPtr("1001")}, "Name": {S: strPtr("Primary")}})) + + got, err := store.GetItem("Accounts", Item{"AccountID": {N: strPtr("1001")}}) + require.NoError(t, err) + require.NotNil(t, got) + assert.Equal(t, strPtr("Primary"), (*got)["Name"].S) +} + +// TestDynamoStore_QuerySortKeyPrefixEscaping verifies escapeLike treats LIKE +// metacharacters (%, _, \) in a sort-key prefix as literals. +func TestDynamoStore_QuerySortKeyPrefixEscaping(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Files", + PartitionKey: KeyDef{Name: "UserID", Type: "S"}, + SortKey: &KeyDef{Name: "Path", Type: "S"}, + })) + for _, p := range []string{"a%b", "a_b", "axb", `a\z`} { + require.NoError(t, store.PutItem("Files", Item{"UserID": {S: strPtr("u1")}, "Path": {S: strPtr(p)}})) + } + + tests := []struct { + name string + prefix string + want int + }{ + {"no special chars matches all", "a", 4}, + {"literal percent", "a%", 1}, + {"literal underscore", "a_", 1}, + {"literal backslash", `a\`, 1}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := store.Query("Files", "u1", tt.prefix) + require.NoError(t, err) + assert.Len(t, got, tt.want) + }) + } +} From c2fded448b0ed4f3eb9c11dadc98d5a5e6a3e56c Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Sat, 18 Jul 2026 00:57:26 +0900 Subject: [PATCH 5/5] fix(dynamodb): correct SQLite-backed GSI/Query regressions Address code-review findings on the badger->SQLite refactor: - Query numeric ('N') sort keys numerically, not lexicographically - Project composite-GSI items only when all index key attrs are present - Order QueryGSI by the index sort key via a new gsi_sk column (migration v3) - Propagate unsupported index-key type errors instead of silently skipping - Hold the read lock across PutItem/DeleteItem writes so a racing DeleteTable can't orphan rows that resurrect on a same-name recreate - Delete ddb_ttl/ddb_tags rows in DeleteTable so a recreated table starts clean - UpdateTable persists before mutating the cache (no rollback-on-failure drift) - begins_with prefix matches case-sensitively via a byte range, not LIKE - Serialize the tag read-modify-write in PutTags/RemoveTags - Build a fresh index slice instead of appending into the shared cached array - Skip the ddb_gsi delete on writes to tables without secondary indexes --- internal/services/dynamodb/store.go | 204 ++++++++++++++++++----- internal/services/dynamodb/store_test.go | 142 +++++++++++++++- 2 files changed, 300 insertions(+), 46 deletions(-) diff --git a/internal/services/dynamodb/store.go b/internal/services/dynamodb/store.go index 9a536d68..e1b7c4c4 100644 --- a/internal/services/dynamodb/store.go +++ b/internal/services/dynamodb/store.go @@ -8,7 +8,6 @@ import ( "errors" "fmt" "path/filepath" - "strings" "sync" "time" @@ -121,6 +120,12 @@ var migrations = []sqlite.Migration{ CREATE INDEX IF NOT EXISTS idx_ddb_gsi_lookup ON ddb_gsi (table_name, index_name, gsi_pk); `}, + {Version: 3, SQL: ` + -- Index RANGE-key value, so QueryGSI can return items ordered by the + -- index sort key (base-table item_pk/item_sk are the row identity, not + -- the index order). + ALTER TABLE ddb_gsi ADD COLUMN gsi_sk TEXT NOT NULL DEFAULT ''; + `}, } // DynamoStore is a SQLite-backed store for DynamoDB tables and items. Table @@ -129,6 +134,7 @@ type DynamoStore struct { db *sqlite.Store tables map[string]*TableInfo mu sync.RWMutex + tagsMu sync.Mutex // serializes the tag read-modify-write in PutTags/RemoveTags } // NewDynamoStore opens (or creates) a SQLite database under dir and loads @@ -257,12 +263,14 @@ func (s *DynamoStore) DeleteTable(name string) error { s.mu.Lock() defer s.mu.Unlock() - if _, exists := s.tables[name]; !exists { + info, exists := s.tables[name] + if !exists { return ErrTableNotFound } - // Delete items and metadata atomically so a mid-way failure can't leave - // orphaned rows. + // Delete items, index rows, and metadata atomically so a mid-way failure + // can't leave orphaned rows — and so a recreated same-name table doesn't + // inherit the deleted table's TTL config or tags. tx, err := s.db.DB().Begin() if err != nil { return err @@ -275,6 +283,14 @@ func (s *DynamoStore) DeleteTable(name string) error { _ = tx.Rollback() return fmt.Errorf("delete table index rows: %w", err) } + if _, err := tx.Exec(`DELETE FROM ddb_ttl WHERE table_name = ?`, name); err != nil { + _ = tx.Rollback() + return fmt.Errorf("delete table ttl config: %w", err) + } + if _, err := tx.Exec(`DELETE FROM ddb_tags WHERE resource_arn = ?`, info.TableArn); err != nil { + _ = tx.Rollback() + return fmt.Errorf("delete table tags: %w", err) + } if _, err := tx.Exec(`DELETE FROM ddb_tables WHERE name = ?`, name); err != nil { _ = tx.Rollback() return fmt.Errorf("delete table metadata: %w", err) @@ -320,21 +336,30 @@ func (s *DynamoStore) UpdateTable(name string, updates func(*TableInfo)) error { if !exists { return ErrTableNotFound } - updates(info) - data, err := json.Marshal(info) + // Apply updates to a copy and persist before mutating the cache, so a failed + // write doesn't leave the in-memory metadata ahead of what's on disk. + updated := *info + updates(&updated) + data, err := json.Marshal(&updated) if err != nil { return fmt.Errorf("marshal table info: %w", err) } - _, err = s.db.DB().Exec(`UPDATE ddb_tables SET info = ? WHERE name = ?`, string(data), name) - return err + if _, err := s.db.DB().Exec(`UPDATE ddb_tables SET info = ? WHERE name = ?`, string(data), name); err != nil { + return err + } + *info = updated + return nil } // PutItem stores (upserts) an item in the specified table. The item key is // derived from the table's partition key (and optional sort key) values. func (s *DynamoStore) PutItem(tableName string, item Item) error { + // Hold the read lock across the DB write so a concurrent DeleteTable (which + // takes the write lock) can't delete the table's rows mid-write and leave + // this put's item/index rows orphaned. s.mu.RLock() + defer s.mu.RUnlock() table, exists := s.tables[tableName] - s.mu.RUnlock() if !exists { return ErrTableNotFound } @@ -369,30 +394,37 @@ func (s *DynamoStore) PutItem(tableName string, item Item) error { // syncItemGSI rebuilds the ddb_gsi rows for a single item: it drops the item's // existing projections (so an update that changes or removes an index key can't -// leave stale rows) and re-inserts one row per index whose HASH key attribute -// is present on the item. +// leave stale rows) and re-inserts one row per index for which the item carries +// every index key attribute. An item missing an index HASH or RANGE attribute is +// omitted from that index (a sparse index), matching DynamoDB; a present-but-non- +// scalar key value is a hard error, not a silent skip. func syncItemGSI(tx *sql.Tx, table *TableInfo, itemPK, itemSK string, item Item, data string) error { + if !hasSecondaryIndexes(table) { + return nil + } if _, err := tx.Exec( `DELETE FROM ddb_gsi WHERE table_name = ? AND item_pk = ? AND item_sk = ?`, table.Name, itemPK, itemSK); err != nil { return err } - for _, idx := range append(table.GlobalSecondaryIndexes, table.LocalSecondaryIndexes...) { - hashAttr := indexHashKeyName(idx) - if hashAttr == "" { - continue + for _, idx := range allIndexes(table) { + gsiPK, ok, err := indexKeyValue(item, idx, "HASH") + if err != nil { + return err } - av, ok := item[hashAttr] if !ok { - continue + continue // sparse index: item lacks the index HASH attribute } - gsiPK, err := attributeStringValue(av) + gsiSK, ok, err := indexKeyValue(item, idx, "RANGE") if err != nil { - continue + return err + } + if indexKey(idx, "RANGE") != nil && !ok { + continue // composite index: item lacks the index RANGE attribute } if _, err := tx.Exec( - `INSERT OR REPLACE INTO ddb_gsi (table_name, index_name, gsi_pk, item_pk, item_sk, data) VALUES (?, ?, ?, ?, ?, ?)`, - table.Name, idx.IndexName, gsiPK, itemPK, itemSK, data); err != nil { + `INSERT OR REPLACE INTO ddb_gsi (table_name, index_name, gsi_pk, gsi_sk, item_pk, item_sk, data) VALUES (?, ?, ?, ?, ?, ?, ?)`, + table.Name, idx.IndexName, gsiPK, gsiSK, itemPK, itemSK, data); err != nil { return err } } @@ -433,9 +465,11 @@ func (s *DynamoStore) GetItem(tableName string, key Item) (*Item, error) { // DeleteItem removes an item by its key attributes. Returns ErrItemNotFound if missing. func (s *DynamoStore) DeleteItem(tableName string, key Item) error { + // Hold the read lock across the DB write for the same reason as PutItem: keep + // a concurrent DeleteTable from racing this delete's transaction. s.mu.RLock() + defer s.mu.RUnlock() table, exists := s.tables[tableName] - s.mu.RUnlock() if !exists { return ErrTableNotFound } @@ -459,11 +493,13 @@ func (s *DynamoStore) DeleteItem(tableName string, key Item) error { _ = tx.Rollback() return ErrItemNotFound } - if _, err := tx.Exec( - `DELETE FROM ddb_gsi WHERE table_name = ? AND item_pk = ? AND item_sk = ?`, - tableName, pk, sk); err != nil { - _ = tx.Rollback() - return err + if hasSecondaryIndexes(table) { + if _, err := tx.Exec( + `DELETE FROM ddb_gsi WHERE table_name = ? AND item_pk = ? AND item_sk = ?`, + tableName, pk, sk); err != nil { + _ = tx.Rollback() + return err + } } return tx.Commit() } @@ -480,11 +516,22 @@ func (s *DynamoStore) Query(tableName string, partitionKeyValue string, sortKeyP query := `SELECT data FROM ddb_items WHERE table_name = ? AND pk = ?` args := []any{tableName, partitionKeyValue} - if table.SortKey != nil && sortKeyPrefix != "" { - query += ` AND sk LIKE ? ESCAPE '\'` - args = append(args, escapeLike(sortKeyPrefix)+"%") + sortKeyType := "" + if table.SortKey != nil { + sortKeyType = table.SortKey.Type + if sortKeyPrefix != "" { + // Half-open byte range rather than LIKE: SQLite's LIKE is ASCII + // case-insensitive, but a begins_with sort-key prefix must match + // case-sensitively (and the range uses the primary-key index). + query += ` AND sk >= ?` + args = append(args, sortKeyPrefix) + if upper := prefixUpperBound(sortKeyPrefix); upper != "" { + query += ` AND sk < ?` + args = append(args, upper) + } + } } - query += ` ORDER BY sk` + query += ` ORDER BY ` + orderByColumn("sk", sortKeyType) return s.queryItems(query, args) } @@ -506,22 +553,92 @@ func (s *DynamoStore) Scan(tableName string) ([]Item, error) { // Query response; a real ValidationException would belong in the provider. func (s *DynamoStore) QueryGSI(tableName, indexName, pkValue string) ([]Item, error) { s.mu.RLock() - _, exists := s.tables[tableName] + table, exists := s.tables[tableName] s.mu.RUnlock() if !exists { return nil, ErrTableNotFound } + rangeType := "" + for _, idx := range allIndexes(table) { + if idx.IndexName == indexName { + if rk := indexKey(idx, "RANGE"); rk != nil { + rangeType = rk.Type + } + break + } + } return s.queryItems( - `SELECT data FROM ddb_gsi WHERE table_name = ? AND index_name = ? AND gsi_pk = ?`, + `SELECT data FROM ddb_gsi WHERE table_name = ? AND index_name = ? AND gsi_pk = ? ORDER BY `+orderByColumn("gsi_sk", rangeType), []any{tableName, indexName, pkValue}) } -// indexHashKeyName returns the HASH key attribute name for an index, or "" if -// the index has no HASH key. -func indexHashKeyName(idx IndexDef) string { - for _, ks := range idx.KeySchema { - if ks.KeyType == "HASH" { - return ks.Name +// hasSecondaryIndexes reports whether the table has any secondary index. +func hasSecondaryIndexes(table *TableInfo) bool { + return len(table.GlobalSecondaryIndexes) > 0 || len(table.LocalSecondaryIndexes) > 0 +} + +// allIndexes returns the table's global and local secondary indexes as a fresh +// slice. It never aliases the cached TableInfo's backing arrays, so callers can +// range over it concurrently without racing on a shared append. +func allIndexes(table *TableInfo) []IndexDef { + out := make([]IndexDef, 0, len(table.GlobalSecondaryIndexes)+len(table.LocalSecondaryIndexes)) + out = append(out, table.GlobalSecondaryIndexes...) + out = append(out, table.LocalSecondaryIndexes...) + return out +} + +// indexKey returns the index's key definition of the given kind ("HASH" or +// "RANGE"), or nil if the index has no such key. +func indexKey(idx IndexDef, kind string) *KeyDef { + for i := range idx.KeySchema { + if idx.KeySchema[i].KeyType == kind { + return &idx.KeySchema[i] + } + } + return nil +} + +// indexKeyValue returns the item's string value for the index key of the given +// kind. ok is false when the index has no such key or the item lacks the +// attribute (a sparse index); a present-but-non-scalar value is an error. +func indexKeyValue(item Item, idx IndexDef, kind string) (value string, ok bool, err error) { + key := indexKey(idx, kind) + if key == nil { + return "", false, nil + } + av, present := item[key.Name] + if !present { + return "", false, nil + } + v, err := attributeStringValue(av) + if err != nil { + return "", false, fmt.Errorf("index %q key %q: %w", idx.IndexName, key.Name, err) + } + return v, true, nil +} + +// orderByColumn returns the ORDER BY expression for a sort/range key column of +// the given attribute type. Numeric ('N') keys must sort numerically, not by the +// lexicographic byte order of their TEXT storage (so 2, 10 sorts before 100). +// ponytail: CAST AS REAL loses precision past ~2^53 and on high-precision +// decimals; fine for an emulator, swap for a sortable-number encoding if it bites. +func orderByColumn(col, keyType string) string { + if keyType == "N" { + return "CAST(" + col + " AS REAL)" + } + return col +} + +// prefixUpperBound returns the least string strictly greater than every string +// beginning with prefix, for a half-open range scan (sk >= prefix AND sk < +// bound). It returns "" when prefix is empty or all 0xFF bytes, meaning no upper +// bound is needed. +func prefixUpperBound(prefix string) string { + b := []byte(prefix) + for i := len(b) - 1; i >= 0; i-- { + if b[i] < 0xFF { + b[i]++ + return string(b[:i+1]) } } return "" @@ -549,11 +666,6 @@ func (s *DynamoStore) queryItems(query string, args []any) ([]Item, error) { return items, rows.Err() } -// escapeLike escapes LIKE metacharacters so a prefix matches literally. -func escapeLike(s string) string { - return strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(s) -} - // PutTTLConfig stores TTL config for a table. func (s *DynamoStore) PutTTLConfig(tableName string, cfg TTLConfig) error { data, err := json.Marshal(cfg) @@ -601,6 +713,8 @@ func (s *DynamoStore) GetTags(resourceArn string) (map[string]string, error) { // PutTags stores tags for a resource ARN (merges with existing). func (s *DynamoStore) PutTags(resourceArn string, newTags map[string]string) error { + s.tagsMu.Lock() + defer s.tagsMu.Unlock() existing, err := s.GetTags(resourceArn) if err != nil { return err @@ -613,6 +727,8 @@ func (s *DynamoStore) PutTags(resourceArn string, newTags map[string]string) err // RemoveTags removes specified tag keys for a resource ARN. func (s *DynamoStore) RemoveTags(resourceArn string, tagKeys []string) error { + s.tagsMu.Lock() + defer s.tagsMu.Unlock() existing, err := s.GetTags(resourceArn) if err != nil { return err diff --git a/internal/services/dynamodb/store_test.go b/internal/services/dynamodb/store_test.go index 7d02e57e..a7360aa1 100644 --- a/internal/services/dynamodb/store_test.go +++ b/internal/services/dynamodb/store_test.go @@ -3,6 +3,7 @@ package dynamodb import ( + "fmt" "testing" "github.com/stretchr/testify/assert" @@ -321,8 +322,8 @@ func TestDynamoStore_NumericKey(t *testing.T) { assert.Equal(t, strPtr("Primary"), (*got)["Name"].S) } -// TestDynamoStore_QuerySortKeyPrefixEscaping verifies escapeLike treats LIKE -// metacharacters (%, _, \) in a sort-key prefix as literals. +// TestDynamoStore_QuerySortKeyPrefixEscaping verifies a sort-key prefix matches +// LIKE metacharacters (%, _, \) literally (the range scan has no wildcards). func TestDynamoStore_QuerySortKeyPrefixEscaping(t *testing.T) { store := newTestStore(t) require.NoError(t, store.CreateTable(TableInfo{ @@ -352,3 +353,140 @@ func TestDynamoStore_QuerySortKeyPrefixEscaping(t *testing.T) { }) } } + +// TestDynamoStore_QueryNumericSortOrder verifies a numeric ('N') sort key sorts +// numerically, not lexicographically (2, 10, 100 — not 10, 100, 2). +func TestDynamoStore_QueryNumericSortOrder(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Readings", + PartitionKey: KeyDef{Name: "Sensor", Type: "S"}, + SortKey: &KeyDef{Name: "Seq", Type: "N"}, + })) + for _, seq := range []string{"100", "2", "10", "9"} { + require.NoError(t, store.PutItem("Readings", Item{"Sensor": {S: strPtr("s1")}, "Seq": {N: strPtr(seq)}})) + } + + got, err := store.Query("Readings", "s1", "") + require.NoError(t, err) + var order []string + for _, it := range got { + order = append(order, *it["Seq"].N) + } + assert.Equal(t, []string{"2", "9", "10", "100"}, order) +} + +// TestDynamoStore_QueryPrefixCaseSensitive verifies a begins_with prefix matches +// case-sensitively (the old LIKE path over-matched across letter case). +func TestDynamoStore_QueryPrefixCaseSensitive(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Files", + PartitionKey: KeyDef{Name: "UserID", Type: "S"}, + SortKey: &KeyDef{Name: "Path", Type: "S"}, + })) + for _, p := range []string{"ABC", "aBc", "abc"} { + require.NoError(t, store.PutItem("Files", Item{"UserID": {S: strPtr("u1")}, "Path": {S: strPtr(p)}})) + } + got, err := store.Query("Files", "u1", "AB") + require.NoError(t, err) + assert.Len(t, got, 1) // only "ABC", not "aBc"/"abc" +} + +// TestDynamoStore_QueryGSISortOrder verifies QueryGSI returns items ordered by +// the index RANGE key, numerically for an 'N' key. +func TestDynamoStore_QueryGSISortOrder(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Messages", + PartitionKey: KeyDef{Name: "MsgID", Type: "S"}, + GlobalSecondaryIndexes: []IndexDef{{ + IndexName: "ByConversation", + KeySchema: []KeyDef{ + {Name: "ConvID", Type: "S", KeyType: "HASH"}, + {Name: "CreatedAt", Type: "N", KeyType: "RANGE"}, + }, + }}, + })) + for i, ts := range []string{"300", "100", "200"} { + require.NoError(t, store.PutItem("Messages", Item{ + "MsgID": {S: strPtr(fmt.Sprintf("m%d", i))}, "ConvID": {S: strPtr("c1")}, "CreatedAt": {N: strPtr(ts)}, + })) + } + got, err := store.QueryGSI("Messages", "ByConversation", "c1") + require.NoError(t, err) + var order []string + for _, it := range got { + order = append(order, *it["CreatedAt"].N) + } + assert.Equal(t, []string{"100", "200", "300"}, order) +} + +// TestDynamoStore_GSISparseComposite verifies an item missing a composite index's +// RANGE attribute is not projected (no phantom item in the index). +func TestDynamoStore_GSISparseComposite(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Messages", + PartitionKey: KeyDef{Name: "MsgID", Type: "S"}, + GlobalSecondaryIndexes: []IndexDef{{ + IndexName: "ByConversation", + KeySchema: []KeyDef{ + {Name: "ConvID", Type: "S", KeyType: "HASH"}, + {Name: "CreatedAt", Type: "N", KeyType: "RANGE"}, + }, + }}, + })) + // Has HASH but no RANGE attribute → excluded from the index. + require.NoError(t, store.PutItem("Messages", Item{"MsgID": {S: strPtr("m1")}, "ConvID": {S: strPtr("c1")}})) + // Full key → included. + require.NoError(t, store.PutItem("Messages", Item{"MsgID": {S: strPtr("m2")}, "ConvID": {S: strPtr("c1")}, "CreatedAt": {N: strPtr("10")}})) + + got, err := store.QueryGSI("Messages", "ByConversation", "c1") + require.NoError(t, err) + assert.Len(t, got, 1) +} + +// TestDynamoStore_GSIBadKeyType verifies a present-but-non-scalar index key value +// fails the put rather than silently dropping the item from the index. +func TestDynamoStore_GSIBadKeyType(t *testing.T) { + store := newTestStore(t) + require.NoError(t, store.CreateTable(TableInfo{ + Name: "Users", + PartitionKey: KeyDef{Name: "UserID", Type: "S"}, + GlobalSecondaryIndexes: []IndexDef{{ + IndexName: "ByFlag", + KeySchema: []KeyDef{{Name: "Flag", Type: "S", KeyType: "HASH"}}, + }}, + })) + tru := true + err := store.PutItem("Users", Item{"UserID": {S: strPtr("u1")}, "Flag": {BOOL: &tru}}) + require.Error(t, err) +} + +// TestDynamoStore_DeleteTableClearsTTLAndTags verifies DeleteTable removes the +// table's TTL config and tags, so a recreated same-name table doesn't inherit them. +func TestDynamoStore_DeleteTableClearsTTLAndTags(t *testing.T) { + store := newTestStore(t) + info := TableInfo{Name: "T", PartitionKey: KeyDef{Name: "ID", Type: "S"}} + require.NoError(t, store.CreateTable(info)) + require.NoError(t, store.PutTTLConfig("T", TTLConfig{Enabled: true, AttributeName: "exp"})) + + arn, err := store.GetTable("T") + require.NoError(t, err) + require.NoError(t, store.PutTags(arn.TableArn, map[string]string{"env": "test"})) + + require.NoError(t, store.DeleteTable("T")) + require.NoError(t, store.CreateTable(info)) + + ttl, err := store.GetTTLConfig("T") + require.NoError(t, err) + assert.False(t, ttl.Enabled) + assert.Empty(t, ttl.AttributeName) + + recreated, err := store.GetTable("T") + require.NoError(t, err) + tags, err := store.GetTags(recreated.TableArn) + require.NoError(t, err) + assert.Empty(t, tags) +}