diff --git a/.vscode/launch.json b/.vscode/launch.json index 0ecdb21..45796b7 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -72,6 +72,18 @@ "args": [], "showLog": true, "console": "integratedTerminal" + }, + { + "name": "Debug LineDB Partitions Example", + "type": "go", + "request": "launch", + "mode": "auto", + "program": "${workspaceFolder}/examples/partitions/main.go", + "cwd": "${workspaceFolder}", + "env": {}, + "args": [], + "showLog": true, + "console": "integratedTerminal" } ] } diff --git a/examples/partitions/data/events_A.jsonl b/examples/partitions/data/events_A.jsonl index 207036c..e98642c 100644 --- a/examples/partitions/data/events_A.jsonl +++ b/examples/partitions/data/events_A.jsonl @@ -1,2 +1,2 @@ -{"id":1,"status":"processed","tenant":"A","ts":1773305152,"type":"signup"} -{"id":2,"status":"processed","tenant":"A","ts":1773305152,"type":"purchase"} +{"id":1,"status":"new","tenant":"A","ts":1773311342,"type":"signup"} +{"id":2,"status":"new","tenant":"A","ts":1773311342,"type":"purchase"} diff --git a/examples/partitions/data/events_B.jsonl b/examples/partitions/data/events_B.jsonl index 67159ca..27a448c 100644 --- a/examples/partitions/data/events_B.jsonl +++ b/examples/partitions/data/events_B.jsonl @@ -1 +1 @@ -{"id":3,"status":"new","tenant":"B","ts":1773305152,"type":"signup"} +{"id":3,"status":"new","tenant":"B","ts":1773311342,"type":"signup"} diff --git a/examples/partitions/data/users.jsonl b/examples/partitions/data/users.jsonl index 876fee1..3c6353f 100644 --- a/examples/partitions/data/users.jsonl +++ b/examples/partitions/data/users.jsonl @@ -1,2 +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"} +{"createdAt":"2026-03-12T16:29:02.588642365+06:00","email":"a@example.com","id":1,"name":"Alice"} +{"createdAt":"2026-03-12T16:29:02.58871104+06:00","email":"b@example.com","id":2,"name":"Bob"} diff --git a/examples/partitions/main.go b/examples/partitions/main.go index 5aba4a7..a3832fe 100644 --- a/examples/partitions/main.go +++ b/examples/partitions/main.go @@ -13,12 +13,13 @@ import ( // Пример работы с партициями + индексируемой коллекцией. // // Запуск: -// go run ./examples/partitions/main.go +// +// go run ./examples/partitions/main.go // // Важно: -// - В текущей реализации индексы строятся для "обычных" коллекций. -// - Партиционированные коллекции (партиции) создаются динамически и сейчас не индексируются -// (см. getPartitionAdapter: JSONLFileOptions{CollectionName: partitionName} без IndexedFields). +// - В текущей реализации индексы строятся для "обычных" коллекций. +// - Партиционированные коллекции (партиции) создаются динамически и сейчас не индексируются +// (см. getPartitionAdapter: JSONLFileOptions{CollectionName: partitionName} без IndexedFields). func main() { dbDir := filepath.Join(".", "examples", "partitions", "data") _ = os.RemoveAll(dbDir) @@ -41,6 +42,7 @@ func main() { { CollectionName: "events", AllocSize: 512, + IndexedFields: []string{"id", "type"}, }, }, Partitions: []linedb.PartitionCollection{ @@ -103,13 +105,13 @@ func main() { log.Fatalf("Insert events failed: %v", err) } - tenantA, err := db.ReadByFilter(map[string]any{"tenant": "A"}, "events", linedb.LineDbAdapterOptions{}) + tenantA, err := db.ReadByFilter(map[string]any{"type": "signup", "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{}) + tenantB, err := db.ReadByFilter(map[string]any{"type": "signup", "tenant": "B"}, "events", linedb.LineDbAdapterOptions{}) if err != nil { log.Fatalf("ReadByFilter events tenant B failed: %v", err) } @@ -158,4 +160,3 @@ func mustLen(label string, got []any, want int) { log.Fatalf("%s: expected %d, got %d (%v)", label, want, len(got), got) } } - diff --git a/pkg/linedb/index.go b/pkg/linedb/index.go index ccb1cf1..184bee2 100644 --- a/pkg/linedb/index.go +++ b/pkg/linedb/index.go @@ -6,26 +6,37 @@ import ( "sync" ) +// IndexPosition — позиция записи в индексе (партиция + номер строки). +// Partition "default" — для обычных коллекций без партиционирования. +type IndexPosition struct { + Partition string // имя партиции ("default" для обычных коллекций) + LineIndex int +} + +// DefaultPartition — значение партиции для непартиционированных коллекций. +const DefaultPartition = "default" + // IndexStore — интерфейс хранилища индексов. // Позволяет подключать разные реализации: в памяти, memcached и др. +// Индекс привязан к логической коллекции; для партиционированных хранит (partition, lineIndex). type IndexStore interface { // Lookup ищет позиции записей по полю и значению. // value — строковое представление (см. valueToIndexKey). - // Возвращает срез индексов строк (0-based) или nil, nil, - // если индекс не используется или записей нет. - Lookup(collection, field, value string) ([]int, error) + // Возвращает (partition, lineIndex) — для непартиционированных partition = DefaultPartition. + Lookup(collection, field, value string) ([]IndexPosition, error) - // IndexRecord добавляет одну запись в индекс по номеру строки (для точечных Update). - IndexRecord(collection string, fields []string, record map[string]any, lineIndex int) + // IndexRecord добавляет одну запись в индекс. + // partition — имя партиции (DefaultPartition для обычных коллекций). + IndexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int) - // UnindexRecord удаляет одну запись из индекса по номеру строки. - UnindexRecord(collection string, fields []string, record map[string]any, lineIndex int) + // UnindexRecord удаляет одну запись из индекса. + UnindexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int) - // Rebuild полностью перестраивает индекс коллекции. - // records — полный список записей, каждая позиция в срезе соответствует номеру строки в файле. - Rebuild(collection string, fields []string, records []any) error + // Rebuild перестраивает вклад в индекс для одной партиции (или всей коллекции, если одна партиция). + // partition — имя партиции; records — записи, позиция в срезе = lineIndex. + Rebuild(collection, partition string, fields []string, records []any) error - // Clear очищает индекс коллекции. + // Clear очищает индекс коллекции (все партиции). Clear(collection string) error } @@ -65,14 +76,14 @@ func getFieldValue(record map[string]any, field string) string { // InMemoryIndexStore — реализация IndexStore в памяти (по умолчанию). type InMemoryIndexStore struct { mu sync.RWMutex - // index: collection:field -> value -> список индексов строк (0-based) - index map[string]map[string][]int + // index: collection:field -> value -> []IndexPosition + index map[string]map[string][]IndexPosition } // NewInMemoryIndexStore создаёт новый in-memory индекс. func NewInMemoryIndexStore() *InMemoryIndexStore { return &InMemoryIndexStore{ - index: make(map[string]map[string][]int), + index: make(map[string]map[string][]IndexPosition), } } @@ -82,27 +93,33 @@ func (s *InMemoryIndexStore) indexKey(collection, field string) string { } // IndexRecord добавляет запись в индекс. -func (s *InMemoryIndexStore) IndexRecord(collection string, fields []string, record map[string]any, lineIndex int) { +func (s *InMemoryIndexStore) IndexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int) { if len(fields) == 0 || record == nil { return } + if partition == "" { + partition = DefaultPartition + } 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] = make(map[string][]IndexPosition) } - s.index[key][val] = append(s.index[key][val], lineIndex) + s.index[key][val] = append(s.index[key][val], IndexPosition{Partition: partition, LineIndex: lineIndex}) } } -// UnindexRecord удаляет запись из индекса. -func (s *InMemoryIndexStore) UnindexRecord(collection string, fields []string, record map[string]any, lineIndex int) { +// UnindexRecord удаляет запись из индекса (по partition и lineIndex). +func (s *InMemoryIndexStore) UnindexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int) { if len(fields) == 0 || record == nil { return } + if partition == "" { + partition = DefaultPartition + } s.mu.Lock() defer s.mu.Unlock() for _, field := range fields { @@ -112,23 +129,23 @@ func (s *InMemoryIndexStore) UnindexRecord(collection string, fields []string, r if bucket == nil { continue } - idxs := bucket[val] - newIdxs := make([]int, 0, len(idxs)) - for _, i := range idxs { - if i != lineIndex { - newIdxs = append(newIdxs, i) + positions := bucket[val] + newPos := make([]IndexPosition, 0, len(positions)) + for _, p := range positions { + if p.Partition != partition || p.LineIndex != lineIndex { + newPos = append(newPos, p) } } - if len(newIdxs) == 0 { + if len(newPos) == 0 { delete(bucket, val) } else { - bucket[val] = newIdxs + bucket[val] = newPos } } } -// Lookup возвращает индексы строк по полю и значению. -func (s *InMemoryIndexStore) Lookup(collection, field, value string) ([]int, error) { +// Lookup возвращает позиции (partition, lineIndex) по полю и значению. +func (s *InMemoryIndexStore) Lookup(collection, field, value string) ([]IndexPosition, error) { s.mu.RLock() defer s.mu.RUnlock() @@ -137,27 +154,46 @@ func (s *InMemoryIndexStore) Lookup(collection, field, value string) ([]int, err if bucket == nil { return nil, nil } - indexes := bucket[value] - if len(indexes) == 0 { + positions := bucket[value] + if len(positions) == 0 { return nil, nil } - // Возвращаем копию, чтобы вызывающий код не модифицировал внутренний срез - out := make([]int, len(indexes)) - copy(out, indexes) + out := make([]IndexPosition, len(positions)) + copy(out, positions) return out, nil } -// Rebuild полностью перестраивает индекс коллекции. -func (s *InMemoryIndexStore) Rebuild(collection string, fields []string, records []any) error { +// Rebuild перестраивает вклад партиции в индекс. +func (s *InMemoryIndexStore) Rebuild(collection, partition string, fields []string, records []any) error { + if partition == "" { + partition = DefaultPartition + } s.mu.Lock() defer s.mu.Unlock() - // Очищаем старые записи по этой коллекции + // Удаляем старые позиции этой партиции из индекса for _, field := range fields { key := s.indexKey(collection, field) - delete(s.index, key) + bucket := s.index[key] + if bucket == nil { + continue + } + for val, positions := range bucket { + newPos := make([]IndexPosition, 0, len(positions)) + for _, p := range positions { + if p.Partition != partition { + newPos = append(newPos, p) + } + } + if len(newPos) == 0 { + delete(bucket, val) + } else { + bucket[val] = newPos + } + } } + // Добавляем новые позиции for idx, rec := range records { recMap, ok := rec.(map[string]any) if !ok { @@ -167,9 +203,9 @@ func (s *InMemoryIndexStore) Rebuild(collection string, fields []string, records val := getFieldValue(recMap, field) key := s.indexKey(collection, field) if s.index[key] == nil { - s.index[key] = make(map[string][]int) + s.index[key] = make(map[string][]IndexPosition) } - s.index[key][val] = append(s.index[key][val], idx) + s.index[key][val] = append(s.index[key][val], IndexPosition{Partition: partition, LineIndex: idx}) } } return nil @@ -190,7 +226,7 @@ func (s *InMemoryIndexStore) Clear(collection string) error { } // GetSnapshotForTest возвращает копию индекса для тестов. Доступ только при accessKey == "give_me_cache". -// Ключ — "collection:field", значение — map[string][]int (значение поля -> номера строк). +// Ключ — "collection:field", значение — map[string][]IndexPosition. func (s *InMemoryIndexStore) GetSnapshotForTest(accessKey string) map[string]any { const testAccessKey = "give_me_cache" if accessKey != testAccessKey { @@ -200,11 +236,11 @@ func (s *InMemoryIndexStore) GetSnapshotForTest(accessKey string) map[string]any 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 + cp := make(map[string][]IndexPosition, len(bucket)) + for val, positions := range bucket { + pos2 := make([]IndexPosition, len(positions)) + copy(pos2, positions) + cp[val] = pos2 } out[k] = cp } diff --git a/pkg/linedb/index_memcached.go b/pkg/linedb/index_memcached.go index 2468e8a..7d53754 100644 --- a/pkg/linedb/index_memcached.go +++ b/pkg/linedb/index_memcached.go @@ -58,18 +58,22 @@ func (s *MemcachedIndexStore) memKey(collection, field, value string) string { } // IndexRecord добавляет запись в индекс. -func (s *MemcachedIndexStore) IndexRecord(collection string, fields []string, record map[string]any, lineIndex int) { +func (s *MemcachedIndexStore) IndexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int) { if len(fields) == 0 || record == nil { return } + if partition == "" { + partition = DefaultPartition + } + pos := IndexPosition{Partition: partition, LineIndex: lineIndex} for _, field := range fields { val := getFieldValue(record, field) key := s.memKey(collection, field, val) - var list []int + var list []IndexPosition if data, err := s.client.Get(key); err == nil && len(data) > 0 { _ = json.Unmarshal(data, &list) } - list = append(list, lineIndex) + list = append(list, pos) if data, err := json.Marshal(list); err == nil { _ = s.client.Set(key, data, s.expireSec) } @@ -77,10 +81,13 @@ func (s *MemcachedIndexStore) IndexRecord(collection string, fields []string, re } // UnindexRecord удаляет запись из индекса. -func (s *MemcachedIndexStore) UnindexRecord(collection string, fields []string, record map[string]any, lineIndex int) { +func (s *MemcachedIndexStore) UnindexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int) { if len(fields) == 0 || record == nil { return } + if partition == "" { + partition = DefaultPartition + } for _, field := range fields { val := getFieldValue(record, field) key := s.memKey(collection, field, val) @@ -88,14 +95,14 @@ func (s *MemcachedIndexStore) UnindexRecord(collection string, fields []string, if err != nil || len(data) == 0 { continue } - var list []int + var list []IndexPosition if json.Unmarshal(data, &list) != nil { continue } - var newList []int - for _, i := range list { - if i != lineIndex { - newList = append(newList, i) + var newList []IndexPosition + for _, p := range list { + if p.Partition != partition || p.LineIndex != lineIndex { + newList = append(newList, p) } } if len(newList) == 0 { @@ -106,8 +113,8 @@ func (s *MemcachedIndexStore) UnindexRecord(collection string, fields []string, } } -// Lookup ищет индексы строк по полю и значению. -func (s *MemcachedIndexStore) Lookup(collection, field, value string) ([]int, error) { +// Lookup ищет позиции по полю и значению. +func (s *MemcachedIndexStore) Lookup(collection, field, value string) ([]IndexPosition, error) { key := s.memKey(collection, field, value) data, err := s.client.Get(key) if err != nil { @@ -116,27 +123,38 @@ func (s *MemcachedIndexStore) Lookup(collection, field, value string) ([]int, er if len(data) == 0 { return nil, nil } - var indexes []int - if json.Unmarshal(data, &indexes) != nil { + var positions []IndexPosition + if json.Unmarshal(data, &positions) != nil { return nil, nil } - // Возвращаем копию, чтобы вызывающий код не модифицировал внутренний срез - out := make([]int, len(indexes)) - copy(out, indexes) + out := make([]IndexPosition, len(positions)) + copy(out, positions) return out, nil } -// Rebuild перестраивает индекс коллекции (удаляет старые ключи по префиксу и записывает новые). -func (s *MemcachedIndexStore) Rebuild(collection string, fields []string, records []any) error { +// Rebuild перестраивает вклад партиции в индекс. +func (s *MemcachedIndexStore) Rebuild(collection, partition string, fields []string, records []any) error { + if partition == "" { + partition = DefaultPartition + } s.mu.Lock() defer s.mu.Unlock() - // Memcached не поддерживает перечисление ключей по шаблону. - // Стратегия: перезаписываем все ключи для текущих записей. - // Старые ключи с другими значениями останутся до TTL. - - // Строим value -> []lineIndex для каждого поля - byFieldValue := make(map[string]map[string][]int) + // Memcached не поддерживает удаление по паттерну — добавляем новые ключи. + // Для корректной замены партиции нужно слить с существующими: получить старые списки, + // удалить позиции с Partition==partition, добавить новые. Упрощённо: строим value -> []IndexPosition + // только для этой партиции и перезаписываем. Но тогда старые позиции других партиций + // для того же value будут потеряны. Поэтому нужен merge: Get, фильтровать по partition, добавлять новые. + // Для merge нужен список всех values — его нет. Реалистично: Rebuild вызывается для одной партиции, + // и мы должны merge. Без перечисления ключей в Memcached merge невозможен. + // Вариант: храним в одном ключе все позиции. При Rebuild(partition) нам нужны ВСЕ ключи + // collection:field:value. Их мы не знаем. Прагматичное решение: Rebuild для Memcached + // делает полную перезапись для этого partition — мы не можем удалить старые, просто добавляем. + // Это приведёт к дубликатам. Правильный путь: хранить составной ключ вида collection:field:value, + // и value — единственный. При Rebuild(partition) мы перезаписываем только те ключи, которые + // встречаются в records. Для остальных value старые позиции этой partition останутся. + // Значит нужен merge: для каждого value в records делаем Get, убираем старые Position с этой partition, добавляем новые. + byFieldValue := make(map[string]map[string][]IndexPosition) for idx, rec := range records { recMap, ok := rec.(map[string]any) if !ok { @@ -145,21 +163,31 @@ func (s *MemcachedIndexStore) Rebuild(collection string, fields []string, record for _, field := range fields { val := getFieldValue(recMap, field) if byFieldValue[field] == nil { - byFieldValue[field] = make(map[string][]int) + byFieldValue[field] = make(map[string][]IndexPosition) } - byFieldValue[field][val] = append(byFieldValue[field][val], idx) + byFieldValue[field][val] = append(byFieldValue[field][val], IndexPosition{Partition: partition, LineIndex: idx}) } } for field, valMap := range byFieldValue { - for val, list := range valMap { + for val, newPositions := range valMap { key := s.memKey(collection, field, val) - data, err := json.Marshal(list) - if err != nil { - continue + var list []IndexPosition + if data, err := s.client.Get(key); err == nil && len(data) > 0 { + _ = json.Unmarshal(data, &list) } - if err := s.client.Set(key, data, s.expireSec); err != nil { - return fmt.Errorf("memcached set %s: %w", key, err) + // Удаляем старые позиции этой партиции + filtered := make([]IndexPosition, 0, len(list)) + for _, p := range list { + if p.Partition != partition { + filtered = append(filtered, p) + } + } + list = append(filtered, newPositions...) + if data, err := json.Marshal(list); err == nil { + if err := s.client.Set(key, data, s.expireSec); err != nil { + return fmt.Errorf("memcached set %s: %w", key, err) + } } } } diff --git a/pkg/linedb/line_db.go b/pkg/linedb/line_db.go index c5f1c50..17d0960 100644 --- a/pkg/linedb/line_db.go +++ b/pkg/linedb/line_db.go @@ -27,7 +27,7 @@ type LineDb struct { constructorOptions *LineDbOptions initOptions *LineDbInitOptions indexStore IndexStore - reindexDone chan struct{} // закрывается при Close, останавливает горутину ReindexByTimer + indexRebuildDone chan struct{} // закрывается при Close, останавливает горутину IndexRebuildTimer } // NewLineDb создает новый экземпляр LineDb @@ -133,28 +133,42 @@ func (db *LineDb) Init(force bool, initOptions *LineDbInitOptions) error { } if db.indexStore != nil { for _, opt := range initOptions.Collections { - if len(opt.IndexedFields) == 0 || db.isCollectionPartitioned(opt.CollectionName) { + if len(opt.IndexedFields) == 0 { 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 db.isCollectionPartitioned(opt.CollectionName) { + // Партиции: обходим все существующие партиции (могут быть созданы ранее) + baseName := opt.CollectionName + for name, adapter := range db.adapters { + if !strings.HasPrefix(name, baseName+"_") { + continue + } + records, err := adapter.Read(LineDbAdapterOptions{}) + if err != nil { + continue + } + _ = db.indexStore.Rebuild(opt.CollectionName, name, opt.IndexedFields, records) + } + } else { + adapter := db.adapters[opt.CollectionName] + if adapter != nil { + 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, DefaultPartition, 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) + // Периодический ребилд индексов (отдельная горутина; при тике — полная блокировка db.mutex) + if db.indexStore != nil && initOptions.IndexRebuildTimer > 0 { + db.indexRebuildDone = make(chan struct{}) + interval := time.Duration(initOptions.IndexRebuildTimer) * time.Second + go db.indexRebuildTimerLoop(interval) } return nil @@ -274,6 +288,12 @@ func (db *LineDb) Insert(data any, collectionName string, options LineDbAdapterO // Записываем данные с флагом транзакции writeOptions := LineDbAdapterOptions{InTransaction: true, InternalCall: true} + if db.indexStore != nil { + opts := db.getCollectionOptions(collectionName) + if opts != nil && len(opts.IndexedFields) > 0 && db.isCollectionPartitioned(collectionName) { + writeOptions.DoIndexing = true // индекс строится точечно при Write в партиции + } + } if err := db.Write(resultDataArray, collectionName, writeOptions); err != nil { return fmt.Errorf("failed to write data: %w", err) } @@ -285,7 +305,7 @@ 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) { @@ -293,7 +313,7 @@ func (db *LineDb) Insert(data any, collectionName string, options LineDbAdapterO if exists { allRecords, readErr := adapter.Read(LineDbAdapterOptions{}) if readErr == nil { - _ = db.indexStore.Rebuild(collectionName, opts.IndexedFields, allRecords) + _ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords) } } } @@ -317,15 +337,27 @@ func (db *LineDb) Write(data any, collectionName string, options LineDbAdapterOp // Проверяем партиционирование if db.isCollectionPartitioned(collectionName) { dataArray := db.normalizeDataArray(data) + opts := db.getCollectionOptions(collectionName) for _, item := range dataArray { adapter, err := db.getPartitionAdapter(item, collectionName) if err != nil { return fmt.Errorf("failed to get partition adapter: %w", err) } - + partitionName := adapter.GetCollectionName() + var startLine int + if options.DoIndexing && db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 { + if c, err := adapter.LineCount(); err == nil { + startLine = c + } + } if err := adapter.Write(item, options); err != nil { return fmt.Errorf("failed to write to partition: %w", err) } + if options.DoIndexing && db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 { + if m, err := db.toMap(item); err == nil { + db.indexStore.IndexRecord(collectionName, partitionName, opts.IndexedFields, m, startLine) + } + } } return nil } @@ -354,7 +386,7 @@ func (db *LineDb) Write(data any, collectionName string, options LineDbAdapterOp 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) + db.indexStore.IndexRecord(collectionName, DefaultPartition, opts.IndexedFields, m, startLine+i) } } } @@ -412,7 +444,7 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li // Пробуем точечный Update через индекс (без полного чтения файла) opts := db.getCollectionOptions(collectionName) - if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 { + if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 && !db.isCollectionPartitioned(collectionName) { result, used, err := db.tryIndexUpdate(adapter, filter, dataMap, collectionName, opts.IndexedFields, options) if err != nil { return nil, err @@ -436,7 +468,7 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li if opts != nil && len(opts.IndexedFields) > 0 { allRecords, readErr := adapter.Read(LineDbAdapterOptions{}) if readErr == nil { - _ = db.indexStore.Rebuild(collectionName, opts.IndexedFields, allRecords) + _ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords) } } } @@ -502,7 +534,7 @@ func (db *LineDb) Delete(data any, collectionName string, options LineDbAdapterO if opts != nil && len(opts.IndexedFields) > 0 { allRecords, readErr := adapter.Read(LineDbAdapterOptions{}) if readErr == nil { - _ = db.indexStore.Rebuild(collectionName, opts.IndexedFields, allRecords) + _ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords) } } } @@ -555,7 +587,29 @@ func (db *LineDb) ReadByFilter(filter any, collectionName string, options LineDb // Проверяем партиционирование if db.isCollectionPartitioned(collectionName) { - return db.readByFilterPartitioned(filter, collectionName, options) + if db.indexStore != nil { + hit, result, err := db.tryIndexLookupPartitioned(filter, collectionName, options) + if err != nil || (options.FailOnFailureIndexRead && !hit) { + return nil, fmt.Errorf("index read failed: %w", err) + } + if hit && err == nil { + if db.cacheExternal != nil && !options.InTransaction { + db.cacheExternal.Set(db.generateCacheKey(filter, collectionName), result) + } + return result, nil + } + } + + // fallback to regular read + result, err := db.readByFilterPartitioned(filter, collectionName, options) + if err != nil { + return nil, err + } + // Обновляем кэш + if db.cacheExternal != nil && !options.InTransaction { + db.cacheExternal.Set(db.generateCacheKey(filter, collectionName), result) + } + return result, nil } adapter, exists := db.adapters[collectionName] @@ -565,7 +619,11 @@ func (db *LineDb) ReadByFilter(filter any, collectionName string, options LineDb // Используем индекс, если фильтр — одно поле из IndexedFields if db.indexStore != nil { - if hit, result, err := db.tryIndexLookup(adapter, filter, collectionName, options); hit && err == nil { + hit, result, err := db.tryIndexLookup(adapter, filter, collectionName, DefaultPartition, options) + if err != nil || (options.FailOnFailureIndexRead && !hit) { + return nil, fmt.Errorf("index read failed: %w", err) + } + if hit && err == nil { if db.cacheExternal != nil && !options.InTransaction { db.cacheExternal.Set(db.generateCacheKey(filter, collectionName), result) } @@ -573,6 +631,7 @@ func (db *LineDb) ReadByFilter(filter any, collectionName string, options LineDb } } + // fallback to regular read result, err := adapter.ReadByFilter(filter, options) if err != nil { return nil, err @@ -633,10 +692,10 @@ func (db *LineDb) Close() { db.mutex.Lock() defer db.mutex.Unlock() - // Останавливаем горутину ReindexByTimer - if db.reindexDone != nil { - close(db.reindexDone) - db.reindexDone = nil + // Останавливаем горутину периодического ребилда индексов (IndexRebuildTimer) + if db.indexRebuildDone != nil { + close(db.indexRebuildDone) + db.indexRebuildDone = nil } // Закрываем все адаптеры @@ -656,31 +715,44 @@ func (db *LineDb) Close() { db.partitionFunctions = make(map[string]func(any) string) } -// reindexByTimerLoop выполняет полный ребилд всех индексов с заданным интервалом. -// Работает в отдельной горутине. Останавливается при закрытии db.reindexDone. -func (db *LineDb) reindexByTimerLoop(interval time.Duration) { +// indexRebuildTimerLoop выполняет полный ребилд всех индексов с заданным интервалом (IndexRebuildTimer). +// Работает в отдельной горутине; при каждом тике берёт db.mutex.Lock() на время ребилда. +// Останавливается при закрытии db.indexRebuildDone. +func (db *LineDb) indexRebuildTimerLoop(interval time.Duration) { ticker := time.NewTicker(interval) defer ticker.Stop() for { select { - case <-db.reindexDone: + case <-db.indexRebuildDone: 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) { + if len(opt.IndexedFields) == 0 { continue } - adapter := db.adapters[opt.CollectionName] - if adapter == nil { - continue + if db.isCollectionPartitioned(opt.CollectionName) { + baseName := opt.CollectionName + for name, adapter := range db.adapters { + if !strings.HasPrefix(name, baseName+"_") { + continue + } + records, err := adapter.Read(LineDbAdapterOptions{}) + if err != nil { + continue + } + _ = db.indexStore.Rebuild(opt.CollectionName, name, opt.IndexedFields, records) + } + } else { + adapter := db.adapters[opt.CollectionName] + if adapter != nil { + records, err := adapter.Read(LineDbAdapterOptions{}) + if err == nil { + _ = db.indexStore.Rebuild(opt.CollectionName, DefaultPartition, opt.IndexedFields, records) + } + } } - records, err := adapter.Read(LineDbAdapterOptions{}) - if err != nil { - continue - } - _ = db.indexStore.Rebuild(opt.CollectionName, opt.IndexedFields, records) } } db.mutex.Unlock() @@ -704,20 +776,16 @@ 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) { +// tryIndexLookup пытается выполнить поиск через индекс (для одной партиции/коллекции). +func (db *LineDb) tryIndexLookup(adapter *JSONLFile, filter any, collectionName, partition 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 { + opts := db.getCollectionOptions(collectionName) + if opts == nil || len(opts.IndexedFields) == 0 { return false, nil, nil } - // Находим первое индексируемое поле, которое есть в фильтре var field string var value any for _, idxField := range opts.IndexedFields { @@ -731,11 +799,21 @@ func (db *LineDb) tryIndexLookup(adapter *JSONLFile, filter any, collectionName return false, nil, nil } valStr := valueToIndexKey(value) - indexes, err := db.indexStore.Lookup(collectionName, field, valStr) - if err != nil || indexes == nil || len(indexes) == 0 { + positions, err := db.indexStore.Lookup(collectionName, field, valStr) + if err != nil || len(positions) == 0 { return false, nil, err } - records, err := adapter.ReadByLineIndexes(indexes, options) + // Фильтруем только позиции нужной партиции + var lineIndexes []int + for _, p := range positions { + if p.Partition == partition { + lineIndexes = append(lineIndexes, p.LineIndex) + } + } + if len(lineIndexes) == 0 { + return false, nil, nil + } + records, err := adapter.ReadByLineIndexes(lineIndexes, options) if err != nil { return false, nil, err } @@ -753,6 +831,65 @@ func (db *LineDb) tryIndexLookup(adapter *JSONLFile, filter any, collectionName return true, filtered, nil } +// tryIndexLookupPartitioned — поиск через индекс для партиционированной коллекции. +func (db *LineDb) tryIndexLookupPartitioned(filter any, collectionName string, options LineDbAdapterOptions) (bool, []any, error) { + opts := db.getCollectionOptions(collectionName) + if opts == nil || len(opts.IndexedFields) == 0 { + return false, nil, nil + } + filterMap, ok := filter.(map[string]any) + if !ok || len(filterMap) == 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) + positions, err := db.indexStore.Lookup(collectionName, field, valStr) + if err != nil || len(positions) == 0 { + return false, nil, err + } + // Группируем по партиции + byPartition := make(map[string][]int) + for _, p := range positions { + byPartition[p.Partition] = append(byPartition[p.Partition], p.LineIndex) + } + var allRecords []any + for partitionName, lineIndexes := range byPartition { + adapter := db.adapters[partitionName] + if adapter == nil { + continue + } + recs, err := adapter.ReadByLineIndexes(lineIndexes, options) + if err != nil { + return false, nil, err + } + allRecords = append(allRecords, recs...) + } + if len(allRecords) == 0 { + return false, nil, nil + } + if len(filterMap) == 1 { + return true, allRecords, nil + } + var filtered []any + for _, rec := range allRecords { + 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) @@ -772,10 +909,19 @@ func (db *LineDb) tryIndexUpdate(adapter *JSONLFile, filter any, dataMap map[str return nil, false, nil } valStr := valueToIndexKey(value) - indexes, err := db.indexStore.Lookup(collectionName, field, valStr) - if err != nil || indexes == nil || len(indexes) == 0 { + posList, err := db.indexStore.Lookup(collectionName, field, valStr) + if err != nil || len(posList) == 0 { return nil, false, err } + var indexes []int + for _, p := range posList { + if p.Partition == DefaultPartition { + indexes = append(indexes, p.LineIndex) + } + } + if len(indexes) == 0 { + return nil, false, nil + } opt := options opt.InTransaction = true records, positions, err := adapter.ReadByLineIndexesWithPositions(indexes, opt) @@ -814,12 +960,12 @@ func (db *LineDb) tryIndexUpdate(adapter *JSONLFile, filter any, dataMap map[str } for i, rec := range toUpdate { if m, ok := rec.(map[string]any); ok { - db.indexStore.UnindexRecord(collectionName, indexedFields, m, toUpdatePos[i]) + db.indexStore.UnindexRecord(collectionName, DefaultPartition, 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]) + db.indexStore.IndexRecord(collectionName, DefaultPartition, indexedFields, m, toUpdatePos[i]) } } return updated, true, nil @@ -844,10 +990,19 @@ func (db *LineDb) tryIndexDelete(adapter *JSONLFile, filter any, collectionName return nil, false, nil } valStr := valueToIndexKey(value) - indexes, err := db.indexStore.Lookup(collectionName, field, valStr) - if err != nil || indexes == nil || len(indexes) == 0 { + posList, err := db.indexStore.Lookup(collectionName, field, valStr) + if err != nil || len(posList) == 0 { return nil, false, err } + var indexes []int + for _, p := range posList { + if p.Partition == DefaultPartition { + indexes = append(indexes, p.LineIndex) + } + } + if len(indexes) == 0 { + return nil, false, nil + } opt := options opt.InTransaction = true records, positions, err := adapter.ReadByLineIndexesWithPositions(indexes, opt) @@ -871,7 +1026,7 @@ func (db *LineDb) tryIndexDelete(adapter *JSONLFile, filter any, collectionName } for i, rec := range toDel { if m, ok := rec.(map[string]any); ok { - db.indexStore.UnindexRecord(collectionName, indexedFields, m, toDelPos[i]) + db.indexStore.UnindexRecord(collectionName, DefaultPartition, indexedFields, m, toDelPos[i]) } } return toDel, true, nil @@ -1321,19 +1476,20 @@ func (db *LineDb) normalizeFilter(filter any) (any, error) { } func (db *LineDb) updatePartitioned(data any, collectionName string, filter any, options LineDbAdapterOptions) ([]any, error) { - // Получаем все партиции partitionFiles, err := db.getPartitionFiles(collectionName) if err != nil { return nil, err } + opts := db.getCollectionOptions(collectionName) var allResults []any for _, filename := range partitionFiles { - // Находим адаптер по имени файла var adapter *JSONLFile + var partitionName string for name, adapterFile := range db.collections { if adapterFile == filename { adapter = db.adapters[name] + partitionName = name break } } @@ -1344,6 +1500,12 @@ func (db *LineDb) updatePartitioned(data any, collectionName string, filter any, return nil, err } allResults = append(allResults, results...) + if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 { + records, readErr := adapter.Read(LineDbAdapterOptions{InTransaction: true}) + if readErr == nil { + _ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records) + } + } } } @@ -1351,19 +1513,20 @@ func (db *LineDb) updatePartitioned(data any, collectionName string, filter any, } func (db *LineDb) deletePartitioned(data any, collectionName string, options LineDbAdapterOptions) ([]any, error) { - // Получаем все партиции partitionFiles, err := db.getPartitionFiles(collectionName) if err != nil { return nil, err } + opts := db.getCollectionOptions(collectionName) var allResults []any for _, filename := range partitionFiles { - // Находим адаптер по имени файла var adapter *JSONLFile + var partitionName string for name, adapterFile := range db.collections { if adapterFile == filename { adapter = db.adapters[name] + partitionName = name break } } @@ -1374,6 +1537,12 @@ func (db *LineDb) deletePartitioned(data any, collectionName string, options Lin return nil, err } allResults = append(allResults, results...) + if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 { + records, readErr := adapter.Read(LineDbAdapterOptions{InTransaction: true}) + if readErr == nil { + _ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records) + } + } } } @@ -1672,7 +1841,7 @@ func (db *LineDb) GetCacheForTest(accessKey string) map[string]any { } // GetIndexSnapshotForTest возвращает снимок индекса для тестов (только InMemoryIndexStore). -// Доступ только при accessKey == "give_me_cache". Ключ — "collection:field", значение — map[value][]int (номера строк). +// Доступ только при accessKey == "give_me_cache". Ключ — "collection:field", значение — map[value][]IndexPosition. func (db *LineDb) GetIndexSnapshotForTest(accessKey string) map[string]any { const testAccessKey = "give_me_cache" if accessKey != testAccessKey || db.indexStore == nil { diff --git a/pkg/linedb/types.go b/pkg/linedb/types.go index 3031955..617b6b7 100644 --- a/pkg/linedb/types.go +++ b/pkg/linedb/types.go @@ -52,8 +52,9 @@ type LineDbInitOptions struct { 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 — периодический полный ребилд всех индексов с блокировкой + Partitions []PartitionCollection `json:"partitions,omitempty"` + // IndexRebuildTimer — интервал в секундах; если > 0 — в отдельной горутине периодический полный ребилд всех индексов с блокировкой базы (db.mutex). + IndexRebuildTimer int `json:"indexRebuildTimer,omitempty"` } // EmptyValueMode определяет, что считать "пустым" при проверке required/unique @@ -180,6 +181,9 @@ type LineDbAdapterOptions struct { OptimisticRead bool `json:"optimisticRead,omitempty"` ReturnChain bool `json:"returnChain,omitempty"` DoIndexing bool `json:"doIndexing,omitempty"` // при true — после Write делать точечное индексирование новых записей + // FailOnFailureIndexRead — при true: если ReadByFilter пошёл по индексу и получил ошибку (Lookup / ReadByLineIndexes), + // вернуть её и не делать полный перебор по коллекции. + FailOnFailureIndexRead bool `json:"failOnFailureIndexRead,omitempty"` } // TransactionOptions представляет опции транзакции diff --git a/tests/linedb_index_test.go b/tests/linedb_index_test.go index 7596f85..44c7fa2 100644 --- a/tests/linedb_index_test.go +++ b/tests/linedb_index_test.go @@ -223,19 +223,19 @@ func TestIndexEncodedCollectionCache(t *testing.T) { if !foundInCache { t.Error("Expected to find bob record in raw cache with updated name") } - // Проверяем индекс: по email bob@secret.com должна быть одна строка, по name bob_updated — тоже + // Проверяем индекс: по 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 emailIdx, ok := idxSnapshot["users:email"].(map[string][]linedb.IndexPosition); ok { + if positions, ok := emailIdx["bob@secret.com"]; !ok || len(positions) != 1 { + t.Errorf("Expected index users:email bob@secret.com to have 1 position, 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"]) + if nameIdx, ok := idxSnapshot["users:name"].(map[string][]linedb.IndexPosition); ok { + if positions, ok := nameIdx["bob_updated"]; !ok || len(positions) != 1 { + t.Errorf("Expected index users:name bob_updated to have 1 position, got %v", nameIdx["bob_updated"]) } } @@ -265,14 +265,14 @@ func TestIndexEncodedCollectionCache(t *testing.T) { } // Индекс: 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 emailIdx, ok := idxSnapshot2["users:email"].(map[string][]linedb.IndexPosition); ok { + if positions, has := emailIdx["bob@secret.com"]; has && len(positions) > 0 { + t.Errorf("After delete, index users:email bob@secret.com should be empty, got %v", positions) } } - 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) + if nameIdx, ok := idxSnapshot2["users:name"].(map[string][]linedb.IndexPosition); ok { + if positions, has := nameIdx["bob_updated"]; has && len(positions) > 0 { + t.Errorf("After delete, index users:name bob_updated should be empty, got %v", positions) } } } diff --git a/tests/nonpartitioned/collection_test.go b/tests/nonpartitioned/collection_test.go new file mode 100644 index 0000000..f494386 --- /dev/null +++ b/tests/nonpartitioned/collection_test.go @@ -0,0 +1,218 @@ +package nonpartitioned + +import ( + "os" + "path/filepath" + "testing" + + "linedb/pkg/linedb" +) + +const dbDir = "./data/nonpartitioned" + +func setupDB(t *testing.T) *linedb.LineDb { + t.Helper() + dir := filepath.Join(dbDir, t.Name()) + _ = os.RemoveAll(dir) + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatalf("mkdir: %v", err) + } + + initOptions := &linedb.LineDbInitOptions{ + DBFolder: dir, + Collections: []linedb.JSONLFileOptions{ + { + CollectionName: "users", + AllocSize: 512, + IndexedFields: []string{"id", "email", "name"}, + }, + }, + } + + db := linedb.NewLineDb(&linedb.LineDbOptions{ + IndexStore: linedb.NewInMemoryIndexStore(), + }) + if err := db.Init(true, initOptions); err != nil { + t.Fatalf("Init: %v", err) + } + return db +} + +func TestNonPartitioned_InsertRead(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + // Insert + users := []any{ + map[string]any{"id": 1, "email": "a@test.com", "name": "Alice"}, + map[string]any{"id": 2, "email": "b@test.com", "name": "Bob"}, + map[string]any{"id": 3, "email": "c@test.com", "name": "Charlie"}, + } + if err := db.Insert(users, "users", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + // Read + all, err := db.Read("users", opts) + if err != nil { + t.Fatalf("Read: %v", err) + } + if len(all) != 3 { + t.Errorf("Read: expected 3, got %d", len(all)) + } +} + +func TestNonPartitioned_ReadByFilter_Indexed(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + if err := db.Insert([]any{ + map[string]any{"id": 1, "email": "x@test.com", "name": "X"}, + map[string]any{"id": 2, "email": "y@test.com", "name": "Y"}, + }, "users", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + // Поиск по индексированному полю + found, err := db.ReadByFilter(map[string]any{"email": "y@test.com"}, "users", opts) + if err != nil { + t.Fatalf("ReadByFilter: %v", err) + } + if len(found) != 1 { + t.Fatalf("ReadByFilter email: expected 1, got %d", len(found)) + } + if m, ok := found[0].(map[string]any); !ok || m["name"] != "Y" { + t.Errorf("ReadByFilter: wrong record %v", found[0]) + } + + foundByID, _ := db.ReadByFilter(map[string]any{"id": 1}, "users", opts) + if len(foundByID) != 1 { + t.Fatalf("ReadByFilter id: expected 1, got %d", len(foundByID)) + } +} + +func TestNonPartitioned_Update(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + if err := db.Insert(map[string]any{"id": 1, "email": "old@test.com", "name": "Old"}, "users", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + updated, err := db.Update( + map[string]any{"email": "new@test.com", "name": "New"}, + "users", + map[string]any{"id": 1}, + opts, + ) + if err != nil { + t.Fatalf("Update: %v", err) + } + if len(updated) != 1 { + t.Fatalf("Update: expected 1, got %d", len(updated)) + } + + found, _ := db.ReadByFilter(map[string]any{"email": "new@test.com"}, "users", opts) + if len(found) != 1 { + t.Fatalf("After update ReadByFilter: expected 1, got %d", len(found)) + } + oldFound, _ := db.ReadByFilter(map[string]any{"email": "old@test.com"}, "users", opts) + if len(oldFound) != 0 { + t.Errorf("Old email should not exist after update, got %d", len(oldFound)) + } +} + +func TestNonPartitioned_Delete(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + if err := db.Insert([]any{ + map[string]any{"id": 1, "email": "del@test.com", "name": "Del"}, + map[string]any{"id": 2, "email": "keep@test.com", "name": "Keep"}, + }, "users", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + deleted, err := db.Delete(map[string]any{"email": "del@test.com"}, "users", opts) + if err != nil { + t.Fatalf("Delete: %v", err) + } + if len(deleted) != 1 { + t.Fatalf("Delete: expected 1, got %d", len(deleted)) + } + + all, _ := db.Read("users", opts) + if len(all) != 1 { + t.Fatalf("After delete expected 1, got %d", len(all)) + } +} + +func TestNonPartitioned_WriteWithDoIndexing(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + // Write с DoIndexing + if err := db.Write( + []any{map[string]any{"id": 10, "email": "write@test.com", "name": "Written"}}, + "users", + linedb.LineDbAdapterOptions{DoIndexing: true}, + ); err != nil { + t.Fatalf("Write: %v", err) + } + + found, _ := db.ReadByFilter(map[string]any{"email": "write@test.com"}, "users", opts) + if len(found) != 1 { + t.Fatalf("Write+DoIndexing: expected 1, got %d", len(found)) + } +} + +func TestNonPartitioned_FullCycle(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + // 1. Insert + if err := db.Insert([]any{ + map[string]any{"id": 1, "email": "one@test.com", "name": "One"}, + map[string]any{"id": 2, "email": "two@test.com", "name": "Two"}, + map[string]any{"id": 3, "email": "three@test.com", "name": "Three"}, + }, "users", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + // 2. Read all + all, _ := db.Read("users", opts) + if len(all) != 3 { + t.Fatalf("Read: expected 3, got %d", len(all)) + } + + // 3. Update + _, err := db.Update(map[string]any{"name": "TwoUpdated"}, "users", map[string]any{"id": 2}, opts) + if err != nil { + t.Fatalf("Update: %v", err) + } + + // 4. Delete + _, err = db.Delete(map[string]any{"id": 3}, "users", opts) + if err != nil { + t.Fatalf("Delete: %v", err) + } + + // 5. Verify + all2, _ := db.Read("users", opts) + if len(all2) != 2 { + t.Fatalf("After cycle expected 2, got %d", len(all2)) + } + found, _ := db.ReadByFilter(map[string]any{"email": "two@test.com"}, "users", opts) + if len(found) != 1 { + t.Fatalf("ReadByFilter two: expected 1, got %d", len(found)) + } + if m, ok := found[0].(map[string]any); ok && m["name"] != "TwoUpdated" { + t.Errorf("Update did not apply: name=%v", m["name"]) + } +} diff --git a/tests/nonpartitioned/data/nonpartitioned/TestNonPartitioned_InsertRead/users.jsonl b/tests/nonpartitioned/data/nonpartitioned/TestNonPartitioned_InsertRead/users.jsonl new file mode 100644 index 0000000..a26c930 --- /dev/null +++ b/tests/nonpartitioned/data/nonpartitioned/TestNonPartitioned_InsertRead/users.jsonl @@ -0,0 +1,3 @@ +{"email":"a@test.com","id":1,"name":"Alice"} +{"email":"b@test.com","id":2,"name":"Bob"} +{"email":"c@test.com","id":3,"name":"Charlie"} diff --git a/tests/nonpartitioned/data/nonpartitioned/index_fail/users.jsonl b/tests/nonpartitioned/data/nonpartitioned/index_fail/users.jsonl new file mode 100644 index 0000000..3f6e6d1 --- /dev/null +++ b/tests/nonpartitioned/data/nonpartitioned/index_fail/users.jsonl @@ -0,0 +1 @@ +{"email":"x@test.com","id":1} diff --git a/tests/nonpartitioned/index_fail_test.go b/tests/nonpartitioned/index_fail_test.go new file mode 100644 index 0000000..ce29709 --- /dev/null +++ b/tests/nonpartitioned/index_fail_test.go @@ -0,0 +1,88 @@ +package nonpartitioned + +import ( + "errors" + "os" + "path/filepath" + "strings" + "testing" + + "linedb/pkg/linedb" +) + +// failingLookupStore — IndexStore, у которого Lookup всегда возвращает ошибку (для проверки FailOnFailureIndexRead). +type failingLookupStore struct { + inner linedb.IndexStore +} + +func (f *failingLookupStore) Lookup(collection, field, value string) ([]linedb.IndexPosition, error) { + return nil, errors.New("simulated index lookup failure") +} + +func (f *failingLookupStore) IndexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int) { + f.inner.IndexRecord(collection, partition, fields, record, lineIndex) +} + +func (f *failingLookupStore) UnindexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int) { + f.inner.UnindexRecord(collection, partition, fields, record, lineIndex) +} + +func (f *failingLookupStore) Rebuild(collection, partition string, fields []string, records []any) error { + return f.inner.Rebuild(collection, partition, fields, records) +} + +func (f *failingLookupStore) Clear(collection string) error { + return f.inner.Clear(collection) +} + +func TestReadByFilter_FailOnFailureIndexRead_ReturnsError(t *testing.T) { + dir := filepath.Join(dbDir, "index_fail") + _ = os.RemoveAll(dir) + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatalf("mkdir: %v", err) + } + + initOptions := &linedb.LineDbInitOptions{ + DBFolder: dir, + Collections: []linedb.JSONLFileOptions{ + { + CollectionName: "users", + AllocSize: 512, + IndexedFields: []string{"id", "email"}, + }, + }, + } + + inner := linedb.NewInMemoryIndexStore() + store := &failingLookupStore{inner: inner} + + db := linedb.NewLineDb(&linedb.LineDbOptions{IndexStore: store}) + if err := db.Init(true, initOptions); err != nil { + t.Fatalf("Init: %v", err) + } + defer db.Close() + + if err := db.Insert(map[string]any{"id": 1, "email": "x@test.com"}, "users", linedb.LineDbAdapterOptions{}); err != nil { + t.Fatalf("Insert: %v", err) + } + + // Без опции — ошибка индекса игнорируется, полный перебор находит запись + found, err := db.ReadByFilter(map[string]any{"email": "x@test.com"}, "users", linedb.LineDbAdapterOptions{}) + if err != nil { + t.Fatalf("ReadByFilter without flag: %v", err) + } + if len(found) != 1 { + t.Fatalf("expected 1 via fallback, got %d", len(found)) + } + + // С FailOnFailureIndexRead — возвращается ошибка индекса + _, err = db.ReadByFilter(map[string]any{"email": "x@test.com"}, "users", linedb.LineDbAdapterOptions{ + FailOnFailureIndexRead: true, + }) + if err == nil { + t.Fatal("expected error when FailOnFailureIndexRead and index Lookup fails") + } + if !strings.Contains(err.Error(), "index read failed") || !strings.Contains(err.Error(), "simulated index lookup failure") { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/tests/partitioned/collection_test.go b/tests/partitioned/collection_test.go new file mode 100644 index 0000000..c2160e0 --- /dev/null +++ b/tests/partitioned/collection_test.go @@ -0,0 +1,274 @@ +package partitioned + +import ( + "os" + "path/filepath" + "testing" + "time" + + "linedb/pkg/linedb" +) + +const dbDir = "./data/partitioned" + +func setupDB(t *testing.T) *linedb.LineDb { + t.Helper() + dir := filepath.Join(dbDir, t.Name()) + _ = os.RemoveAll(dir) + if err := os.MkdirAll(dir, 0755); err != nil { + t.Fatalf("mkdir: %v", err) + } + + initOptions := &linedb.LineDbInitOptions{ + DBFolder: dir, + Collections: []linedb.JSONLFileOptions{ + { + CollectionName: "events", + AllocSize: 512, + IndexedFields: []string{"id", "tenant", "type"}, + }, + }, + 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 { + t.Fatalf("Init: %v", err) + } + return db +} + +func TestPartitioned_InsertRead(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + 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()}, + map[string]any{"id": -1, "tenant": "", "type": "signup", "status": "new", "ts": time.Now().Unix()}, + } + if err := db.Insert(events, "events", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + tenantA, err := db.ReadByFilter(map[string]any{"tenant": "A"}, "events", linedb.LineDbAdapterOptions{FailOnFailureIndexRead: true}) + if err != nil { + t.Fatalf("ReadByFilter(nil): %v", err) + } + if len(tenantA) != 2 { + t.Errorf("expected 2 events, got %d", len(tenantA)) + } + + all, err := db.ReadByFilter(nil, "events", linedb.LineDbAdapterOptions{FailOnFailureIndexRead: false}) + if err != nil { + t.Fatalf("ReadByFilter(nil): %v", err) + } + if len(all) != 4 { + t.Errorf("expected 4 events, got %d", len(all)) + } +} + +func TestPartitioned_ReadByFilter_ByTenant(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + if err := db.Insert([]any{ + map[string]any{"id": 1, "tenant": "X", "type": "a"}, + map[string]any{"id": 2, "tenant": "X", "type": "b"}, + map[string]any{"id": 3, "tenant": "Y", "type": "c"}, + }, "events", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + tenantX, err := db.ReadByFilter(map[string]any{"tenant": "X"}, "events", opts) + if err != nil { + t.Fatalf("ReadByFilter tenant X: %v", err) + } + if len(tenantX) != 2 { + t.Fatalf("tenant X: expected 2, got %d", len(tenantX)) + } + + tenantY, _ := db.ReadByFilter(map[string]any{"tenant": "Y"}, "events", opts) + if len(tenantY) != 1 { + t.Fatalf("tenant Y: expected 1, got %d", len(tenantY)) + } +} + +func TestPartitioned_ReadByFilter_Indexed(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + if err := db.Insert([]any{ + map[string]any{"id": 100, "tenant": "T1", "type": "signup"}, + map[string]any{"id": 200, "tenant": "T1", "type": "purchase"}, + }, "events", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + // Поиск по индексированному полю id + found, err := db.ReadByFilter(map[string]any{"id": 100}, "events", opts) + if err != nil { + t.Fatalf("ReadByFilter id: %v", err) + } + if len(found) != 1 { + t.Fatalf("ReadByFilter id 100: expected 1, got %d", len(found)) + } + + // Поиск по type + foundType, err := db.ReadByFilter(map[string]any{"type": "purchase"}, "events", linedb.LineDbAdapterOptions{FailOnFailureIndexRead: true}) + if err != nil { + t.Fatalf("ReadByFilter type: %v", err) + } + if len(foundType) != 1 { + t.Fatalf("ReadByFilter type: expected 1, got %d", len(foundType)) + } +} + +func TestPartitioned_Update(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + if err := db.Insert([]any{ + map[string]any{"id": 1, "tenant": "A", "type": "new", "status": "pending"}, + map[string]any{"id": 2, "tenant": "A", "type": "new", "status": "pending"}, + }, "events", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + updated, err := db.Update( + map[string]any{"status": "processed"}, + "events", + map[string]any{"tenant": "A"}, + opts, + ) + if err != nil { + t.Fatalf("Update: %v", err) + } + if len(updated) != 2 { + t.Fatalf("Update: expected 2, got %d", len(updated)) + } + + processed, _ := db.ReadByFilter(map[string]any{"tenant": "A", "status": "processed"}, "events", opts) + if len(processed) != 2 { + t.Fatalf("After update: expected 2 processed, got %d", len(processed)) + } +} + +func TestPartitioned_Delete(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + if err := db.Insert([]any{ + map[string]any{"id": 1, "tenant": "P", "type": "a"}, + map[string]any{"id": 2, "tenant": "P", "type": "b"}, + map[string]any{"id": 3, "tenant": "Q", "type": "c"}, + }, "events", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + deleted, err := db.Delete(map[string]any{"id": 2}, "events", opts) + if err != nil { + t.Fatalf("Delete: %v", err) + } + if len(deleted) != 1 { + t.Fatalf("Delete: expected 1, got %d", len(deleted)) + } + + all, _ := db.ReadByFilter(nil, "events", opts) + if len(all) != 2 { + t.Fatalf("After delete expected 2, got %d", len(all)) + } +} + +func TestPartitioned_WriteWithDoIndexing(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + if err := db.Write( + []any{map[string]any{"id": 999, "tenant": "W", "type": "direct"}}, + "events", + linedb.LineDbAdapterOptions{DoIndexing: true}, + ); err != nil { + t.Fatalf("Write: %v", err) + } + + found, _ := db.ReadByFilter(map[string]any{"id": 999}, "events", opts) + if len(found) != 1 { + t.Fatalf("Write+DoIndexing: expected 1, got %d", len(found)) + } +} + +func TestPartitioned_FullCycle(t *testing.T) { + db := setupDB(t) + defer db.Close() + opts := linedb.LineDbAdapterOptions{} + + // 1. Insert в разные партиции + if err := db.Insert([]any{ + map[string]any{"id": 1, "tenant": "Alpha", "type": "signup"}, + map[string]any{"id": 2, "tenant": "Alpha", "type": "login"}, + map[string]any{"id": 3, "tenant": "Beta", "type": "signup"}, + }, "events", opts); err != nil { + t.Fatalf("Insert: %v", err) + } + + // 2. Чтение + alpha, _ := db.ReadByFilter(map[string]any{"tenant": "Alpha"}, "events", opts) + if len(alpha) != 2 { + t.Fatalf("Alpha: expected 2, got %d", len(alpha)) + } + + // 3. Update в одной партиции + _, err := db.Update( + map[string]any{"type": "updated"}, + "events", + map[string]any{"id": 2}, + opts, + ) + if err != nil { + t.Fatalf("Update: %v", err) + } + + // 4. Delete + _, err = db.Delete(map[string]any{"id": 3}, "events", opts) + if err != nil { + t.Fatalf("Delete: %v", err) + } + + // 5. Итог + all, _ := db.ReadByFilter(nil, "events", opts) + if len(all) != 2 { + t.Fatalf("After cycle expected 2, got %d", len(all)) + } + updated, _ := db.ReadByFilter(map[string]any{"id": 2}, "events", opts) + if len(updated) != 1 { + t.Fatalf("id 2: expected 1, got %d", len(updated)) + } + if m, ok := updated[0].(map[string]any); ok && m["type"] != "updated" { + t.Errorf("Update did not apply: type=%v", m["type"]) + } +} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_Delete/events.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_Delete/events.jsonl new file mode 100644 index 0000000..e69de29 diff --git a/tests/partitioned/data/partitioned/TestPartitioned_Delete/events_P.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_Delete/events_P.jsonl new file mode 100644 index 0000000..e0bc9e2 --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_Delete/events_P.jsonl @@ -0,0 +1,2 @@ +{"id":1,"tenant":"P","type":"a"} +{"id":2,"tenant":"P","type":"b"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_Delete/events_Q.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_Delete/events_Q.jsonl new file mode 100644 index 0000000..43742ef --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_Delete/events_Q.jsonl @@ -0,0 +1 @@ +{"id":3,"tenant":"Q","type":"c"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_FullCycle/events.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_FullCycle/events.jsonl new file mode 100644 index 0000000..e69de29 diff --git a/tests/partitioned/data/partitioned/TestPartitioned_FullCycle/events_Alpha.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_FullCycle/events_Alpha.jsonl new file mode 100644 index 0000000..2d39e99 --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_FullCycle/events_Alpha.jsonl @@ -0,0 +1,2 @@ +{"id":1,"tenant":"Alpha","type":"signup"} +{"id":2,"tenant":"Alpha","type":"login"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_FullCycle/events_Beta.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_FullCycle/events_Beta.jsonl new file mode 100644 index 0000000..1351dcb --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_FullCycle/events_Beta.jsonl @@ -0,0 +1 @@ +{"id":3,"tenant":"Beta","type":"signup"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events.jsonl new file mode 100644 index 0000000..e69de29 diff --git a/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events_A.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events_A.jsonl new file mode 100644 index 0000000..4a623c8 --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events_A.jsonl @@ -0,0 +1,2 @@ +{"id":1,"status":"new","tenant":"A","ts":1775462238,"type":"signup"} +{"id":2,"status":"new","tenant":"A","ts":1775462238,"type":"purchase"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events_B.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events_B.jsonl new file mode 100644 index 0000000..5e18ea7 --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events_B.jsonl @@ -0,0 +1 @@ +{"id":3,"status":"new","tenant":"B","ts":1775462238,"type":"signup"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events_unknown.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events_unknown.jsonl new file mode 100644 index 0000000..e91d2c9 --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_InsertRead/events_unknown.jsonl @@ -0,0 +1 @@ +{"id":4,"status":"new","tenant":"","ts":1775462238,"type":"signup"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_ByTenant/events.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_ByTenant/events.jsonl new file mode 100644 index 0000000..e69de29 diff --git a/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_ByTenant/events_X.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_ByTenant/events_X.jsonl new file mode 100644 index 0000000..da5b48f --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_ByTenant/events_X.jsonl @@ -0,0 +1,2 @@ +{"id":1,"tenant":"X","type":"a"} +{"id":2,"tenant":"X","type":"b"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_ByTenant/events_Y.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_ByTenant/events_Y.jsonl new file mode 100644 index 0000000..b71cd1c --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_ByTenant/events_Y.jsonl @@ -0,0 +1 @@ +{"id":3,"tenant":"Y","type":"c"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_Indexed/events.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_Indexed/events.jsonl new file mode 100644 index 0000000..e69de29 diff --git a/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_Indexed/events_T1.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_Indexed/events_T1.jsonl new file mode 100644 index 0000000..7598589 --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_ReadByFilter_Indexed/events_T1.jsonl @@ -0,0 +1,2 @@ +{"id":100,"tenant":"T1","type":"signup"} +{"id":200,"tenant":"T1","type":"purchase"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_Update/events.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_Update/events.jsonl new file mode 100644 index 0000000..e69de29 diff --git a/tests/partitioned/data/partitioned/TestPartitioned_Update/events_A.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_Update/events_A.jsonl new file mode 100644 index 0000000..72271bd --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_Update/events_A.jsonl @@ -0,0 +1,2 @@ +{"id":1,"status":"processed","tenant":"A","type":"new"} +{"id":2,"status":"processed","tenant":"A","type":"new"} diff --git a/tests/partitioned/data/partitioned/TestPartitioned_WriteWithDoIndexing/events.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_WriteWithDoIndexing/events.jsonl new file mode 100644 index 0000000..e69de29 diff --git a/tests/partitioned/data/partitioned/TestPartitioned_WriteWithDoIndexing/events_W.jsonl b/tests/partitioned/data/partitioned/TestPartitioned_WriteWithDoIndexing/events_W.jsonl new file mode 100644 index 0000000..42aefd0 --- /dev/null +++ b/tests/partitioned/data/partitioned/TestPartitioned_WriteWithDoIndexing/events_W.jsonl @@ -0,0 +1 @@ +{"id":999,"tenant":"W","type":"direct"} diff --git a/tests/run_all.sh b/tests/run_all.sh new file mode 100755 index 0000000..f5fcd49 --- /dev/null +++ b/tests/run_all.sh @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +# Запуск всех тестов коллекций (партиционированных и непартиционированных) +set -e +cd "$(dirname "$0")/.." +echo "=== Непартиционированные ===" +go test -v ./tests/nonpartitioned/... -count=1 +echo "" +echo "=== Партиционированные ===" +go test -v ./tests/partitioned/... -count=1 +echo "" +echo "=== Все тесты коллекций пройдены ===" diff --git a/tests/run_nonpartitioned.sh b/tests/run_nonpartitioned.sh new file mode 100755 index 0000000..e88dd59 --- /dev/null +++ b/tests/run_nonpartitioned.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +# Запуск тестов для непартиционированных коллекций +set -e +cd "$(dirname "$0")/.." +echo "=== Тесты непартиционированных коллекций ===" +go test -v ./tests/nonpartitioned/... -count=1 +echo "=== OK ===" diff --git a/tests/run_partitioned.sh b/tests/run_partitioned.sh new file mode 100755 index 0000000..05c6e27 --- /dev/null +++ b/tests/run_partitioned.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +# Запуск тестов для партиционированных коллекций +set -e +cd "$(dirname "$0")/.." +echo "=== Тесты партиционированных коллекций ===" +go test -v ./tests/partitioned/... -count=1 +echo "=== OK ==="