GiteaAdmin/elowdb-go
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

before refactor index store to complex file-line pattern

Browse Source
This commit is contained in:
Кузнецов Антон (aka Direct-dev.ru)
2026-03-12 16:13:44 +06:00
parent 491ccbea89
commit 8ba956d8c5
21 changed files with 7804 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
.vscode/launch.json vendored
View File

@@ -60,6 +60,18 @@
"args": [],
"showLog": true,
"console": "integratedTerminal"
},
{
"name": "Debug LineDB Perf Example",
"type": "go",
"request": "launch",
"mode": "auto",
"program": "${workspaceFolder}/examples/perf/main.go",
"cwd": "${workspaceFolder}",
"env": {},
"args": [],
"showLog": true,
"console": "integratedTerminal"
}
]
}

172
INDEXES.md Normal file
View File

@@ -0,0 +1,172 @@
# Индексы в LineDB
## Обзор
Индексы ускоряют поиск записей по полям при вызове `ReadByFilter`. Без индексов каждый запрос сканирует весь JSONL‑файл. С индексами для простых фильтров вида `{ field: value }` используется предварительно построенный индекс, и возвращаются только подходящие записи.
## Как это работает
1. **Конфигурация**: в `JSONLFileOptions` задаётся `IndexedFields` — список полей, по которым строится индекс.
2. **Хранилище**: данные индекса лежат в `IndexStore` — реализация интерфейса, которую можно подключать.
3. **При Init**: если есть коллекции с `IndexedFields`, индекс строится из текущих данных.
4. **При Insert**: новые записи добавляются в индекс.
5. **При Update/Delete**: индекс коллекции полностью перестраивается.
6. **При ReadByFilter**: если фильтр — одно поле из `IndexedFields` и точное совпадение, используется индекс; иначе — полный скан файла.
## Интерфейс IndexStore
```go
type IndexStore interface {
// Возвращает индексы строк (0-based) для заданного значения поля.
Lookup(collection, field, value string) ([]int, error)
// Полная пересборка индекса по всем записям коллекции.
Rebuild(collection string, fields []string, records []any) error
// Очистить индекс коллекции.
Clear(collection string) error
}
```
- `Lookup` — поиск по `(collection, field, value)`, возвращает **номера строк в файле**.
- `Rebuild` — полная пересборка индекса коллекции (использует порядок записей как номера строк).
- `Clear` — очистить индекс коллекции.
## Реализации
### 1. InMemoryIndexStore (по умолчанию)
Используется автоматически при `IndexedFields` и `IndexStore == nil`:
```go
initOptions := &linedb.LineDbInitOptions{
DBFolder: "./data",
Collections: []linedb.JSONLFileOptions{
{
CollectionName: "users",
AllocSize: 256,
IndexedFields: []string{"id", "email", "name"},
},
},
}
db := linedb.NewLineDb(nil)
db.Init(false, initOptions)
// Индекс создаётся в памяти автоматически
```
Явная установка:
```go
store := linedb.NewInMemoryIndexStore()
db := linedb.NewLineDb(&linedb.LineDbOptions{IndexStore: store})
db.Init(false, initOptions)
```
### 2. MemcachedIndexStore
Хранит индексы в Memcached. Требуется реализовать интерфейс `MemcachedClient`:
```go
type MemcachedClient interface {
Get(key string) ([]byte, error)
Set(key string, value []byte, expireSeconds int) error
Delete(key string) error
}
```
Пример с [gomemcache](https://github.com/bradfitz/gomemcache):
```go
import "github.com/bradfitz/gomemcache/memcache"
// Адаптер под MemcachedClient
type memcacheAdapter struct {
c *memcache.Client
}
func (m *memcacheAdapter) Get(key string) ([]byte, error) {
it, err := m.c.Get(key)
if err == memcache.ErrCacheMiss {
return nil, nil
}
if err != nil {
return nil, err
}
return it.Value, nil
}
func (m *memcacheAdapter) Set(key string, value []byte, exp int) error {
return m.c.Set(&memcache.Item{Key: key, Value: value, Expiration: int32(exp)})
}
func (m *memcacheAdapter) Delete(key string) error {
return m.c.Delete(key)
}
// Использование
client := memcache.New("127.0.0.1:11211")
store, err := linedb.NewMemcachedIndexStore(linedb.MemcachedIndexStoreOptions{
Client: &memcacheAdapter{c: client},
KeyPrefix: "linedb:idx:",
ExpireSeconds: 3600,
})
if err != nil {
log.Fatal(err)
}
db := linedb.NewLineDb(&linedb.LineDbOptions{IndexStore: store})
db.Init(false, initOptions)
```
## Подключение реализации
### Вариант A: In-memory по умолчанию
Ничего не указывать — при наличии `IndexedFields` создаётся in-memory индекс:
```go
db := linedb.NewLineDb(nil)
initOptions.Collections[0].IndexedFields = []string{"id", "email"}
db.Init(false, initOptions)
```
### Вариант B: Явно in-memory
```go
store := linedb.NewInMemoryIndexStore()
db := linedb.NewLineDb(&linedb.LineDbOptions{IndexStore: store})
db.Init(false, initOptions)
```
### Вариант C: Memcached
```go
store, _ := linedb.NewMemcachedIndexStore(linedb.MemcachedIndexStoreOptions{
Client: myMemcachedClient,
})
db := linedb.NewLineDb(&linedb.LineDbOptions{IndexStore: store})
db.Init(false, initOptions)
```
### Вариант D: Собственная реализация
Реализуйте интерфейс `IndexStore` и передайте её в `LineDbOptions.IndexStore`.
## Ограничения
- Индекс используется только при **одном** поле в фильтре и **точном совпадении** значения.
- Для партиционированных коллекций индекс не используется.
- `MemcachedIndexStore.Clear` не удаляет ключи (Memcached не поддерживает поиск по префиксу); старые ключи исчезнут по TTL.
- При Update и Delete индекс перестраивается полностью (чтение всех записей и Rebuild).
## Примеры
```go
// Поиск по email (если email в IndexedFields)
users, err := db.ReadByFilter(map[string]any{"email": "alice@test.com"}, "users", opts)
// Поиск по id
user, err := db.ReadByFilter(map[string]any{"id": 1}, "users", opts)
// Фильтр по нескольким полям — индекс не используется, идёт полный скан
users, err := db.ReadByFilter(map[string]any{"name": "alice", "active": true}, "users", opts)
```

2500
data/perf-benchmark/no-index/perf_items.jsonl Normal file
View File

File diff suppressed because it is too large Load Diff

3000
data/perf-benchmark/with-index/perf_items.jsonl Normal file
View File

File diff suppressed because it is too large Load Diff

5
examples/basic/data/users.jsonl
View File

@@ -1,2 +1,3 @@
{"age":31,"created":"2026-03-04T13:58:25+06:00","email":"john@example.com","id":1,"name":"John Doe"}
{"age":35,"created":"2026-03-04T13:58:25+06:00","email":"bob@example.com","id":3,"name":"Bob Johnson"}
{"age":31,"created":"2026-03-06T16:33:26+06:00","email":"john@example.com","id":1,"name":"John Doe"}
{"age":35,"created":"2026-03-06T16:33:26+06:00","email":"bob@example.com","id":3,"name":"Bob Johnson"}

0
examples/partitions/data/events.jsonl Normal file
View File

4
data/secret_users.jsonl.bak → examples/partitions/data/events_A.jsonl
View File

@@ -1,2 +1,2 @@
4jywagSLv0grz1G1ZoRu484Abu5GXaNYA9LsMmRQIxmTMuFczm4jRGYZm4JwQ+8YuWhjsOs0CRyOtB+dg0skg176sz9ES85DPGGiemJB9bZBKORMB+O4UL9dH5j9
poBf6u/pAsjCFu0twwoSHspgUXXNGliXHoJ8eqtL82demTjpTo9+SXzbUSaAUsUgdy1XQJZlncOIWeTlxgfOJXuqpXPpqNJtxyZd1E/9+jOgRreQXOmyMg==
{"id":1,"status":"processed","tenant":"A","ts":1773305152,"type":"signup"}
{"id":2,"status":"processed","tenant":"A","ts":1773305152,"type":"purchase"}

1
examples/partitions/data/events_B.jsonl Normal file
View File

@@ -0,0 +1 @@
{"id":3,"status":"new","tenant":"B","ts":1773305152,"type":"signup"}

2
examples/partitions/data/users.jsonl Normal file
View File

@@ -0,0 +1,2 @@
{"createdAt":"2026-03-12T14:45:52.66645719+06:00","email":"a@example.com","id":1,"name":"Alice"}
{"createdAt":"2026-03-12T14:45:52.666505533+06:00","email":"b@example.com","id":2,"name":"Bob"}

161
examples/partitions/main.go Normal file
View File

@@ -0,0 +1,161 @@
package main
import (
"fmt"
"log"
"os"
"path/filepath"
"time"
"linedb/pkg/linedb"
)
// Пример работы с партициями + индексируемой коллекцией.
//
// Запуск:
// go run ./examples/partitions/main.go
//
// Важно:
// - В текущей реализации индексы строятся для "обычных" коллекций.
// - Партиционированные коллекции (партиции) создаются динамически и сейчас не индексируются
// (см. getPartitionAdapter: JSONLFileOptions{CollectionName: partitionName} без IndexedFields).
func main() {
dbDir := filepath.Join(".", "examples", "partitions", "data")
_ = os.RemoveAll(dbDir)
if err := os.MkdirAll(dbDir, 0755); err != nil {
log.Fatalf("mkdir: %v", err)
}
// Настройка:
// - users: обычная коллекция с индексами
// - events: базовая коллекция для партиций (сама по себе не используется для записи),
// а реальные данные будут в events_<tenant>.jsonl
initOptions := &linedb.LineDbInitOptions{
DBFolder: dbDir,
Collections: []linedb.JSONLFileOptions{
{
CollectionName: "users",
AllocSize: 512,
IndexedFields: []string{"id", "email"},
},
{
CollectionName: "events",
AllocSize: 512,
},
},
Partitions: []linedb.PartitionCollection{
{
CollectionName: "events",
PartIDFn: func(v any) string {
m, ok := v.(map[string]any)
if !ok {
return "unknown"
}
tenant, ok := m["tenant"].(string)
if !ok || tenant == "" {
return "unknown"
}
return tenant
},
},
},
}
db := linedb.NewLineDb(&linedb.LineDbOptions{
IndexStore: linedb.NewInMemoryIndexStore(),
})
if err := db.Init(true, initOptions); err != nil {
log.Fatalf("Init failed: %v", err)
}
defer db.Close()
fmt.Println("=== 1) Индексируемая коллекция users ===")
users := []any{
map[string]any{"id": 1, "email": "a@example.com", "name": "Alice", "createdAt": time.Now().Format(time.RFC3339Nano)},
map[string]any{"id": 2, "email": "b@example.com", "name": "Bob", "createdAt": time.Now().Format(time.RFC3339Nano)},
}
// Используем Write + DoIndexing: true, чтобы индекс был актуален сразу после записи.
if err := db.Write(users, "users", linedb.LineDbAdapterOptions{DoIndexing: true}); err != nil {
log.Fatalf("Write users failed: %v", err)
}
byEmail, err := db.ReadByFilter(map[string]any{"email": "a@example.com"}, "users", linedb.LineDbAdapterOptions{})
if err != nil {
log.Fatalf("ReadByFilter users by email failed: %v", err)
}
mustLen("users by email", byEmail, 1)
byID, err := db.ReadByFilter(map[string]any{"id": 2}, "users", linedb.LineDbAdapterOptions{})
if err != nil {
log.Fatalf("ReadByFilter users by id failed: %v", err)
}
mustLen("users by id", byID, 1)
fmt.Println("OK: users inserted and searchable by indexed fields (id/email).")
fmt.Println("\n=== 2) Партиционированная коллекция events (events_<tenant>) ===")
events := []any{
map[string]any{"id": 1, "tenant": "A", "type": "signup", "status": "new", "ts": time.Now().Unix()},
map[string]any{"id": 2, "tenant": "A", "type": "purchase", "status": "new", "ts": time.Now().Unix()},
map[string]any{"id": 3, "tenant": "B", "type": "signup", "status": "new", "ts": time.Now().Unix()},
}
if err := db.Insert(events, "events", linedb.LineDbAdapterOptions{}); err != nil {
log.Fatalf("Insert events failed: %v", err)
}
tenantA, err := db.ReadByFilter(map[string]any{"tenant": "A"}, "events", linedb.LineDbAdapterOptions{})
if err != nil {
log.Fatalf("ReadByFilter events tenant A failed: %v", err)
}
mustLen("events tenant A after insert", tenantA, 2)
tenantB, err := db.ReadByFilter(map[string]any{"tenant": "B"}, "events", linedb.LineDbAdapterOptions{})
if err != nil {
log.Fatalf("ReadByFilter events tenant B failed: %v", err)
}
mustLen("events tenant B after insert", tenantB, 1)
fmt.Println("OK: события разложены по партициям (A и B).")
fmt.Println("\n=== 3) Update по всем партициям ===")
updated, err := db.Update(
map[string]any{"status": "processed"},
"events",
map[string]any{"tenant": "A"},
linedb.LineDbAdapterOptions{},
)
if err != nil {
log.Fatalf("Update events failed: %v", err)
}
mustLen("updated events for tenant A", updated, 2)
processedA, err := db.ReadByFilter(map[string]any{"tenant": "A", "status": "processed"}, "events", linedb.LineDbAdapterOptions{})
if err != nil {
log.Fatalf("ReadByFilter processed events failed: %v", err)
}
mustLen("processed events for tenant A", processedA, 2)
fmt.Println("OK: обновление затронуло записи в партиции A.")
fmt.Println("\n=== 4) Delete по всем партициям ===")
deleted, err := db.Delete(map[string]any{"id": 3}, "events", linedb.LineDbAdapterOptions{})
if err != nil {
log.Fatalf("Delete events failed: %v", err)
}
mustLen("deleted events id=3", deleted, 1)
allRemaining, err := db.ReadByFilter(nil, "events", linedb.LineDbAdapterOptions{})
if err != nil {
log.Fatalf("ReadByFilter all remaining events failed: %v", err)
}
mustLen("remaining events after delete", allRemaining, 2)
fmt.Printf("\nГотово. Данные примера в: %s\n", dbDir)
}
func mustLen(label string, got []any, want int) {
if len(got) != want {
log.Fatalf("%s: expected %d, got %d (%v)", label, want, len(got), got)
}
}

178
examples/perf/main.go
View File

@@ -5,7 +5,7 @@
//
// Сценарий:
// 1. Вставка 5000 записей, каждая ~1800 символов.
// 2. 10 случайных обновлений, замер среднего времени.
// 2. N выборок ReadByFilter по индексируемому полю, замер среднего времени.
// 3. 100 случайных удалений, замер среднего времени.
// 4. Замеры памяти процесса через runtime.MemStats.
package main
@@ -15,6 +15,7 @@ import (
"log"
"math/rand"
"os"
"path/filepath"
"runtime"
"strings"
"time"
@@ -23,79 +24,152 @@ import (
)
const (
recordsCount = 5000
payloadSize = 1800
updateOps = 100
deleteOps = 100
recordsCount = 3000
payloadSize = 1024
readOps = 2000
updateOps = 500
deleteOps = 500
collectionName = "perf_items"
dbDir = "./data/perf-benchmark"
baseDBDir = "./data/perf-benchmark"
allocSizeEstimate = 2048
)
func main() {
rand.Seed(time.Now().UnixNano())
if err := os.MkdirAll(dbDir, 0755); err != nil {
if err := os.MkdirAll(baseDBDir, 0755); err != nil {
log.Fatalf("mkdir: %v", err)
}
fmt.Printf("LineDB perf сравнение: без индекса vs с индексом\n")
fmt.Printf("records=%d payload~%d readOps=%d updateOps=%d deleteOps=%d allocSize=%d\n\n",
recordsCount, payloadSize, readOps, updateOps, deleteOps, allocSizeEstimate)
noIdx := runScenario(false)
runtime.GC()
withIdx := runScenario(true)
fmt.Printf("\n=== Сводка (рядом) ===\n")
fmt.Printf("%-26s | %-18s | %-18s\n", "Метрика", "Без индекса", "С индексом")
fmt.Printf("%-26s | %-18v | %-18v\n", "Insert total", noIdx.insertTotal, withIdx.insertTotal)
fmt.Printf("%-26s | %-18v | %-18v\n", "Index build", noIdx.indexBuildTotal, withIdx.indexBuildTotal)
fmt.Printf("%-26s | %-18v | %-18v\n", "ReadByFilter avg", noIdx.readAvg, withIdx.readAvg)
fmt.Printf("%-26s | %-18v | %-18v\n", "Update avg", noIdx.updateAvg, withIdx.updateAvg)
fmt.Printf("%-26s | %-18v | %-18v\n", "Delete avg", noIdx.deleteAvg, withIdx.deleteAvg)
fmt.Printf("%-26s | %-18d | %-18d\n", "Final records", noIdx.finalCount, withIdx.finalCount)
fmt.Printf("%-26s | %-18.2f | %-18.2f\n", "Mem Alloc (MB)", noIdx.memAllocMB, withIdx.memAllocMB)
fmt.Printf("%-26s | %-18.2f | %-18.2f\n", "Mem Sys (MB)", noIdx.memSysMB, withIdx.memSysMB)
fmt.Printf("%-26s | %-18d | %-18d\n", "NumGC", noIdx.numGC, withIdx.numGC)
fmt.Printf("\nData directories:\n no-index: %s\n with-index:%s\n", noIdx.dbDir, withIdx.dbDir)
}
type scenarioResult struct {
dbDir string
insertTotal time.Duration
indexBuildTotal time.Duration
readAvg time.Duration
updateAvg time.Duration
deleteAvg time.Duration
finalCount int
memAllocMB float64
memSysMB float64
numGC uint32
}
func runScenario(useIndex bool) scenarioResult {
label := "no-index"
if useIndex {
label = "with-index"
}
dbDir := filepath.Join(baseDBDir, label)
_ = os.RemoveAll(dbDir)
var store linedb.IndexStore
lineOpts := linedb.LineDbOptions{}
collOpts := linedb.JSONLFileOptions{
CollectionName: collectionName,
AllocSize: allocSizeEstimate,
}
if useIndex {
// Индексируем поля id и index (по ним идут фильтры в тесте)
collOpts.IndexedFields = []string{"id", "index"}
store = linedb.NewInMemoryIndexStore()
lineOpts.IndexStore = store
}
initOptions := &linedb.LineDbInitOptions{
CacheSize: 1000,
CacheTTL: time.Minute,
DBFolder: dbDir,
Collections: []linedb.JSONLFileOptions{
{
CollectionName: collectionName,
AllocSize: allocSizeEstimate,
},
collOpts,
},
}
db := linedb.NewLineDb(nil)
db := linedb.NewLineDb(&lineOpts)
if err := db.Init(true, initOptions); err != nil {
log.Fatalf("Init failed: %v", err)
log.Fatalf("[%s] Init failed: %v", label, err)
}
defer db.Close()
fmt.Printf("=== %s ===\n", label)
printMem("Before insert")
fmt.Printf("1) Insert %d records (payload ~%d chars)...\n", recordsCount, payloadSize)
fmt.Printf("1) Insert %d records...\n", recordsCount)
start := time.Now()
if err := insertRecords(db); err != nil {
log.Fatalf("InsertRecords failed: %v", err)
log.Fatalf("[%s] insertRecords failed: %v", label, err)
}
elapsedInsert := time.Since(start)
fmt.Printf(" Total insert time: %v, per record: %v\n",
elapsedInsert, elapsedInsert/time.Duration(recordsCount))
insertDur := time.Since(start)
fmt.Printf(" Total insert time: %v, per record: %v\n", insertDur, insertDur/time.Duration(recordsCount))
// Индекс строится внутри Write при DoIndexing: true (точечная индексация)
printMem("After insert")
all, err := db.Read(collectionName, linedb.LineDbAdapterOptions{})
if err != nil {
log.Fatalf("Read after insert failed: %v", err)
log.Fatalf("[%s] Read after insert failed: %v", label, err)
}
fmt.Printf(" Records in collection: %d\n", len(all))
ids := collectIDs(all)
if len(ids) == 0 {
log.Fatalf("No IDs collected, cannot continue")
indexVals := collectFieldValues(all, "index")
if len(ids) == 0 || len(indexVals) == 0 {
log.Fatalf("[%s] No IDs or index values collected, cannot continue", label)
}
fmt.Printf("\n2) Random update of %d records...\n", updateOps)
fmt.Printf("\n2) Random ReadByFilter of %d ops (field=index)...\n", readOps)
readAvg := benchmarkReads(db, indexVals, readOps)
fmt.Printf(" Average ReadByFilter time: %v\n", readAvg)
fmt.Printf("\n3) Random update of %d records...\n", updateOps)
avgUpdate := benchmarkUpdates(db, ids, updateOps)
fmt.Printf(" Average update time: %v\n", avgUpdate)
printMem("After updates")
fmt.Printf("\n3) Random delete of %d records...\n", deleteOps)
fmt.Printf("\n4) Random delete of %d records...\n", deleteOps)
avgDelete := benchmarkDeletes(db, ids, deleteOps)
fmt.Printf(" Average delete time: %v\n", avgDelete)
printMem("After deletes")
final, err := db.Read(collectionName, linedb.LineDbAdapterOptions{})
if err != nil {
log.Fatalf("Final read failed: %v", err)
log.Fatalf("[%s] Final read failed: %v", label, err)
}
fmt.Printf("\nFinal records in collection: %d\n", len(final))
fmt.Printf("Data directory: %s\n", dbDir)
mem := memSnapshot()
return scenarioResult{
dbDir: dbDir,
insertTotal: insertDur,
indexBuildTotal: 0, // индекс строится точечно при Write с DoIndexing
readAvg: readAvg,
updateAvg: avgUpdate,
deleteAvg: avgDelete,
finalCount: len(final),
memAllocMB: mem.allocMB,
memSysMB: mem.sysMB,
numGC: mem.numGC,
}
}
func insertRecords(db *linedb.LineDb) error {
@@ -105,20 +179,21 @@ func insertRecords(db *linedb.LineDb) error {
batch := make([]any, 0, 100)
for i := 0; i < recordsCount; i++ {
rec := map[string]any{
"id": i + 1,
"index": i,
"payload": base,
"created": time.Now().Format(time.RFC3339Nano),
}
batch = append(batch, rec)
if len(batch) >= cap(batch) {
if err := db.Insert(batch, collectionName, linedb.LineDbAdapterOptions{}); err != nil {
if err := db.Write(batch, collectionName, linedb.LineDbAdapterOptions{DoIndexing: true}); err != nil {
return err
}
batch = batch[:0]
}
}
if len(batch) > 0 {
if err := db.Insert(batch, collectionName, linedb.LineDbAdapterOptions{}); err != nil {
if err := db.Write(batch, collectionName, linedb.LineDbAdapterOptions{DoIndexing: true}); err != nil {
return err
}
}
@@ -137,6 +212,37 @@ func collectIDs(all []any) []any {
return ids
}
func collectFieldValues(all []any, field string) []any {
vals := make([]any, 0, len(all))
for _, r := range all {
if m, ok := r.(map[string]any); ok {
if v, ok := m[field]; ok {
vals = append(vals, v)
}
}
}
return vals
}
func benchmarkReads(db *linedb.LineDb, values []any, ops int) time.Duration {
if len(values) == 0 || ops == 0 {
return 0
}
var total time.Duration
for i := 0; i < ops; i++ {
v := values[rand.Intn(len(values))]
start := time.Now()
_, err := db.ReadByFilter(map[string]any{"index": v}, collectionName, linedb.LineDbAdapterOptions{})
dur := time.Since(start)
if err != nil {
log.Printf("ReadByFilter error (index=%v): %v", v, err)
continue
}
total += dur
}
return total / time.Duration(ops)
}
func benchmarkUpdates(db *linedb.LineDb, ids []any, ops int) time.Duration {
if len(ids) == 0 {
return 0
@@ -204,3 +310,19 @@ func printMem(label string) {
fmt.Printf(" Sys: %.2f MB\n", float64(m.Sys)/1024.0/1024.0)
fmt.Printf(" NumGC: %d\n", m.NumGC)
}
type memSnap struct {
allocMB float64
sysMB float64
numGC uint32
}
func memSnapshot() memSnap {
var m runtime.MemStats
runtime.ReadMemStats(&m)
return memSnap{
allocMB: float64(m.Alloc) / 1024.0 / 1024.0,
sysMB: float64(m.Sys) / 1024.0 / 1024.0,
numGC: m.NumGC,
}
}

60
pkg/linedb/cache.go
View File

@@ -87,13 +87,71 @@ func (c *RecordCache) Delete(key string) {
delete(c.cache, key)
}
// Clear очищает кэш
// Clear очищает кэш полностью.
func (c *RecordCache) Clear() {
c.mutex.Lock()
defer c.mutex.Unlock()
c.cache = make(map[string]*CacheEntry)
}
// ClearEntriesContainingIDs удаляет из кэша только те записи, в данных которых
// встречается хотя бы один из переданных id. Если ids пуст — ничего не делает.
func (c *RecordCache) ClearEntriesContainingIDs(ids []any) {
if len(ids) == 0 {
return
}
c.mutex.Lock()
defer c.mutex.Unlock()
for key, entry := range c.cache {
if entry.Data == nil {
continue
}
records, ok := entry.Data.([]any)
if !ok {
continue
}
for _, rec := range records {
m, ok := rec.(map[string]any)
if !ok {
continue
}
recID := m["id"]
for _, id := range ids {
if idsEqual(recID, id) {
delete(c.cache, key)
goto nextKey
}
}
}
nextKey:
}
}
func idsEqual(a, b any) bool {
if a == b {
return true
}
// JSON unmarshals numbers as float64; сравниваем численно
if an, ok := toFloat64(a); ok {
if bn, ok := toFloat64(b); ok {
return an == bn
}
}
return false
}
func toFloat64(v any) (float64, bool) {
switch x := v.(type) {
case float64:
return x, true
case int:
return float64(x), true
case int64:
return float64(x), true
}
return 0, false
}
// Stop останавливает кэш
func (c *RecordCache) Stop() {
close(c.stopChan)

212
pkg/linedb/index.go Normal file
View File

@@ -0,0 +1,212 @@
package linedb
import (
"fmt"
"strconv"
"sync"
)
// IndexStore — интерфейс хранилища индексов.
// Позволяет подключать разные реализации: в памяти, memcached и др.
type IndexStore interface {
// Lookup ищет позиции записей по полю и значению.
// value — строковое представление (см. valueToIndexKey).
// Возвращает срез индексов строк (0-based) или nil, nil,
// если индекс не используется или записей нет.
Lookup(collection, field, value string) ([]int, error)
// IndexRecord добавляет одну запись в индекс по номеру строки (для точечных Update).
IndexRecord(collection string, fields []string, record map[string]any, lineIndex int)
// UnindexRecord удаляет одну запись из индекса по номеру строки.
UnindexRecord(collection string, fields []string, record map[string]any, lineIndex int)
// Rebuild полностью перестраивает индекс коллекции.
// records — полный список записей, каждая позиция в срезе соответствует номеру строки в файле.
Rebuild(collection string, fields []string, records []any) error
// Clear очищает индекс коллекции.
Clear(collection string) error
}
// valueToIndexKey преобразует значение фильтра в ключ для индекса.
func valueToIndexKey(v any) string {
if v == nil {
return ""
}
switch t := v.(type) {
case string:
return t
case int:
return strconv.Itoa(t)
case int64:
return strconv.FormatInt(t, 10)
case float64:
return strconv.FormatFloat(t, 'g', -1, 64)
case bool:
if t {
return "true"
}
return "false"
default:
return fmt.Sprintf("%v", v)
}
}
// getFieldValue извлекает значение поля из записи и превращает в ключ индекса.
func getFieldValue(record map[string]any, field string) string {
v, ok := record[field]
if !ok {
return ""
}
return valueToIndexKey(v)
}
// InMemoryIndexStore — реализация IndexStore в памяти (по умолчанию).
type InMemoryIndexStore struct {
mu sync.RWMutex
// index: collection:field -> value -> список индексов строк (0-based)
index map[string]map[string][]int
}
// NewInMemoryIndexStore создаёт новый in-memory индекс.
func NewInMemoryIndexStore() *InMemoryIndexStore {
return &InMemoryIndexStore{
index: make(map[string]map[string][]int),
}
}
// indexKey формирует ключ карты для collection:field.
func (s *InMemoryIndexStore) indexKey(collection, field string) string {
return collection + ":" + field
}
// IndexRecord добавляет запись в индекс.
func (s *InMemoryIndexStore) IndexRecord(collection string, fields []string, record map[string]any, lineIndex int) {
if len(fields) == 0 || record == nil {
return
}
s.mu.Lock()
defer s.mu.Unlock()
for _, field := range fields {
val := getFieldValue(record, field)
key := s.indexKey(collection, field)
if s.index[key] == nil {
s.index[key] = make(map[string][]int)
}
s.index[key][val] = append(s.index[key][val], lineIndex)
}
}
// UnindexRecord удаляет запись из индекса.
func (s *InMemoryIndexStore) UnindexRecord(collection string, fields []string, record map[string]any, lineIndex int) {
if len(fields) == 0 || record == nil {
return
}
s.mu.Lock()
defer s.mu.Unlock()
for _, field := range fields {
val := getFieldValue(record, field)
key := s.indexKey(collection, field)
bucket := s.index[key]
if bucket == nil {
continue
}
idxs := bucket[val]
newIdxs := make([]int, 0, len(idxs))
for _, i := range idxs {
if i != lineIndex {
newIdxs = append(newIdxs, i)
}
}
if len(newIdxs) == 0 {
delete(bucket, val)
} else {
bucket[val] = newIdxs
}
}
}
// Lookup возвращает индексы строк по полю и значению.
func (s *InMemoryIndexStore) Lookup(collection, field, value string) ([]int, error) {
s.mu.RLock()
defer s.mu.RUnlock()
key := s.indexKey(collection, field)
bucket := s.index[key]
if bucket == nil {
return nil, nil
}
indexes := bucket[value]
if len(indexes) == 0 {
return nil, nil
}
// Возвращаем копию, чтобы вызывающий код не модифицировал внутренний срез
out := make([]int, len(indexes))
copy(out, indexes)
return out, nil
}
// Rebuild полностью перестраивает индекс коллекции.
func (s *InMemoryIndexStore) Rebuild(collection string, fields []string, records []any) error {
s.mu.Lock()
defer s.mu.Unlock()
// Очищаем старые записи по этой коллекции
for _, field := range fields {
key := s.indexKey(collection, field)
delete(s.index, key)
}
for idx, rec := range records {
recMap, ok := rec.(map[string]any)
if !ok {
continue
}
for _, field := range fields {
val := getFieldValue(recMap, field)
key := s.indexKey(collection, field)
if s.index[key] == nil {
s.index[key] = make(map[string][]int)
}
s.index[key][val] = append(s.index[key][val], idx)
}
}
return nil
}
// Clear очищает индекс коллекции.
func (s *InMemoryIndexStore) Clear(collection string) error {
s.mu.Lock()
defer s.mu.Unlock()
prefix := collection + ":"
for k := range s.index {
if len(k) > len(prefix) && k[:len(prefix)] == prefix {
delete(s.index, k)
}
}
return nil
}
// GetSnapshotForTest возвращает копию индекса для тестов. Доступ только при accessKey == "give_me_cache".
// Ключ — "collection:field", значение — map[string][]int (значение поля -> номера строк).
func (s *InMemoryIndexStore) GetSnapshotForTest(accessKey string) map[string]any {
const testAccessKey = "give_me_cache"
if accessKey != testAccessKey {
return map[string]any{}
}
s.mu.RLock()
defer s.mu.RUnlock()
out := make(map[string]any, len(s.index))
for k, bucket := range s.index {
cp := make(map[string][]int, len(bucket))
for val, idxs := range bucket {
idxs2 := make([]int, len(idxs))
copy(idxs2, idxs)
cp[val] = idxs2
}
out[k] = cp
}
return out
}

177
pkg/linedb/index_memcached.go Normal file
View File

@@ -0,0 +1,177 @@
package linedb
import (
"encoding/json"
"fmt"
"sync"
)
// MemcachedClient — минимальный интерфейс для работы с Memcached.
// Реализуйте его поверх вашего клиента (например github.com/bradfitz/gomemcache/memcache).
type MemcachedClient interface {
Get(key string) ([]byte, error)
Set(key string, value []byte, expireSeconds int) error
Delete(key string) error
}
// MemcachedIndexStore — реализация IndexStore с хранением индексов в Memcached.
type MemcachedIndexStore struct {
client MemcachedClient
prefix string
expireSec int
mu sync.Mutex
}
// MemcachedIndexStoreOptions — опции для MemcachedIndexStore.
type MemcachedIndexStoreOptions struct {
Client MemcachedClient
KeyPrefix string // префикс ключей (по умолчанию "linedb:idx:")
ExpireSeconds int // TTL записей (0 = без истечения)
}
// NewMemcachedIndexStore создаёт IndexStore с бэкендом Memcached.
func NewMemcachedIndexStore(opts MemcachedIndexStoreOptions) (*MemcachedIndexStore, error) {
if opts.Client == nil {
return nil, fmt.Errorf("MemcachedClient is required")
}
prefix := opts.KeyPrefix
if prefix == "" {
prefix = "linedb:idx:"
}
return &MemcachedIndexStore{
client: opts.Client,
prefix: prefix,
expireSec: opts.ExpireSeconds,
}, nil
}
func (s *MemcachedIndexStore) memKey(collection, field, value string) string {
// Memcached ограничивает длину ключа (обычно 250 байт)
k := s.prefix + collection + ":" + field + ":" + value
if len(k) > 250 {
// Хэшируем длинные значения — упрощённо берём последние 200 символов
if len(value) > 200 {
k = s.prefix + collection + ":" + field + ":" + value[len(value)-200:]
}
}
return k
}
// IndexRecord добавляет запись в индекс.
func (s *MemcachedIndexStore) IndexRecord(collection string, fields []string, record map[string]any, lineIndex int) {
if len(fields) == 0 || record == nil {
return
}
for _, field := range fields {
val := getFieldValue(record, field)
key := s.memKey(collection, field, val)
var list []int
if data, err := s.client.Get(key); err == nil && len(data) > 0 {
_ = json.Unmarshal(data, &list)
}
list = append(list, lineIndex)
if data, err := json.Marshal(list); err == nil {
_ = s.client.Set(key, data, s.expireSec)
}
}
}
// UnindexRecord удаляет запись из индекса.
func (s *MemcachedIndexStore) UnindexRecord(collection string, fields []string, record map[string]any, lineIndex int) {
if len(fields) == 0 || record == nil {
return
}
for _, field := range fields {
val := getFieldValue(record, field)
key := s.memKey(collection, field, val)
data, err := s.client.Get(key)
if err != nil || len(data) == 0 {
continue
}
var list []int
if json.Unmarshal(data, &list) != nil {
continue
}
var newList []int
for _, i := range list {
if i != lineIndex {
newList = append(newList, i)
}
}
if len(newList) == 0 {
_ = s.client.Delete(key)
} else if data2, err := json.Marshal(newList); err == nil {
_ = s.client.Set(key, data2, s.expireSec)
}
}
}
// Lookup ищет индексы строк по полю и значению.
func (s *MemcachedIndexStore) Lookup(collection, field, value string) ([]int, error) {
key := s.memKey(collection, field, value)
data, err := s.client.Get(key)
if err != nil {
return nil, nil
}
if len(data) == 0 {
return nil, nil
}
var indexes []int
if json.Unmarshal(data, &indexes) != nil {
return nil, nil
}
// Возвращаем копию, чтобы вызывающий код не модифицировал внутренний срез
out := make([]int, len(indexes))
copy(out, indexes)
return out, nil
}
// Rebuild перестраивает индекс коллекции (удаляет старые ключи по префиксу и записывает новые).
func (s *MemcachedIndexStore) Rebuild(collection string, fields []string, records []any) error {
s.mu.Lock()
defer s.mu.Unlock()
// Memcached не поддерживает перечисление ключей по шаблону.
// Стратегия: перезаписываем все ключи для текущих записей.
// Старые ключи с другими значениями останутся до TTL.
// Строим value -> []lineIndex для каждого поля
byFieldValue := make(map[string]map[string][]int)
for idx, rec := range records {
recMap, ok := rec.(map[string]any)
if !ok {
continue
}
for _, field := range fields {
val := getFieldValue(recMap, field)
if byFieldValue[field] == nil {
byFieldValue[field] = make(map[string][]int)
}
byFieldValue[field][val] = append(byFieldValue[field][val], idx)
}
}
for field, valMap := range byFieldValue {
for val, list := range valMap {
key := s.memKey(collection, field, val)
data, err := json.Marshal(list)
if err != nil {
continue
}
if err := s.client.Set(key, data, s.expireSec); err != nil {
return fmt.Errorf("memcached set %s: %w", key, err)
}
}
}
return nil
}
// Clear очищает индекс коллекции. Memcached не поддерживает delete by pattern,
// поэтому метод является заглушкой — старые ключи истекут по TTL.
func (s *MemcachedIndexStore) Clear(collection string) error {
_ = collection
// Опционально: сохранить в отдельном ключе список всех ключей индекса
// и удалять их по одному. Упрощённо — ничего не делаем.
return nil
}

291
pkg/linedb/jsonl_file.go
View File

@@ -253,8 +253,10 @@ func NewJSONLFile(filename string, cypherKey string, options JSONLFileOptions) *
// Init инициализирует файл
func (j *JSONLFile) Init(force bool, options LineDbAdapterOptions) error {
j.mutex.Lock()
defer j.mutex.Unlock()
if !options.InTransaction {
j.mutex.Lock()
defer j.mutex.Unlock()
}
if j.initialized && !force {
return nil
@@ -350,8 +352,10 @@ func (j *JSONLFile) normalizeExistingFile() error {
// Read читает все записи из файла
func (j *JSONLFile) Read(options LineDbAdapterOptions) ([]any, error) {
j.mutex.RLock()
defer j.mutex.RUnlock()
if !options.InTransaction {
j.mutex.RLock()
defer j.mutex.RUnlock()
}
if !j.initialized {
return nil, fmt.Errorf("file not initialized")
@@ -398,10 +402,285 @@ func (j *JSONLFile) Read(options LineDbAdapterOptions) ([]any, error) {
return records, scanner.Err()
}
// ReadByLineIndexes читает записи по номерам строк (0-based) с использованием random access.
// Ожидается, что файл нормализован и каждая строка имеет длину allocSize байт (включая \n),
// как это обеспечивает Init/normalizeExistingFile/rewriteFile.
func (j *JSONLFile) ReadByLineIndexes(indexes []int, options LineDbAdapterOptions) ([]any, error) {
// Внутри транзакций (options.InTransaction == true) блокировка внешним кодом
// уже обеспечена, повторный лок мог бы привести к дедлоку.
if !options.InTransaction {
j.mutex.RLock()
defer j.mutex.RUnlock()
}
if !j.initialized {
return nil, fmt.Errorf("file not initialized")
}
if len(indexes) == 0 {
return []any{}, nil
}
// Копируем, сортируем и убираем дубликаты
sorted := make([]int, len(indexes))
copy(sorted, indexes)
sort.Ints(sorted)
uniq := sorted[:0]
prev := -1
for _, v := range sorted {
if v < 0 {
continue
}
if v != prev {
uniq = append(uniq, v)
prev = v
}
}
file, err := os.Open(j.filename)
if err != nil {
return nil, fmt.Errorf("failed to open file: %w", err)
}
defer file.Close()
var records []any
buf := make([]byte, j.allocSize)
for _, lineIndex := range uniq {
offset := int64(lineIndex) * int64(j.allocSize)
if _, err := file.Seek(offset, io.SeekStart); err != nil {
return nil, fmt.Errorf("failed to seek to offset %d: %w", offset, err)
}
n, err := io.ReadFull(file, buf)
if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
return nil, fmt.Errorf("failed to read line at index %d: %w", lineIndex, err)
}
if n <= 0 {
continue
}
line := string(buf[:n])
line = strings.TrimRight(line, "\n")
line = strings.TrimRight(line, " ")
if strings.TrimSpace(line) == "" {
continue
}
// Расшифровываем если нужно (только cypherKey; Encode обрабатывается в jsonUnmarshal)
if j.cypherKey != "" && !j.options.Encode {
decoded, err := base64.StdEncoding.DecodeString(line)
if err != nil {
if j.options.SkipInvalidLines {
continue
}
return nil, fmt.Errorf("failed to decode base64: %w", err)
}
line = string(decoded)
}
var record any
if err := j.jsonUnmarshal([]byte(line), &record); err != nil {
if j.options.SkipInvalidLines {
continue
}
return nil, fmt.Errorf("failed to unmarshal JSON: %w", err)
}
records = append(records, record)
}
return records, nil
}
// ReadByLineIndexesWithPositions как ReadByLineIndexes, но возвращает пары (record, lineIndex).
// Нужно для точечного Update, когда требуется знать позицию каждой записи.
func (j *JSONLFile) ReadByLineIndexesWithPositions(indexes []int, options LineDbAdapterOptions) ([]any, []int, error) {
if !options.InTransaction {
j.mutex.RLock()
defer j.mutex.RUnlock()
}
if !j.initialized {
return nil, nil, fmt.Errorf("file not initialized")
}
if len(indexes) == 0 {
return []any{}, []int{}, nil
}
sorted := make([]int, len(indexes))
copy(sorted, indexes)
sort.Ints(sorted)
uniq := sorted[:0]
prev := -1
for _, v := range sorted {
if v < 0 {
continue
}
if v != prev {
uniq = append(uniq, v)
prev = v
}
}
file, err := os.Open(j.filename)
if err != nil {
return nil, nil, fmt.Errorf("failed to open file: %w", err)
}
defer file.Close()
var records []any
var positions []int
buf := make([]byte, j.allocSize)
for _, lineIndex := range uniq {
offset := int64(lineIndex) * int64(j.allocSize)
if _, err := file.Seek(offset, io.SeekStart); err != nil {
return nil, nil, fmt.Errorf("failed to seek to offset %d: %w", offset, err)
}
n, err := io.ReadFull(file, buf)
if err != nil && err != io.EOF && err != io.ErrUnexpectedEOF {
return nil, nil, fmt.Errorf("failed to read line at index %d: %w", lineIndex, err)
}
if n <= 0 {
continue
}
line := string(buf[:n])
line = strings.TrimRight(line, "\n")
line = strings.TrimRight(line, " ")
if strings.TrimSpace(line) == "" {
continue
}
if j.cypherKey != "" && !j.options.Encode {
decoded, err := base64.StdEncoding.DecodeString(line)
if err != nil {
if j.options.SkipInvalidLines {
continue
}
return nil, nil, fmt.Errorf("failed to decode base64: %w", err)
}
line = string(decoded)
}
var record any
if err := j.jsonUnmarshal([]byte(line), &record); err != nil {
if j.options.SkipInvalidLines {
continue
}
return nil, nil, fmt.Errorf("failed to unmarshal JSON: %w", err)
}
records = append(records, record)
positions = append(positions, lineIndex)
}
return records, positions, nil
}
// WriteAtLineIndexes точечно записывает записи по заданным номерам строк (0-based).
// records и lineIndexes должны быть одинаковой длины и в одном порядке.
func (j *JSONLFile) WriteAtLineIndexes(records []any, lineIndexes []int, options LineDbAdapterOptions) error {
if !options.InTransaction {
j.mutex.Lock()
defer j.mutex.Unlock()
}
if !j.initialized {
return fmt.Errorf("file not initialized")
}
if len(records) != len(lineIndexes) {
return fmt.Errorf("records and lineIndexes length mismatch")
}
if len(records) == 0 {
return nil
}
file, err := os.OpenFile(j.filename, os.O_RDWR, 0644)
if err != nil {
return fmt.Errorf("failed to open file for write: %w", err)
}
defer file.Close()
maxLineLen := j.allocSize - 1
if maxLineLen < 1 {
maxLineLen = 1
}
for i, record := range records {
lineIndex := lineIndexes[i]
jsonData, err := j.jsonMarshal(record)
if err != nil {
return fmt.Errorf("marshal record at index %d: %w", lineIndex, err)
}
line := string(jsonData)
if j.cypherKey != "" && !j.options.Encode {
line = base64.StdEncoding.EncodeToString([]byte(line))
}
if len(line) > maxLineLen {
return fmt.Errorf("record at line %d size %d exceeds allocSize-1 (%d)", lineIndex, len(line), maxLineLen)
}
if len(line) < maxLineLen {
line += strings.Repeat(" ", maxLineLen-len(line))
}
line += "\n"
offset := int64(lineIndex) * int64(j.allocSize)
if _, err := file.Seek(offset, io.SeekStart); err != nil {
return fmt.Errorf("seek to %d: %w", offset, err)
}
if _, err := file.WriteString(line); err != nil {
return fmt.Errorf("write at line %d: %w", lineIndex, err)
}
}
return nil
}
// BlankLinesAtPositions затирает строки пробелами (allocSize-1 + \n). При чтении они пропускаются.
func (j *JSONLFile) BlankLinesAtPositions(lineIndexes []int, options LineDbAdapterOptions) error {
if !options.InTransaction {
j.mutex.Lock()
defer j.mutex.Unlock()
}
if !j.initialized {
return fmt.Errorf("file not initialized")
}
if len(lineIndexes) == 0 {
return nil
}
file, err := os.OpenFile(j.filename, os.O_RDWR, 0644)
if err != nil {
return fmt.Errorf("failed to open file for blank: %w", err)
}
defer file.Close()
blank := strings.Repeat(" ", j.allocSize-1) + "\n"
for _, lineIndex := range lineIndexes {
offset := int64(lineIndex) * int64(j.allocSize)
if _, err := file.Seek(offset, io.SeekStart); err != nil {
return fmt.Errorf("seek to %d: %w", offset, err)
}
if _, err := file.WriteString(blank); err != nil {
return fmt.Errorf("write blank at line %d: %w", lineIndex, err)
}
}
return nil
}
// LineCount возвращает число строк в файле (fileSize / allocSize).
// Используется для точечного индексирования после Write.
func (j *JSONLFile) LineCount() (int, error) {
info, err := os.Stat(j.filename)
if err != nil {
if os.IsNotExist(err) {
return 0, nil
}
return 0, err
}
return int(info.Size()) / j.allocSize, nil
}
// Write записывает данные в файл
func (j *JSONLFile) Write(data any, options LineDbAdapterOptions) error {
j.mutex.Lock()
defer j.mutex.Unlock()
if !options.InTransaction {
j.mutex.Lock()
defer j.mutex.Unlock()
}
if !j.initialized {
return fmt.Errorf("file not initialized")

504
pkg/linedb/line_db.go
View File

@@ -23,10 +23,11 @@ type LineDb struct {
cacheExternal *RecordCache
nextIDFn func(any, string) (any, error)
lastIDManager *LastIDManager
// inTransaction bool
cacheTTL time.Duration
constructorOptions *LineDbOptions
initOptions *LineDbInitOptions
indexStore IndexStore
reindexDone chan struct{} // закрывается при Close, останавливает горутину ReindexByTimer
}
// NewLineDb создает новый экземпляр LineDb
@@ -57,6 +58,8 @@ func NewLineDb(options *LineDbOptions, adapters ...*JSONLFile) *LineDb {
db.collections[collectionName] = adapter.GetFilename()
}
db.indexStore = options.IndexStore
return db
}
@@ -119,6 +122,41 @@ func (db *LineDb) Init(force bool, initOptions *LineDbInitOptions) error {
db.collections[collectionName] = filename
}
// Индексы: создаём in-memory по умолчанию, если не задан IndexStore и есть IndexedFields
if db.indexStore == nil {
for _, opt := range initOptions.Collections {
if len(opt.IndexedFields) > 0 {
db.indexStore = NewInMemoryIndexStore()
break
}
}
}
if db.indexStore != nil {
for _, opt := range initOptions.Collections {
if len(opt.IndexedFields) == 0 || db.isCollectionPartitioned(opt.CollectionName) {
continue
}
adapter := db.adapters[opt.CollectionName]
if adapter == nil {
continue
}
records, err := adapter.Read(LineDbAdapterOptions{})
if err != nil {
return fmt.Errorf("failed to read collection %s for index rebuild: %w", opt.CollectionName, err)
}
if err := db.indexStore.Rebuild(opt.CollectionName, opt.IndexedFields, records); err != nil {
return fmt.Errorf("failed to rebuild index for %s: %w", opt.CollectionName, err)
}
}
}
// Периодический ребилд индексов
if initOptions.ReindexByTimer > 0 && db.indexStore != nil {
db.reindexDone = make(chan struct{})
interval := time.Duration(initOptions.ReindexByTimer) * time.Millisecond
go db.reindexByTimerLoop(interval)
}
return nil
}
@@ -247,6 +285,20 @@ func (db *LineDb) Insert(data any, collectionName string, options LineDbAdapterO
}
}
// Обновляем индекс (полная пересборка по коллекции)
if db.indexStore != nil {
opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 && !db.isCollectionPartitioned(collectionName) {
adapter, exists := db.adapters[collectionName]
if exists {
allRecords, readErr := adapter.Read(LineDbAdapterOptions{})
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, opts.IndexedFields, allRecords)
}
}
}
}
return nil
}
@@ -284,7 +336,30 @@ func (db *LineDb) Write(data any, collectionName string, options LineDbAdapterOp
return fmt.Errorf("collection %s not found", collectionName)
}
return adapter.Write(data, options)
var startLine int
if options.DoIndexing && db.indexStore != nil {
if c, err := adapter.LineCount(); err == nil {
startLine = c
}
}
if err := adapter.Write(data, options); err != nil {
return err
}
// Точечное индексирование при DoIndexing
if options.DoIndexing && db.indexStore != nil {
opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 {
dataArray := db.normalizeDataArray(data)
for i, record := range dataArray {
if m, err := db.toMap(record); err == nil {
db.indexStore.IndexRecord(collectionName, opts.IndexedFields, m, startLine+i)
}
}
}
}
return nil
}
// Update обновляет записи в коллекции
@@ -294,17 +369,20 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li
db.mutex.Lock()
defer db.mutex.Unlock()
}
if collectionName == "" {
collectionName = db.getFirstCollection()
}
// Конвертируем data в map (struct или map)
dataMap, err := db.toMap(data)
if err != nil {
return nil, fmt.Errorf("invalid update data format: %w", err)
}
// Нормализуем фильтр (строка/struct -> map)
normFilter, err := db.normalizeFilter(filter)
if err != nil {
return nil, err
}
filter = normFilter
// Проверяем конфликт ID
if filterMap, ok := filter.(map[string]any); ok {
if dataMap["id"] != nil && filterMap["id"] != nil {
@@ -321,7 +399,6 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li
if err := db.checkUniqueFieldsUpdate(dataMap, filter, collectionName, options); err != nil {
return nil, err
}
// Проверяем партиционирование
if db.isCollectionPartitioned(collectionName) {
return db.updatePartitioned(dataMap, collectionName, filter, options)
@@ -333,7 +410,41 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li
return nil, fmt.Errorf("collection %s not found", collectionName)
}
return adapter.Update(dataMap, filter, options)
// Пробуем точечный Update через индекс (без полного чтения файла)
opts := db.getCollectionOptions(collectionName)
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
result, used, err := db.tryIndexUpdate(adapter, filter, dataMap, collectionName, opts.IndexedFields, options)
if err != nil {
return nil, err
}
if used {
if db.cacheExternal != nil {
ids := extractIDsFromRecords(result)
db.cacheExternal.ClearEntriesContainingIDs(ids)
}
return result, nil
}
}
result, err := adapter.Update(dataMap, filter, options)
if err != nil {
return nil, err
}
// Перестраиваем индекс после Update
if db.indexStore != nil {
opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 {
allRecords, readErr := adapter.Read(LineDbAdapterOptions{})
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, opts.IndexedFields, allRecords)
}
}
}
if db.cacheExternal != nil {
ids := extractIDsFromRecords(result)
db.cacheExternal.ClearEntriesContainingIDs(ids)
}
return result, nil
}
// Delete удаляет записи из коллекции
@@ -348,6 +459,13 @@ func (db *LineDb) Delete(data any, collectionName string, options LineDbAdapterO
collectionName = db.getFirstCollection()
}
// Нормализуем фильтр удаления (строка/struct -> map)
normFilter, err := db.normalizeFilter(data)
if err != nil {
return nil, err
}
data = normFilter
// Проверяем партиционирование
if db.isCollectionPartitioned(collectionName) {
return db.deletePartitioned(data, collectionName, options)
@@ -359,7 +477,40 @@ func (db *LineDb) Delete(data any, collectionName string, options LineDbAdapterO
return nil, fmt.Errorf("collection %s not found", collectionName)
}
return adapter.Delete(data, options)
opts := db.getCollectionOptions(collectionName)
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
result, used, err := db.tryIndexDelete(adapter, data, collectionName, opts.IndexedFields, options)
if err != nil {
return nil, err
}
if used {
if db.cacheExternal != nil {
ids := extractIDsFromRecords(result)
db.cacheExternal.ClearEntriesContainingIDs(ids)
}
return result, nil
}
}
result, err := adapter.Delete(data, options)
if err != nil {
return nil, err
}
// Перестраиваем индекс после Delete
if db.indexStore != nil {
opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 {
allRecords, readErr := adapter.Read(LineDbAdapterOptions{})
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, opts.IndexedFields, allRecords)
}
}
}
if db.cacheExternal != nil {
ids := extractIDsFromRecords(result)
db.cacheExternal.ClearEntriesContainingIDs(ids)
}
return result, nil
}
// Select выполняет выборку с поддержкой цепочки
@@ -382,6 +533,13 @@ func (db *LineDb) ReadByFilter(filter any, collectionName string, options LineDb
defer db.mutex.RUnlock()
}
// Нормализуем фильтр (строка/struct -> map)
normFilter, err := db.normalizeFilter(filter)
if err != nil {
return nil, err
}
filter = normFilter
if collectionName == "" {
collectionName = db.getFirstCollection()
}
@@ -400,12 +558,21 @@ func (db *LineDb) ReadByFilter(filter any, collectionName string, options LineDb
return db.readByFilterPartitioned(filter, collectionName, options)
}
// Обычная фильтрация
adapter, exists := db.adapters[collectionName]
if !exists {
return nil, fmt.Errorf("collection %s not found", collectionName)
}
// Используем индекс, если фильтр — одно поле из IndexedFields
if db.indexStore != nil {
if hit, result, err := db.tryIndexLookup(adapter, filter, collectionName, options); hit && err == nil {
if db.cacheExternal != nil && !options.InTransaction {
db.cacheExternal.Set(db.generateCacheKey(filter, collectionName), result)
}
return result, nil
}
}
result, err := adapter.ReadByFilter(filter, options)
if err != nil {
return nil, err
@@ -466,6 +633,12 @@ func (db *LineDb) Close() {
db.mutex.Lock()
defer db.mutex.Unlock()
// Останавливаем горутину ReindexByTimer
if db.reindexDone != nil {
close(db.reindexDone)
db.reindexDone = nil
}
// Закрываем все адаптеры
for _, adapter := range db.adapters {
adapter.Destroy()
@@ -483,6 +656,38 @@ func (db *LineDb) Close() {
db.partitionFunctions = make(map[string]func(any) string)
}
// reindexByTimerLoop выполняет полный ребилд всех индексов с заданным интервалом.
// Работает в отдельной горутине. Останавливается при закрытии db.reindexDone.
func (db *LineDb) reindexByTimerLoop(interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for {
select {
case <-db.reindexDone:
return
case <-ticker.C:
db.mutex.Lock()
if db.indexStore != nil && db.initOptions != nil {
for _, opt := range db.initOptions.Collections {
if len(opt.IndexedFields) == 0 || db.isCollectionPartitioned(opt.CollectionName) {
continue
}
adapter := db.adapters[opt.CollectionName]
if adapter == nil {
continue
}
records, err := adapter.Read(LineDbAdapterOptions{})
if err != nil {
continue
}
_ = db.indexStore.Rebuild(opt.CollectionName, opt.IndexedFields, records)
}
}
db.mutex.Unlock()
}
}
}
// Вспомогательные методы
func (db *LineDb) getFirstCollection() string {
@@ -499,6 +704,179 @@ func (db *LineDb) getBaseCollectionName(collectionName string) string {
return collectionName
}
// tryIndexLookup пытается выполнить поиск через индекс.
// Если фильтр — map с несколькими полями, выбирается первое поле из IndexedFields,
// по нему читаются строки через индекс, затем в памяти накладывается полный фильтр.
// Возвращает (true, result) если индекс использован, (false, nil) иначе.
func (db *LineDb) tryIndexLookup(adapter *JSONLFile, filter any, collectionName string, options LineDbAdapterOptions) (bool, []any, error) {
filterMap, ok := filter.(map[string]any)
if !ok || len(filterMap) == 0 {
return false, nil, nil
}
opts := adapter.GetOptions()
if len(opts.IndexedFields) == 0 {
return false, nil, nil
}
// Находим первое индексируемое поле, которое есть в фильтре
var field string
var value any
for _, idxField := range opts.IndexedFields {
if v, exists := filterMap[idxField]; exists {
field = idxField
value = v
break
}
}
if field == "" {
return false, nil, nil
}
valStr := valueToIndexKey(value)
indexes, err := db.indexStore.Lookup(collectionName, field, valStr)
if err != nil || indexes == nil || len(indexes) == 0 {
return false, nil, err
}
records, err := adapter.ReadByLineIndexes(indexes, options)
if err != nil {
return false, nil, err
}
// Если фильтр — одно поле, уже отфильтровано индексом
if len(filterMap) == 1 {
return true, records, nil
}
// Иначе накладываем полный фильтр по остальным полям в памяти
var filtered []any
for _, rec := range records {
if db.matchesFilter(rec, filter, options.StrictCompare) {
filtered = append(filtered, rec)
}
}
return true, filtered, nil
}
// tryIndexUpdate выполняет точечное обновление через индекс. Возвращает (result, used, err).
func (db *LineDb) tryIndexUpdate(adapter *JSONLFile, filter any, dataMap map[string]any, collectionName string, indexedFields []string, options LineDbAdapterOptions) ([]any, bool, error) {
filterMap, ok := filter.(map[string]any)
if !ok || len(filterMap) == 0 {
return nil, false, nil
}
var field string
var value any
for _, idxField := range indexedFields {
if v, exists := filterMap[idxField]; exists {
field = idxField
value = v
break
}
}
if field == "" {
return nil, false, nil
}
valStr := valueToIndexKey(value)
indexes, err := db.indexStore.Lookup(collectionName, field, valStr)
if err != nil || indexes == nil || len(indexes) == 0 {
return nil, false, err
}
opt := options
opt.InTransaction = true
records, positions, err := adapter.ReadByLineIndexesWithPositions(indexes, opt)
if err != nil {
return nil, false, err
}
var toUpdate []any
var toUpdatePos []int
for i, rec := range records {
if !db.matchesFilter(rec, filter, options.StrictCompare) {
continue
}
toUpdate = append(toUpdate, rec)
toUpdatePos = append(toUpdatePos, positions[i])
}
if len(toUpdate) == 0 {
return []any{}, true, nil
}
var updated []any
for _, rec := range toUpdate {
m, ok := rec.(map[string]any)
if !ok {
continue
}
upd := make(map[string]any)
for k, v := range m {
upd[k] = v
}
for k, v := range dataMap {
upd[k] = v
}
updated = append(updated, upd)
}
if err := adapter.WriteAtLineIndexes(updated, toUpdatePos, opt); err != nil {
return nil, false, err
}
for i, rec := range toUpdate {
if m, ok := rec.(map[string]any); ok {
db.indexStore.UnindexRecord(collectionName, indexedFields, m, toUpdatePos[i])
}
}
for i, rec := range updated {
if m, ok := rec.(map[string]any); ok {
db.indexStore.IndexRecord(collectionName, indexedFields, m, toUpdatePos[i])
}
}
return updated, true, nil
}
// tryIndexDelete выполняет точечное удаление через индекс. Возвращает (deletedRecords, used, err).
func (db *LineDb) tryIndexDelete(adapter *JSONLFile, filter any, collectionName string, indexedFields []string, options LineDbAdapterOptions) ([]any, bool, error) {
filterMap, ok := filter.(map[string]any)
if !ok || len(filterMap) == 0 {
return nil, false, nil
}
var field string
var value any
for _, idxField := range indexedFields {
if v, exists := filterMap[idxField]; exists {
field = idxField
value = v
break
}
}
if field == "" {
return nil, false, nil
}
valStr := valueToIndexKey(value)
indexes, err := db.indexStore.Lookup(collectionName, field, valStr)
if err != nil || indexes == nil || len(indexes) == 0 {
return nil, false, err
}
opt := options
opt.InTransaction = true
records, positions, err := adapter.ReadByLineIndexesWithPositions(indexes, opt)
if err != nil {
return nil, false, err
}
var toDel []any
var toDelPos []int
for i, rec := range records {
if !db.matchesFilter(rec, filter, options.StrictCompare) {
continue
}
toDel = append(toDel, rec)
toDelPos = append(toDelPos, positions[i])
}
if len(toDel) == 0 {
return []any{}, true, nil
}
if err := adapter.BlankLinesAtPositions(toDelPos, opt); err != nil {
return nil, false, err
}
for i, rec := range toDel {
if m, ok := rec.(map[string]any); ok {
db.indexStore.UnindexRecord(collectionName, indexedFields, m, toDelPos[i])
}
}
return toDel, true, nil
}
// getCollectionOptions возвращает опции коллекции (для партиционированных — опции базовой коллекции)
func (db *LineDb) getCollectionOptions(collectionName string) *JSONLFileOptions {
if db.initOptions == nil {
@@ -860,11 +1238,88 @@ func (db *LineDb) compareIDs(a, b any) bool {
return a == b
}
// extractIDsFromRecords извлекает id из списка записей (map[string]any).
func extractIDsFromRecords(records []any) []any {
if len(records) == 0 {
return nil
}
ids := make([]any, 0, len(records))
for _, rec := range records {
m, ok := rec.(map[string]any)
if !ok {
continue
}
if id, exists := m["id"]; exists && id != nil {
ids = append(ids, id)
}
}
return ids
}
func (db *LineDb) generateCacheKey(filter any, collectionName string) string {
// Упрощенная реализация генерации ключа кэша
return fmt.Sprintf("%s:%v", collectionName, filter)
}
// normalizeFilter приводит произвольный filter к более удобной форме:
// - string вида "field:value, field2:value2" -> map[string]any{"field": "value", "field2": "value2"}
// - простая string без ":" -> map[string]any{"id": value}
// - struct -> map[string]any (через toMap)
// - map[string]any и func(any) bool не меняются.
func (db *LineDb) normalizeFilter(filter any) (any, error) {
switch f := filter.(type) {
case nil:
return nil, nil
case map[string]any:
return f, nil
case func(any) bool:
return f, nil
case string:
s := strings.TrimSpace(f)
if s == "" {
return nil, nil
}
// pattern: field:value, field2:value2
if strings.Contains(s, ":") {
result := make(map[string]any)
parts := strings.Split(s, ",")
for _, p := range parts {
p = strings.TrimSpace(p)
if p == "" {
continue
}
kv := strings.SplitN(p, ":", 2)
if len(kv) != 2 {
continue
}
key := strings.TrimSpace(kv[0])
val := strings.TrimSpace(kv[1])
if key == "" {
continue
}
result[key] = val
}
if len(result) == 0 {
return map[string]any{"id": s}, nil
}
return result, nil
}
// простая строка — считаем фильтром по id
return map[string]any{"id": s}, nil
default:
if m, ok := f.(map[string]any); ok {
return m, nil
}
// Пытаемся трактовать как struct и конвертировать в map
m, err := db.toMap(f)
if err != nil {
// если не получилось — возвращаем как есть
return filter, nil
}
return m, nil
}
}
func (db *LineDb) updatePartitioned(data any, collectionName string, filter any, options LineDbAdapterOptions) ([]any, error) {
// Получаем все партиции
partitionFiles, err := db.getPartitionFiles(collectionName)
@@ -1198,6 +1653,37 @@ func (db *LineDb) GetCacheMap() map[string]*CacheEntry {
return make(map[string]*CacheEntry)
}
// GetCacheForTest возвращает сырую карту кэша для тестов: ключ — ключ кэша, значение — Data записи.
// Доступ только при accessKey == "give_me_cache", иначе возвращается пустая мапа.
func (db *LineDb) GetCacheForTest(accessKey string) map[string]any {
const testAccessKey = "give_me_cache"
if accessKey != testAccessKey {
return map[string]any{}
}
if db.cacheExternal == nil {
return map[string]any{}
}
flat := db.cacheExternal.GetFlatCacheMap()
out := make(map[string]any, len(flat))
for k, entry := range flat {
out[k] = entry.Data
}
return out
}
// GetIndexSnapshotForTest возвращает снимок индекса для тестов (только InMemoryIndexStore).
// Доступ только при accessKey == "give_me_cache". Ключ — "collection:field", значение — map[value][]int (номера строк).
func (db *LineDb) GetIndexSnapshotForTest(accessKey string) map[string]any {
const testAccessKey = "give_me_cache"
if accessKey != testAccessKey || db.indexStore == nil {
return map[string]any{}
}
if mem, ok := db.indexStore.(*InMemoryIndexStore); ok {
return mem.GetSnapshotForTest(testAccessKey)
}
return map[string]any{}
}
func (db *LineDb) GetFirstCollection() string {
return db.getFirstCollection()
}

21
pkg/linedb/types.go
View File

@@ -38,20 +38,22 @@ func (r *BaseRecord) SetTimestamp(timestamp *time.Time) {
// LineDbOptions представляет опции для создания LineDb
// Соответствует TypeScript интерфейсу LineDbOptions
type LineDbOptions struct {
CacheSize int `json:"cacheSize,omitempty"`
CacheTTL time.Duration `json:"cacheTTL,omitempty"`
DBFolder string `json:"dbFolder,omitempty"`
ObjName string `json:"objName,omitempty"`
CacheSize int `json:"cacheSize,omitempty"`
CacheTTL time.Duration `json:"cacheTTL,omitempty"`
DBFolder string `json:"dbFolder,omitempty"`
ObjName string `json:"objName,omitempty"`
IndexStore IndexStore `json:"-"` // Хранилище индексов (nil = не использовать; по умолчанию InMemoryIndexStore при наличии IndexedFields)
}
// LineDbInitOptions представляет опции инициализации LineDb
// Соответствует TypeScript интерфейсу LineDbInitOptions
type LineDbInitOptions struct {
CacheSize int `json:"cacheSize,omitempty"`
CacheTTL time.Duration `json:"cacheTTL,omitempty"`
Collections []JSONLFileOptions `json:"collections"`
DBFolder string `json:"dbFolder,omitempty"`
Partitions []PartitionCollection `json:"partitions,omitempty"`
CacheSize int `json:"cacheSize,omitempty"`
CacheTTL time.Duration `json:"cacheTTL,omitempty"`
Collections []JSONLFileOptions `json:"collections"`
DBFolder string `json:"dbFolder,omitempty"`
Partitions []PartitionCollection `json:"partitions,omitempty"`
ReindexByTimer int `json:"reindexByTimer,omitempty"` // интервал в ms; если > 0 — периодический полный ребилд всех индексов с блокировкой
}
// EmptyValueMode определяет, что считать "пустым" при проверке required/unique
@@ -177,6 +179,7 @@ type LineDbAdapterOptions struct {
SkipCheckExistingForWrite bool `json:"skipCheckExistingForWrite,omitempty"`
OptimisticRead bool `json:"optimisticRead,omitempty"`
ReturnChain bool `json:"returnChain,omitempty"`
DoIndexing bool `json:"doIndexing,omitempty"` // при true — после Write делать точечное индексирование новых записей
}
// TransactionOptions представляет опции транзакции

1
tests/data/test-linedb-index-enc/users.jsonl Normal file
View File

@@ -0,0 +1 @@
{"email":"alice@secret.com","id":1,"name":"alice"}

2
tests/data/test-point-update/users.jsonl Normal file
View File

@@ -0,0 +1,2 @@
{"email":"alice@test.com","id":1,"name":"alice"}
{"email":"bob_new@test.com","id":2,"name":"bob"}

419
tests/linedb_index_test.go Normal file
View File

@@ -0,0 +1,419 @@
package tests
import (
"os"
"testing"
"time"
"linedb/pkg/linedb"
)
// mockMemcached — in-memory реализация MemcachedClient для тестов.
type mockMemcached struct {
data map[string][]byte
}
func newMockMemcached() *mockMemcached {
return &mockMemcached{data: make(map[string][]byte)}
}
func (m *mockMemcached) Get(key string) ([]byte, error) {
v, ok := m.data[key]
if !ok {
return nil, nil
}
return v, nil
}
func (m *mockMemcached) Set(key string, value []byte, _ int) error {
m.data[key] = value
return nil
}
func (m *mockMemcached) Delete(key string) error {
delete(m.data, key)
return nil
}
func setupIndexedCollection(t *testing.T, indexStore linedb.IndexStore) (*linedb.LineDb, func()) {
t.Helper()
os.RemoveAll("./testdata")
opts := &linedb.LineDbOptions{}
if indexStore != nil {
opts.IndexStore = indexStore
}
initOptions := &linedb.LineDbInitOptions{
CacheSize: 100,
CacheTTL: time.Minute,
DBFolder: "./data/test-linedb-index",
Collections: []linedb.JSONLFileOptions{
{
CollectionName: "users",
AllocSize: 256,
IndexedFields: []string{"id", "email", "name"},
},
},
}
db := linedb.NewLineDb(opts)
if err := db.Init(false, initOptions); err != nil {
t.Fatalf("Init failed: %v", err)
}
return db, func() { db.Close(); os.RemoveAll("./testdata") }
}
func TestIndexInMemoryReadByFilter(t *testing.T) {
db, cleanup := setupIndexedCollection(t, nil) // nil = auto in-memory
defer cleanup()
opts := linedb.LineDbAdapterOptions{}
if err := db.Insert(map[string]any{"name": "alice", "email": "alice@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 1 failed: %v", err)
}
if err := db.Insert(map[string]any{"name": "bob", "email": "bob@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 2 failed: %v", err)
}
if err := db.Insert(map[string]any{"name": "charlie", "email": "charlie@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 3 failed: %v", err)
}
// Поиск по индексированному полю email
found, err := db.ReadByFilter("email:bob@test.com,name:bob", "users", opts)
// found, err := db.ReadByFilter(map[string]any{"email": "bob@test.com"}, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter failed: %v", err)
}
if len(found) != 1 {
t.Fatalf("Expected 1 record for email bob@test.com, got %d", len(found))
}
if m, ok := found[0].(map[string]any); ok {
if m["name"] != "bob" {
t.Errorf("Expected name bob, got %v", m["name"])
}
}
// Поиск по name
found2, err := db.ReadByFilter(map[string]any{"name": "alice"}, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter name failed: %v", err)
}
if len(found2) != 1 {
t.Fatalf("Expected 1 record for name alice, got %d", len(found2))
}
}
// TestIndexEncodedCollectionCache проверяет шифрованную коллекцию и двойное чтение одного фильтра (кэш).
func TestIndexEncodedCollectionCache(t *testing.T) {
err := os.RemoveAll("./data/test-linedb-index-enc")
if err != nil {
t.Logf("RemoveAll failed: %v", err)
}
initOptions := &linedb.LineDbInitOptions{
CacheSize: 100,
CacheTTL: time.Minute*10,
DBFolder: "./data/test-linedb-index-enc",
Collections: []linedb.JSONLFileOptions{
{
CollectionName: "users",
AllocSize: 512,
IndexedFields: []string{"id", "email", "name"},
Encode: false,
EncodeKey: "test-secret-key",
},
},
}
db := linedb.NewLineDb(nil)
if err := db.Init(false, initOptions); err != nil {
t.Fatalf("Init failed: %v", err)
}
defer func() { db.Close(); os.RemoveAll("./testdata") }()
opts := linedb.LineDbAdapterOptions{}
if err := db.Insert(map[string]any{"name": "alice", "email": "alice@secret.com"}, "users", opts); err != nil {
t.Fatalf("Insert 1 failed: %v", err)
}
if err := db.Insert(map[string]any{"name": "bob", "email": "bob@secret.com"}, "users", opts); err != nil {
t.Fatalf("Insert 2 failed: %v", err)
}
filter := map[string]any{"email": "bob@secret.com"}
// Первое чтение — из файла (с дешифровкой)
found1, err := db.ReadByFilter(filter, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter (1st) failed: %v", err)
}
if len(found1) != 1 {
t.Fatalf("Expected 1 record, got %d", len(found1))
}
if m, ok := found1[0].(map[string]any); ok {
if m["name"] != "bob" || m["email"] != "bob@secret.com" {
t.Errorf("Expected name=bob email=bob@secret.com, got %+v", m)
}
}
cacheSize1 := db.GetActualCacheSize()
if cacheSize1 < 1 {
t.Errorf("Expected cache to have at least 1 entry after first read, got %d", cacheSize1)
}
// Второе чтение — из кэша (должно вернуть те же данные без повторного чтения файла)
found2, err := db.ReadByFilter(filter, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter (2nd, cached) failed: %v", err)
}
if len(found2) != 1 {
t.Fatalf("Expected 1 record on second read, got %d", len(found2))
}
if m, ok := found2[0].(map[string]any); ok {
if m["name"] != "bob" || m["email"] != "bob@secret.com" {
t.Errorf("Expected name=bob email=bob@secret.com on cached read, got %+v", m)
}
}
// Кэш не должен расти при повторном запросе с тем же ключом
cacheSize2 := db.GetActualCacheSize()
if cacheSize2 != cacheSize1 {
t.Errorf("Cache size should stay same after cache hit: was %d, now %d", cacheSize1, cacheSize2)
}
// Обновляем запись (name) и проверяем, что кэш и индекс обновились
_, err = db.Update(
map[string]any{"name": "bob_updated"},
"users",
map[string]any{"email": "bob@secret.com"},
opts,
)
if err != nil {
t.Fatalf("Update failed: %v", err)
}
// После Update кэш сбрасывается — читаем снова, чтобы заполнить кэш актуальными данными
found3, err := db.ReadByFilter(filter, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter after update failed: %v", err)
}
if len(found3) != 1 {
t.Fatalf("Expected 1 record after update, got %d", len(found3))
}
if m, ok := found3[0].(map[string]any); ok && m["name"] != "bob_updated" {
t.Errorf("Expected name=bob_updated after update, got %v", m["name"])
}
// Проверяем сырой кэш: в нём должна быть запись с name=bob_updated
rawCache := db.GetCacheForTest("give_me_cache")
if len(rawCache) == 0 {
t.Error("Expected cache to have entries after read")
}
var foundInCache bool
for _, v := range rawCache {
arr, ok := v.([]any)
if !ok || len(arr) != 1 {
continue
}
m, ok := arr[0].(map[string]any)
if !ok {
continue
}
if m["email"] == "bob@secret.com" {
foundInCache = true
if m["name"] != "bob_updated" {
t.Errorf("Expected cached name=bob_updated, got %v", m["name"])
}
break
}
}
if !foundInCache {
t.Error("Expected to find bob record in raw cache with updated name")
}
// Проверяем индекс: по email bob@secret.com должна быть одна строка, по name bob_updated — тоже
idxSnapshot := db.GetIndexSnapshotForTest("give_me_cache")
if len(idxSnapshot) == 0 {
t.Error("Expected index snapshot to have entries")
}
if emailIdx, ok := idxSnapshot["users:email"].(map[string][]int); ok {
if lines, ok := emailIdx["bob@secret.com"]; !ok || len(lines) != 1 {
t.Errorf("Expected index users:email bob@secret.com to have 1 line, got %v", emailIdx["bob@secret.com"])
}
}
if nameIdx, ok := idxSnapshot["users:name"].(map[string][]int); ok {
if lines, ok := nameIdx["bob_updated"]; !ok || len(lines) != 1 {
t.Errorf("Expected index users:name bob_updated to have 1 line, got %v", nameIdx["bob_updated"])
}
}
// Удаляем запись и проверяем, что в кэше и в индексе её больше нет
_, err = db.Delete(map[string]any{"email": "bob@secret.com"}, "users", opts)
if err != nil {
t.Fatalf("Delete failed: %v", err)
}
// После Delete кэш сбрасывается — читаем снова
found4, err := db.ReadByFilter(filter, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter after delete failed: %v", err)
}
if len(found4) != 0 {
t.Fatalf("Expected 0 records after delete, got %d", len(found4))
}
// В сыром кэше не должно остаться записи bob
rawCache2 := db.GetCacheForTest("give_me_cache")
for _, v := range rawCache2 {
arr, ok := v.([]any)
if !ok || len(arr) != 1 {
continue
}
if m, ok := arr[0].(map[string]any); ok && m["email"] == "bob@secret.com" {
t.Error("Cached result after delete should not contain bob record")
}
}
// Индекс: bob@secret.com и bob_updated не должны быть в индексе (или пустые срезы)
idxSnapshot2 := db.GetIndexSnapshotForTest("give_me_cache")
if emailIdx, ok := idxSnapshot2["users:email"].(map[string][]int); ok {
if lines, has := emailIdx["bob@secret.com"]; has && len(lines) > 0 {
t.Errorf("After delete, index users:email bob@secret.com should be empty, got %v", lines)
}
}
if nameIdx, ok := idxSnapshot2["users:name"].(map[string][]int); ok {
if lines, has := nameIdx["bob_updated"]; has && len(lines) > 0 {
t.Errorf("After delete, index users:name bob_updated should be empty, got %v", lines)
}
}
}
func TestIndexExplicitInMemory(t *testing.T) {
store := linedb.NewInMemoryIndexStore()
db, cleanup := setupIndexedCollection(t, store)
defer cleanup()
opts := linedb.LineDbAdapterOptions{}
if err := db.Insert(map[string]any{"name": "x", "email": "x@y.com"}, "users", opts); err != nil {
t.Fatalf("Insert failed: %v", err)
}
found, err := db.ReadByFilter(map[string]any{"email": "x@y.com"}, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter failed: %v", err)
}
if len(found) != 1 {
t.Fatalf("Expected 1 record, got %d", len(found))
}
}
func TestIndexMemcachedStore(t *testing.T) {
mock := newMockMemcached()
store, err := linedb.NewMemcachedIndexStore(linedb.MemcachedIndexStoreOptions{
Client: mock,
})
if err != nil {
t.Fatalf("NewMemcachedIndexStore: %v", err)
}
db, cleanup := setupIndexedCollection(t, store)
defer cleanup()
opts := linedb.LineDbAdapterOptions{}
if err := db.Insert(map[string]any{"name": "mem", "email": "mem@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert failed: %v", err)
}
found, err := db.ReadByFilter(map[string]any{"email": "mem@test.com"}, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter failed: %v", err)
}
if len(found) != 1 {
t.Fatalf("Expected 1 record, got %d", len(found))
}
// Проверяем, что в mockMemcached что-то записалось
if len(mock.data) == 0 {
t.Error("Expected memcached store to have data")
}
}
func TestIndexUpdateRebuild(t *testing.T) {
db, cleanup := setupIndexedCollection(t, nil)
defer cleanup()
opts := linedb.LineDbAdapterOptions{}
if err := db.Insert(map[string]any{"name": "a", "email": "a@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert failed: %v", err)
}
_, err := db.Update(
map[string]any{"email": "a_updated@test.com"},
"users",
map[string]any{"email": "a@test.com"},
opts,
)
if err != nil {
t.Fatalf("Update failed: %v", err)
}
// Старый email не должен находиться
old, _ := db.ReadByFilter(map[string]any{"email": "a@test.com"}, "users", opts)
if len(old) != 0 {
t.Fatalf("Expected 0 records for old email, got %d", len(old))
}
// Новый — должен
newFound, _ := db.ReadByFilter(map[string]any{"email": "a_updated@test.com"}, "users", opts)
if len(newFound) != 1 {
t.Fatalf("Expected 1 record for new email, got %d", len(newFound))
}
}
func TestIndexDeleteRebuild(t *testing.T) {
db, cleanup := setupIndexedCollection(t, nil)
defer cleanup()
opts := linedb.LineDbAdapterOptions{}
if err := db.Insert(map[string]any{"name": "del", "email": "del@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert failed: %v", err)
}
if err := db.Insert(map[string]any{"name": "keep", "email": "keep@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 2 failed: %v", err)
}
_, err := db.Delete(map[string]any{"email": "del@test.com"}, "users", opts)
if err != nil {
t.Fatalf("Delete failed: %v", err)
}
found, _ := db.ReadByFilter(map[string]any{"email": "del@test.com"}, "users", opts)
if len(found) != 0 {
t.Fatalf("Expected 0 after delete, got %d", len(found))
}
kept, _ := db.ReadByFilter(map[string]any{"email": "keep@test.com"}, "users", opts)
if len(kept) != 1 {
t.Fatalf("Expected 1 kept record, got %d", len(kept))
}
}
func TestIndexNoIndexedFields(t *testing.T) {
os.RemoveAll("./testdata")
initOptions := &linedb.LineDbInitOptions{
CacheSize: 100,
CacheTTL: time.Minute,
DBFolder: "./testdata",
Collections: []linedb.JSONLFileOptions{
{
CollectionName: "users",
AllocSize: 256,
IndexedFields: nil, // без индексов
},
},
}
db := linedb.NewLineDb(nil)
if err := db.Init(false, initOptions); err != nil {
t.Fatalf("Init failed: %v", err)
}
defer func() { db.Close(); os.RemoveAll("./testdata") }()
if err := db.Insert(map[string]any{"name": "a", "email": "a@test.com"}, "users", linedb.LineDbAdapterOptions{}); err != nil {
t.Fatalf("Insert failed: %v", err)
}
// ReadByFilter всё равно работает (полный скан)
found, err := db.ReadByFilter(map[string]any{"email": "a@test.com"}, "users", linedb.LineDbAdapterOptions{})
if err != nil {
t.Fatalf("ReadByFilter failed: %v", err)
}
if len(found) != 1 {
t.Fatalf("Expected 1 record, got %d", len(found))
}
}

139
tests/linedb_point_update_test.go Normal file
View File

@@ -0,0 +1,139 @@
// Тесты для дебага точечного Update через индекс.
// Случай 1: изменение поля, не входящего в индекс (поиск по email, обновляем name).
// Случай 2: изменение поля, входящего в индекс (поиск по email, обновляем email).
package tests
import (
"os"
"testing"
"time"
"linedb/pkg/linedb"
)
func setupPointUpdateDB(t *testing.T) (*linedb.LineDb, func()) {
t.Helper()
dir := "./data/test-point-update"
os.RemoveAll(dir)
initOptions := &linedb.LineDbInitOptions{
CacheSize: 100,
CacheTTL: time.Minute,
DBFolder: dir,
Collections: []linedb.JSONLFileOptions{
{
CollectionName: "users",
AllocSize: 256,
IndexedFields: []string{"id", "email", "name"},
},
},
}
db := linedb.NewLineDb(nil)
if err := db.Init(false, initOptions); err != nil {
t.Fatalf("Init failed: %v", err)
}
return db, func() { db.Close(); os.RemoveAll(dir) }
}
// TestPointUpdate_NonIndexedField: поиск по email (индекс), обновление name (не индекс).
func TestPointUpdate_NonIndexedField(t *testing.T) {
// db, cleanup := setupPointUpdateDB(t)
db, _ := setupPointUpdateDB(t)
// defer cleanup()
opts := linedb.LineDbAdapterOptions{}
if err := db.Insert(map[string]any{"id": 1, "name": "alice", "email": "alice@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 1 failed: %v", err)
}
if err := db.Insert(map[string]any{"id": 2, "name": "bob", "email": "bob@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 2 failed: %v", err)
}
if err := db.Insert(map[string]any{"id": 3, "name": "sam", "email": "sam@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 2 failed: %v", err)
}
if err := db.Insert(map[string]any{"id": 4, "name": "chuck", "email": "chuck@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 2 failed: %v", err)
}
// Поиск по email (индекс), обновляем name (не индекс)
updated, err := db.Update(
map[string]any{"name": "Bobina"},
"users",
map[string]any{"email": "bob@test.com"},
opts,
)
if err != nil {
t.Fatalf("Update failed: %v", err)
}
if len(updated) != 1 {
t.Fatalf("Expected 1 updated record, got %d", len(updated))
}
m := updated[0].(map[string]any)
if m["name"] != "Bobina" || m["email"] != "bob@test.com" {
t.Errorf("Expected name=Bobina email=bob@test.com, got name=%v email=%v", m["name"], m["email"])
}
// Проверяем ReadByFilter по старому email — должен найти обновлённую запись
found, err := db.ReadByFilter(map[string]any{"email": "bob@test.com"}, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter failed: %v", err)
}
if len(found) != 1 {
t.Fatalf("Expected 1 record by email bob@test.com, got %d", len(found))
}
if m2 := found[0].(map[string]any); m2["name"] != "Bobina" {
t.Errorf("Expected name=bob_renamed, got %v", m2["name"])
}
}
// TestPointUpdate_IndexedField: поиск по email (индекс), обновление email.
func TestPointUpdate_IndexedField(t *testing.T) {
db, cleanup := setupPointUpdateDB(t)
defer cleanup()
opts := linedb.LineDbAdapterOptions{}
if err := db.Insert(map[string]any{"name": "alice", "email": "alice@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 1 failed: %v", err)
}
if err := db.Insert(map[string]any{"name": "bob", "email": "bob@test.com"}, "users", opts); err != nil {
t.Fatalf("Insert 2 failed: %v", err)
}
// Поиск по email, обновляем тот же email
updated, err := db.Update(
map[string]any{"email": "bob_new@test.com"},
"users",
map[string]any{"email": "bob@test.com"},
opts,
)
if err != nil {
t.Fatalf("Update failed: %v", err)
}
if len(updated) != 1 {
t.Fatalf("Expected 1 updated record, got %d", len(updated))
}
m := updated[0].(map[string]any)
if m["email"] != "bob_new@test.com" || m["name"] != "bob" {
t.Errorf("Expected email=bob_new@test.com name=bob, got email=%v name=%v", m["email"], m["name"])
}
// По старому email записей быть не должно
foundOld, err := db.ReadByFilter(map[string]any{"email": "bob@test.com"}, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter old failed: %v", err)
}
if len(foundOld) != 0 {
t.Errorf("Expected 0 records for old email bob@test.com, got %d", len(foundOld))
}
// По новому email — одна запись
foundNew, err := db.ReadByFilter(map[string]any{"email": "bob_new@test.com"}, "users", opts)
if err != nil {
t.Fatalf("ReadByFilter new failed: %v", err)
}
if len(foundNew) != 1 {
t.Fatalf("Expected 1 record for new email bob_new@test.com, got %d", len(foundNew))
}
if m2 := foundNew[0].(map[string]any); m2["name"] != "bob" {
t.Errorf("Expected name=bob, got %v", m2["name"])
}
}