4 Commits
v1.1.5 ... main

6 changed files with 437 additions and 182 deletions

View File

@@ -1,6 +1,7 @@
package linedb package linedb
import ( import (
"strings"
"sync" "sync"
"time" "time"
) )
@@ -94,6 +95,37 @@ func (c *RecordCache) Clear() {
c.cache = make(map[string]*CacheEntry) c.cache = make(map[string]*CacheEntry)
} }
// ClearCollection удаляет записи кэша, относящиеся к коллекции collectionName:
// ключи ReadByFilter ("collection:filter" и "collection_part:filter") и SetByRecord ("id:collection", "id:collection_part").
func (c *RecordCache) ClearCollection(collectionName string) {
if collectionName == "" {
c.Clear()
return
}
c.mutex.Lock()
defer c.mutex.Unlock()
for key := range c.cache {
if cacheKeyBelongsToCollection(key, collectionName) {
delete(c.cache, key)
}
}
}
func cacheKeyBelongsToCollection(key, collectionName string) bool {
if strings.HasPrefix(key, collectionName+":") {
return true
}
if strings.HasPrefix(key, collectionName+"_") {
return true
}
idx := strings.LastIndex(key, ":")
if idx < 0 {
return false
}
suf := key[idx+1:]
return suf == collectionName || strings.HasPrefix(suf, collectionName+"_")
}
// ClearEntriesContainingIDs удаляет из кэша только те записи, в данных которых // ClearEntriesContainingIDs удаляет из кэша только те записи, в данных которых
// встречается хотя бы один из переданных id. Если ids пуст — ничего не делает. // встречается хотя бы один из переданных id. Если ids пуст — ничего не делает.
func (c *RecordCache) ClearEntriesContainingIDs(ids []any) { func (c *RecordCache) ClearEntriesContainingIDs(ids []any) {

View File

@@ -33,8 +33,9 @@ type IndexStore interface {
UnindexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int) UnindexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int)
// Rebuild перестраивает вклад в индекс для одной партиции (или всей коллекции, если одна партиция). // Rebuild перестраивает вклад в индекс для одной партиции (или всей коллекции, если одна партиция).
// partition — имя партиции; records — записи, позиция в срезе = lineIndex. // lineIndexes[i] — 0-based номер физической строки в файле для records[i] (после пустых/удалённых строк).
Rebuild(collection, partition string, fields []string, records []any) error // Если lineIndexes == nil или len != len(records), используются плотные индексы 0..len-1 (устаревшее поведение).
Rebuild(collection, partition string, fields []string, records []any, lineIndexes []int) error
// Clear очищает индекс коллекции (все партиции). // Clear очищает индекс коллекции (все партиции).
Clear(collection string) error Clear(collection string) error
@@ -164,7 +165,7 @@ func (s *InMemoryIndexStore) Lookup(collection, field, value string) ([]IndexPos
} }
// Rebuild перестраивает вклад партиции в индекс. // Rebuild перестраивает вклад партиции в индекс.
func (s *InMemoryIndexStore) Rebuild(collection, partition string, fields []string, records []any) error { func (s *InMemoryIndexStore) Rebuild(collection, partition string, fields []string, records []any, lineIndexes []int) error {
if partition == "" { if partition == "" {
partition = DefaultPartition partition = DefaultPartition
} }
@@ -194,7 +195,11 @@ func (s *InMemoryIndexStore) Rebuild(collection, partition string, fields []stri
} }
// Добавляем новые позиции // Добавляем новые позиции
for idx, rec := range records { for i, rec := range records {
lineIdx := i
if lineIndexes != nil && i < len(lineIndexes) {
lineIdx = lineIndexes[i]
}
recMap, ok := rec.(map[string]any) recMap, ok := rec.(map[string]any)
if !ok { if !ok {
continue continue
@@ -205,7 +210,7 @@ func (s *InMemoryIndexStore) Rebuild(collection, partition string, fields []stri
if s.index[key] == nil { if s.index[key] == nil {
s.index[key] = make(map[string][]IndexPosition) s.index[key] = make(map[string][]IndexPosition)
} }
s.index[key][val] = append(s.index[key][val], IndexPosition{Partition: partition, LineIndex: idx}) s.index[key][val] = append(s.index[key][val], IndexPosition{Partition: partition, LineIndex: lineIdx})
} }
} }
return nil return nil

View File

@@ -133,7 +133,7 @@ func (s *MemcachedIndexStore) Lookup(collection, field, value string) ([]IndexPo
} }
// Rebuild перестраивает вклад партиции в индекс. // Rebuild перестраивает вклад партиции в индекс.
func (s *MemcachedIndexStore) Rebuild(collection, partition string, fields []string, records []any) error { func (s *MemcachedIndexStore) Rebuild(collection, partition string, fields []string, records []any, lineIndexes []int) error {
if partition == "" { if partition == "" {
partition = DefaultPartition partition = DefaultPartition
} }
@@ -155,7 +155,11 @@ func (s *MemcachedIndexStore) Rebuild(collection, partition string, fields []str
// встречаются в records. Для остальных value старые позиции этой partition останутся. // встречаются в records. Для остальных value старые позиции этой partition останутся.
// Значит нужен merge: для каждого value в records делаем Get, убираем старые Position с этой partition, добавляем новые. // Значит нужен merge: для каждого value в records делаем Get, убираем старые Position с этой partition, добавляем новые.
byFieldValue := make(map[string]map[string][]IndexPosition) byFieldValue := make(map[string]map[string][]IndexPosition)
for idx, rec := range records { for i, rec := range records {
lineIdx := i
if lineIndexes != nil && i < len(lineIndexes) {
lineIdx = lineIndexes[i]
}
recMap, ok := rec.(map[string]any) recMap, ok := rec.(map[string]any)
if !ok { if !ok {
continue continue
@@ -165,7 +169,7 @@ func (s *MemcachedIndexStore) Rebuild(collection, partition string, fields []str
if byFieldValue[field] == nil { if byFieldValue[field] == nil {
byFieldValue[field] = make(map[string][]IndexPosition) byFieldValue[field] = make(map[string][]IndexPosition)
} }
byFieldValue[field][val] = append(byFieldValue[field][val], IndexPosition{Partition: partition, LineIndex: idx}) byFieldValue[field][val] = append(byFieldValue[field][val], IndexPosition{Partition: partition, LineIndex: lineIdx})
} }
} }

View File

@@ -402,6 +402,133 @@ func (j *JSONLFile) Read(options LineDbAdapterOptions) ([]any, error) {
return records, scanner.Err() return records, scanner.Err()
} }
// ReadWithPhysicalLineIndexes как Read, но для каждой записи возвращает 0-based индекс строки (слота),
// в том же смысле, что ReadByLineIndexes/WriteAtLineIndexes: смещение в файле = lineIndex * allocSize.
// Пустые слоты (пробелы после удаления и т.п.) пропускаются; индекс — номер слота, а не порядковый номер записи.
func (j *JSONLFile) ReadWithPhysicalLineIndexes(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 j.allocSize <= 0 {
return nil, nil, fmt.Errorf("invalid allocSize")
}
file, err := os.Open(j.filename)
if err != nil {
return nil, nil, fmt.Errorf("failed to open file: %w", err)
}
defer file.Close()
info, err := file.Stat()
if err != nil {
return nil, nil, fmt.Errorf("stat file: %w", err)
}
nSlots := int(info.Size()) / j.allocSize
buf := make([]byte, j.allocSize)
var records []any
var lineIndexes []int
for slot := 0; slot < nSlots; slot++ {
n, err := io.ReadFull(file, buf)
if err == io.EOF || err == io.ErrUnexpectedEOF {
break
}
if err != nil {
return nil, nil, fmt.Errorf("read slot %d: %w", slot, err)
}
if n != j.allocSize {
break
}
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)
lineIndexes = append(lineIndexes, slot)
}
return records, lineIndexes, nil
}
// HasBlankSlots возвращает true, если в файле есть хотя бы один пустой слот
// (пробелы + перевод строки — типичный след точечного удаления BlankLinesAtPositions).
func (j *JSONLFile) HasBlankSlots(options LineDbAdapterOptions) (bool, error) {
if !options.InTransaction {
j.mutex.RLock()
defer j.mutex.RUnlock()
}
if !j.initialized {
return false, fmt.Errorf("file not initialized")
}
if j.allocSize <= 0 {
return false, fmt.Errorf("invalid allocSize")
}
file, err := os.Open(j.filename)
if err != nil {
return false, fmt.Errorf("failed to open file: %w", err)
}
defer file.Close()
info, err := file.Stat()
if err != nil {
return false, fmt.Errorf("stat file: %w", err)
}
nSlots := int(info.Size()) / j.allocSize
buf := make([]byte, j.allocSize)
for slot := 0; slot < nSlots; slot++ {
n, err := io.ReadFull(file, buf)
if err == io.EOF || err == io.ErrUnexpectedEOF {
break
}
if err != nil {
return false, fmt.Errorf("read slot %d: %w", slot, err)
}
if n != j.allocSize {
break
}
line := string(buf[:n])
line = strings.TrimRight(line, "\n")
line = strings.TrimRight(line, " ")
if strings.TrimSpace(line) == "" {
return true, nil
}
}
return false, nil
}
// ReadByLineIndexes читает записи по номерам строк (0-based) с использованием random access. // ReadByLineIndexes читает записи по номерам строк (0-based) с использованием random access.
// Ожидается, что файл нормализован и каждая строка имеет длину allocSize байт (включая \n), // Ожидается, что файл нормализован и каждая строка имеет длину allocSize байт (включая \n),
// как это обеспечивает Init/normalizeExistingFile/rewriteFile. // как это обеспечивает Init/normalizeExistingFile/rewriteFile.
@@ -662,6 +789,86 @@ func (j *JSONLFile) BlankLinesAtPositions(lineIndexes []int, options LineDbAdapt
return nil return nil
} }
// CompactFile перезаписывает файл, копируя подряд только непустые слоты (сырые байты без JSON).
// Пустой слот: после обрезки \n и пробелов строка пустая — как после BlankLinesAtPositions.
// Один буфер на весь проход; без раскодирования и без удержания всех записей в памяти.
func (j *JSONLFile) CompactFile(options LineDbAdapterOptions) error {
if !options.InTransaction {
j.mutex.Lock()
defer j.mutex.Unlock()
}
if !j.initialized {
return fmt.Errorf("file not initialized")
}
if j.allocSize <= 0 {
return fmt.Errorf("invalid allocSize")
}
info, err := os.Stat(j.filename)
if err != nil {
return fmt.Errorf("compact stat: %w", err)
}
nSlots := int(info.Size()) / j.allocSize
tempFile := j.filename + ".tmp"
dst, err := os.Create(tempFile)
if err != nil {
return fmt.Errorf("compact create temp: %w", err)
}
src, err := os.Open(j.filename)
if err != nil {
dst.Close()
_ = os.Remove(tempFile)
return fmt.Errorf("compact open source: %w", err)
}
buf := make([]byte, j.allocSize)
compactErr := func() error {
defer src.Close()
for slot := 0; slot < nSlots; slot++ {
n, err := io.ReadFull(src, buf)
if err == io.EOF || err == io.ErrUnexpectedEOF {
break
}
if err != nil {
return fmt.Errorf("compact read slot %d: %w", slot, err)
}
if n != j.allocSize {
break
}
slot := buf[:n]
t := bytes.TrimRight(bytes.TrimRight(slot, "\n"), " ")
if len(bytes.TrimSpace(t)) == 0 {
continue
}
if _, err := dst.Write(slot); err != nil {
return fmt.Errorf("compact write: %w", err)
}
}
return nil
}()
if err := dst.Close(); err != nil {
_ = os.Remove(tempFile)
if compactErr != nil {
return compactErr
}
return fmt.Errorf("compact close temp: %w", err)
}
if compactErr != nil {
_ = os.Remove(tempFile)
return compactErr
}
if err := os.Rename(tempFile, j.filename); err != nil {
_ = os.Remove(tempFile)
return fmt.Errorf("compact rename: %w", err)
}
return nil
}
// LineCount возвращает число строк в файле (fileSize / allocSize). // LineCount возвращает число строк в файле (fileSize / allocSize).
// Используется для точечного индексирования после Write. // Используется для точечного индексирования после Write.
func (j *JSONLFile) LineCount() (int, error) { func (j *JSONLFile) LineCount() (int, error) {

View File

@@ -4,6 +4,7 @@ import (
"fmt" "fmt"
"os" "os"
"path/filepath" "path/filepath"
"sort"
"strconv" "strconv"
"strings" "strings"
"sync" "sync"
@@ -137,26 +138,24 @@ func (db *LineDb) Init(force bool, initOptions *LineDbInitOptions) error {
continue continue
} }
if db.isCollectionPartitioned(opt.CollectionName) { if db.isCollectionPartitioned(opt.CollectionName) {
// Партиции: обходим все существующие партиции (могут быть созданы ранее) // Основной файл + все base_* (пустой partition id → основной файл)
baseName := opt.CollectionName baseName := opt.CollectionName
for name, adapter := range db.adapters { for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
if !strings.HasPrefix(name, baseName+"_") { name := adapter.GetCollectionName()
continue records, lineIdx, err := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
}
records, err := adapter.Read(LineDbAdapterOptions{})
if err != nil { if err != nil {
continue continue
} }
_ = db.indexStore.Rebuild(opt.CollectionName, name, opt.IndexedFields, records) _ = db.indexStore.Rebuild(opt.CollectionName, name, opt.IndexedFields, records, lineIdx)
} }
} else { } else {
adapter := db.adapters[opt.CollectionName] adapter := db.adapters[opt.CollectionName]
if adapter != nil { if adapter != nil {
records, err := adapter.Read(LineDbAdapterOptions{}) records, lineIdx, err := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
if err != nil { if err != nil {
return fmt.Errorf("failed to read collection %s for index rebuild: %w", opt.CollectionName, err) 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 { if err := db.indexStore.Rebuild(opt.CollectionName, DefaultPartition, opt.IndexedFields, records, lineIdx); err != nil {
return fmt.Errorf("failed to rebuild index for %s: %w", opt.CollectionName, err) return fmt.Errorf("failed to rebuild index for %s: %w", opt.CollectionName, err)
} }
} }
@@ -249,7 +248,9 @@ func (db *LineDb) seedLastIDPartitioned(dbFolder, baseName string) error {
return nil return nil
} }
// Read читает все записи из коллекции // Read читает все записи из коллекции.
// Для партиционированной логической коллекции (имя без суффикса _*) читает основной файл
// и все файлы партиций base_* подряд. Запрос по имени одной партиции (base_part) читает только её файл.
func (db *LineDb) Read(collectionName string, options LineDbAdapterOptions) ([]any, error) { func (db *LineDb) Read(collectionName string, options LineDbAdapterOptions) ([]any, error) {
db.mutex.RLock() db.mutex.RLock()
defer db.mutex.RUnlock() defer db.mutex.RUnlock()
@@ -258,6 +259,23 @@ func (db *LineDb) Read(collectionName string, options LineDbAdapterOptions) ([]a
collectionName = db.getFirstCollection() collectionName = db.getFirstCollection()
} }
baseName := db.getBaseCollectionName(collectionName)
if db.isCollectionPartitioned(baseName) && collectionName == baseName {
adapters := db.partitionStorageAdaptersOrdered(baseName)
if len(adapters) == 0 {
return nil, fmt.Errorf("collection %s not found", collectionName)
}
var all []any
for _, a := range adapters {
recs, err := a.Read(options)
if err != nil {
return nil, err
}
all = append(all, recs...)
}
return all, nil
}
adapter, exists := db.adapters[collectionName] adapter, exists := db.adapters[collectionName]
if !exists { if !exists {
return nil, fmt.Errorf("collection %s not found", collectionName) return nil, fmt.Errorf("collection %s not found", collectionName)
@@ -365,34 +383,16 @@ func (db *LineDb) Insert(data any, collectionName string, options LineDbAdapterO
writeOptions := LineDbAdapterOptions{InTransaction: true, InternalCall: true} writeOptions := LineDbAdapterOptions{InTransaction: true, InternalCall: true}
if db.indexStore != nil { if db.indexStore != nil {
opts := db.getCollectionOptions(collectionName) opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 && db.isCollectionPartitioned(collectionName) { if opts != nil && len(opts.IndexedFields) > 0 {
writeOptions.DoIndexing = true // индекс строится точечно при Write в партиции // Точечное IndexRecord в Write (LineCount → стартовая строка для новых записей)
writeOptions.DoIndexing = true
} }
} }
if err := db.Write(resultDataArray, collectionName, writeOptions); err != nil { if err := db.Write(resultDataArray, collectionName, writeOptions); err != nil {
return fmt.Errorf("failed to write data: %w", err) return fmt.Errorf("failed to write data: %w", err)
} }
// Обновляем кэш db.clearCacheForCollection(collectionName)
if db.cacheExternal != nil {
for _, item := range resultDataArray {
db.cacheExternal.UpdateCacheAfterInsert(item, collectionName)
}
}
// Обновляем индекс (полная пересборка по коллекции) — только для непартиционированных
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, DefaultPartition, opts.IndexedFields, allRecords)
}
}
}
}
return nil return nil
} }
@@ -520,15 +520,12 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li
// Пробуем точечный Update через индекс (без полного чтения файла) // Пробуем точечный Update через индекс (без полного чтения файла)
opts := db.getCollectionOptions(collectionName) opts := db.getCollectionOptions(collectionName)
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 && !db.isCollectionPartitioned(collectionName) { 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) result, used, err := db.tryIndexUpdate(adapter, filter, dataMap, collectionName, DefaultPartition, opts.IndexedFields, options)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if used { if used {
if db.cacheExternal != nil { db.clearCacheForCollection(collectionName)
ids := extractIDsFromRecords(result)
db.cacheExternal.ClearEntriesContainingIDs(ids)
}
return result, nil return result, nil
} }
} }
@@ -541,16 +538,13 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li
if db.indexStore != nil { if db.indexStore != nil {
opts := db.getCollectionOptions(collectionName) opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 { if opts != nil && len(opts.IndexedFields) > 0 {
allRecords, readErr := adapter.Read(LineDbAdapterOptions{}) allRecords, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
if readErr == nil { if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords) _ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords, lineIdx)
} }
} }
} }
if db.cacheExternal != nil { db.clearCacheForCollection(collectionName)
ids := extractIDsFromRecords(result)
db.cacheExternal.ClearEntriesContainingIDs(ids)
}
return result, nil return result, nil
} }
@@ -586,15 +580,12 @@ func (db *LineDb) Delete(data any, collectionName string, options LineDbAdapterO
opts := db.getCollectionOptions(collectionName) opts := db.getCollectionOptions(collectionName)
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 { if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
result, used, err := db.tryIndexDelete(adapter, data, collectionName, opts.IndexedFields, options) result, used, err := db.tryIndexDelete(adapter, data, collectionName, DefaultPartition, opts.IndexedFields, options)
if err != nil { if err != nil {
return nil, err return nil, err
} }
if used { if used {
if db.cacheExternal != nil { db.clearCacheForCollection(collectionName)
ids := extractIDsFromRecords(result)
db.cacheExternal.ClearEntriesContainingIDs(ids)
}
return result, nil return result, nil
} }
} }
@@ -607,16 +598,13 @@ func (db *LineDb) Delete(data any, collectionName string, options LineDbAdapterO
if db.indexStore != nil { if db.indexStore != nil {
opts := db.getCollectionOptions(collectionName) opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 { if opts != nil && len(opts.IndexedFields) > 0 {
allRecords, readErr := adapter.Read(LineDbAdapterOptions{}) allRecords, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
if readErr == nil { if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords) _ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords, lineIdx)
} }
} }
} }
if db.cacheExternal != nil { db.clearCacheForCollection(collectionName)
ids := extractIDsFromRecords(result)
db.cacheExternal.ClearEntriesContainingIDs(ids)
}
return result, nil return result, nil
} }
@@ -753,15 +741,21 @@ func (db *LineDb) ClearCache(collectionName string, options LineDbAdapterOptions
if collectionName == "" { if collectionName == "" {
db.cacheExternal.Clear() db.cacheExternal.Clear()
} else { } else {
// Очищаем только записи для конкретной коллекции db.cacheExternal.ClearCollection(collectionName)
// Это упрощенная реализация
db.cacheExternal.Clear()
} }
} }
return nil return nil
} }
// clearCacheForCollection сбрасывает кэш запросов по логической коллекции (включая ключи партиций base_*).
func (db *LineDb) clearCacheForCollection(collectionName string) {
if db.cacheExternal == nil || collectionName == "" {
return
}
db.cacheExternal.ClearCollection(collectionName)
}
// Close закрывает базу данных // Close закрывает базу данных
func (db *LineDb) Close() { func (db *LineDb) Close() {
db.mutex.Lock() db.mutex.Lock()
@@ -792,6 +786,8 @@ func (db *LineDb) Close() {
// indexRebuildTimerLoop выполняет полный ребилд всех индексов с заданным интервалом (IndexRebuildTimer). // indexRebuildTimerLoop выполняет полный ребилд всех индексов с заданным интервалом (IndexRebuildTimer).
// Работает в отдельной горутине; при каждом тике берёт db.mutex.Lock() на время ребилда. // Работает в отдельной горутине; при каждом тике берёт db.mutex.Lock() на время ребилда.
// Перед ребилдом коллекции: если в файле есть пустые слоты (след точечного удаления), вызывается
// CompactFile — перезапись без «дыр», затем индексирование по сжатому файлу.
// Останавливается при закрытии db.indexRebuildDone. // Останавливается при закрытии db.indexRebuildDone.
func (db *LineDb) indexRebuildTimerLoop(interval time.Duration) { func (db *LineDb) indexRebuildTimerLoop(interval time.Duration) {
ticker := time.NewTicker(interval) ticker := time.NewTicker(interval)
@@ -809,22 +805,31 @@ func (db *LineDb) indexRebuildTimerLoop(interval time.Duration) {
} }
if db.isCollectionPartitioned(opt.CollectionName) { if db.isCollectionPartitioned(opt.CollectionName) {
baseName := opt.CollectionName baseName := opt.CollectionName
for name, adapter := range db.adapters { for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
if !strings.HasPrefix(name, baseName+"_") { name := adapter.GetCollectionName()
continue dirty, err := adapter.HasBlankSlots(LineDbAdapterOptions{})
}
records, err := adapter.Read(LineDbAdapterOptions{})
if err != nil { if err != nil {
continue continue
} }
_ = db.indexStore.Rebuild(opt.CollectionName, name, opt.IndexedFields, records) if dirty {
_ = adapter.CompactFile(LineDbAdapterOptions{})
}
records, lineIdx, err := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
if err != nil {
continue
}
_ = db.indexStore.Rebuild(opt.CollectionName, name, opt.IndexedFields, records, lineIdx)
} }
} else { } else {
adapter := db.adapters[opt.CollectionName] adapter := db.adapters[opt.CollectionName]
if adapter != nil { if adapter != nil {
records, err := adapter.Read(LineDbAdapterOptions{}) dirty, err := adapter.HasBlankSlots(LineDbAdapterOptions{})
if err == nil && dirty {
_ = adapter.CompactFile(LineDbAdapterOptions{})
}
records, lineIdx, err := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
if err == nil { if err == nil {
_ = db.indexStore.Rebuild(opt.CollectionName, DefaultPartition, opt.IndexedFields, records) _ = db.indexStore.Rebuild(opt.CollectionName, DefaultPartition, opt.IndexedFields, records, lineIdx)
} }
} }
} }
@@ -965,8 +970,23 @@ func (db *LineDb) tryIndexLookupPartitioned(filter any, collectionName string, o
return true, filtered, nil return true, filtered, nil
} }
// indexedFieldKeysEqual возвращает true, если ключи индекса по всем indexedFields совпадают
// (та же семантика, что getFieldValue/valueToIndexKey в индексе).
func indexedFieldKeysEqual(oldRec, newRec map[string]any, indexedFields []string) bool {
for _, f := range indexedFields {
if getFieldValue(oldRec, f) != getFieldValue(newRec, f) {
return false
}
}
return true
}
// tryIndexUpdate выполняет точечное обновление через индекс. Возвращает (result, used, err). // 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) { // partition — имя партиции для IndexStore (для непартиционированных коллекций: DefaultPartition).
func (db *LineDb) tryIndexUpdate(adapter *JSONLFile, filter any, dataMap map[string]any, collectionName string, partition string, indexedFields []string, options LineDbAdapterOptions) ([]any, bool, error) {
if partition == "" {
partition = DefaultPartition
}
filterMap, ok := filter.(map[string]any) filterMap, ok := filter.(map[string]any)
if !ok || len(filterMap) == 0 { if !ok || len(filterMap) == 0 {
return nil, false, nil return nil, false, nil
@@ -990,7 +1010,7 @@ func (db *LineDb) tryIndexUpdate(adapter *JSONLFile, filter any, dataMap map[str
} }
var indexes []int var indexes []int
for _, p := range posList { for _, p := range posList {
if p.Partition == DefaultPartition { if p.Partition == partition {
indexes = append(indexes, p.LineIndex) indexes = append(indexes, p.LineIndex)
} }
} }
@@ -1033,21 +1053,26 @@ func (db *LineDb) tryIndexUpdate(adapter *JSONLFile, filter any, dataMap map[str
if err := adapter.WriteAtLineIndexes(updated, toUpdatePos, opt); err != nil { if err := adapter.WriteAtLineIndexes(updated, toUpdatePos, opt); err != nil {
return nil, false, err return nil, false, err
} }
for i, rec := range toUpdate { for i := range toUpdate {
if m, ok := rec.(map[string]any); ok { oldM, ok1 := toUpdate[i].(map[string]any)
db.indexStore.UnindexRecord(collectionName, DefaultPartition, indexedFields, m, toUpdatePos[i]) newM, ok2 := updated[i].(map[string]any)
if !ok1 || !ok2 {
continue
} }
} if indexedFieldKeysEqual(oldM, newM, indexedFields) {
for i, rec := range updated { continue
if m, ok := rec.(map[string]any); ok {
db.indexStore.IndexRecord(collectionName, DefaultPartition, indexedFields, m, toUpdatePos[i])
} }
db.indexStore.UnindexRecord(collectionName, partition, indexedFields, oldM, toUpdatePos[i])
db.indexStore.IndexRecord(collectionName, partition, indexedFields, newM, toUpdatePos[i])
} }
return updated, true, nil return updated, true, nil
} }
// tryIndexDelete выполняет точечное удаление через индекс. Возвращает (deletedRecords, used, err). // tryIndexDelete выполняет точечное удаление через индекс. Возвращает (deletedRecords, used, err).
func (db *LineDb) tryIndexDelete(adapter *JSONLFile, filter any, collectionName string, indexedFields []string, options LineDbAdapterOptions) ([]any, bool, error) { func (db *LineDb) tryIndexDelete(adapter *JSONLFile, filter any, collectionName string, partition string, indexedFields []string, options LineDbAdapterOptions) ([]any, bool, error) {
if partition == "" {
partition = DefaultPartition
}
filterMap, ok := filter.(map[string]any) filterMap, ok := filter.(map[string]any)
if !ok || len(filterMap) == 0 { if !ok || len(filterMap) == 0 {
return nil, false, nil return nil, false, nil
@@ -1071,7 +1096,7 @@ func (db *LineDb) tryIndexDelete(adapter *JSONLFile, filter any, collectionName
} }
var indexes []int var indexes []int
for _, p := range posList { for _, p := range posList {
if p.Partition == DefaultPartition { if p.Partition == partition {
indexes = append(indexes, p.LineIndex) indexes = append(indexes, p.LineIndex)
} }
} }
@@ -1101,7 +1126,7 @@ func (db *LineDb) tryIndexDelete(adapter *JSONLFile, filter any, collectionName
} }
for i, rec := range toDel { for i, rec := range toDel {
if m, ok := rec.(map[string]any); ok { if m, ok := rec.(map[string]any); ok {
db.indexStore.UnindexRecord(collectionName, DefaultPartition, indexedFields, m, toDelPos[i]) db.indexStore.UnindexRecord(collectionName, partition, indexedFields, m, toDelPos[i])
} }
} }
return toDel, true, nil return toDel, true, nil
@@ -1316,17 +1341,28 @@ func (db *LineDb) isCollectionPartitioned(collectionName string) bool {
return exists return exists
} }
func (db *LineDb) getPartitionFiles(collectionName string) ([]string, error) { // partitionStorageAdaptersOrdered — для логического имени партиционированной коллекции baseName:
baseName := db.getBaseCollectionName(collectionName) // сначала адаптер основного файла (baseName.jsonl), затем все baseName_* в лексикографическом порядке.
var files []string // Пустой ключ партиции пишет в основной файл; эти записи учитываются здесь первым адаптером.
func (db *LineDb) partitionStorageAdaptersOrdered(baseName string) []*JSONLFile {
for name, filename := range db.collections { var out []*JSONLFile
if a := db.adapters[baseName]; a != nil {
out = append(out, a)
}
var partNames []string
for name := range db.adapters {
if name == baseName {
continue
}
if strings.HasPrefix(name, baseName+"_") { if strings.HasPrefix(name, baseName+"_") {
files = append(files, filename) partNames = append(partNames, name)
} }
} }
sort.Strings(partNames)
return files, nil for _, name := range partNames {
out = append(out, db.adapters[name])
}
return out
} }
func (db *LineDb) getPartitionAdapter(data any, collectionName string) (*JSONLFile, error) { func (db *LineDb) getPartitionAdapter(data any, collectionName string) (*JSONLFile, error) {
@@ -1336,6 +1372,14 @@ func (db *LineDb) getPartitionAdapter(data any, collectionName string) (*JSONLFi
} }
partitionID := partitionFn(data) partitionID := partitionFn(data)
if strings.TrimSpace(partitionID) == "" {
adapter, ok := db.adapters[collectionName]
if !ok {
return nil, fmt.Errorf("collection %s not found", collectionName)
}
return adapter, nil
}
partitionName := fmt.Sprintf("%s_%s", collectionName, partitionID) partitionName := fmt.Sprintf("%s_%s", collectionName, partitionID)
adapter, exists := db.adapters[partitionName] adapter, exists := db.adapters[partitionName]
@@ -1506,24 +1550,6 @@ func (db *LineDb) compareIDs(a, b any) bool {
return a == b 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 { func (db *LineDb) generateCacheKey(filter any, collectionName string) string {
// Упрощенная реализация генерации ключа кэша // Упрощенная реализация генерации ключа кэша
return fmt.Sprintf("%s:%v", collectionName, filter) return fmt.Sprintf("%s:%v", collectionName, filter)
@@ -1589,106 +1615,87 @@ func (db *LineDb) normalizeFilter(filter any) (any, error) {
} }
func (db *LineDb) updatePartitioned(data any, collectionName string, filter any, options LineDbAdapterOptions) ([]any, error) { func (db *LineDb) updatePartitioned(data any, collectionName string, filter any, options LineDbAdapterOptions) ([]any, error) {
partitionFiles, err := db.getPartitionFiles(collectionName) dataMap, err := db.toMap(data)
if err != nil { if err != nil {
return nil, err return nil, fmt.Errorf("invalid update data format: %w", err)
} }
opts := db.getCollectionOptions(collectionName) opts := db.getCollectionOptions(collectionName)
baseName := db.getBaseCollectionName(collectionName)
var allResults []any var allResults []any
for _, filename := range partitionFiles { for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
var adapter *JSONLFile partitionName := adapter.GetCollectionName()
var partitionName string if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
for name, adapterFile := range db.collections { upd, used, terr := db.tryIndexUpdate(adapter, filter, dataMap, collectionName, partitionName, opts.IndexedFields, options)
if adapterFile == filename { if terr != nil {
adapter = db.adapters[name] return nil, terr
partitionName = name }
break if used {
allResults = append(allResults, upd...)
continue
} }
} }
results, err := adapter.Update(data, filter, options)
if adapter != nil { if err != nil {
results, err := adapter.Update(data, filter, options) return nil, err
if err != nil { }
return nil, err allResults = append(allResults, results...)
} if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
allResults = append(allResults, results...) records, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{InTransaction: true})
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 { if readErr == nil {
records, readErr := adapter.Read(LineDbAdapterOptions{InTransaction: true}) _ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records)
}
} }
} }
} }
db.clearCacheForCollection(collectionName)
return allResults, nil return allResults, nil
} }
func (db *LineDb) deletePartitioned(data any, collectionName string, options LineDbAdapterOptions) ([]any, error) { 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) opts := db.getCollectionOptions(collectionName)
baseName := db.getBaseCollectionName(collectionName)
var allResults []any var allResults []any
for _, filename := range partitionFiles { for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
var adapter *JSONLFile partitionName := adapter.GetCollectionName()
var partitionName string if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
for name, adapterFile := range db.collections { del, used, derr := db.tryIndexDelete(adapter, data, collectionName, partitionName, opts.IndexedFields, options)
if adapterFile == filename { if derr != nil {
adapter = db.adapters[name] return nil, derr
partitionName = name }
break if used {
allResults = append(allResults, del...)
continue
} }
} }
results, err := adapter.Delete(data, options)
if adapter != nil { if err != nil {
results, err := adapter.Delete(data, options) return nil, err
if err != nil { }
return nil, err allResults = append(allResults, results...)
} if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
allResults = append(allResults, results...) records, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{InTransaction: true})
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 { if readErr == nil {
records, readErr := adapter.Read(LineDbAdapterOptions{InTransaction: true}) _ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records)
}
} }
} }
} }
db.clearCacheForCollection(collectionName)
return allResults, nil return allResults, nil
} }
func (db *LineDb) readByFilterPartitioned(filter any, collectionName string, options LineDbAdapterOptions) ([]any, error) { func (db *LineDb) readByFilterPartitioned(filter any, collectionName string, options LineDbAdapterOptions) ([]any, error) {
// Получаем все партиции baseName := db.getBaseCollectionName(collectionName)
partitionFiles, err := db.getPartitionFiles(collectionName)
if err != nil {
return nil, err
}
var allResults []any var allResults []any
for _, filename := range partitionFiles { for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
// Находим адаптер по имени файла results, err := adapter.ReadByFilter(filter, options)
var adapter *JSONLFile if err != nil {
for name, adapterFile := range db.collections { return nil, err
if adapterFile == filename {
adapter = db.adapters[name]
break
}
}
if adapter != nil {
results, err := adapter.ReadByFilter(filter, options)
if err != nil {
return nil, err
}
allResults = append(allResults, results...)
} }
allResults = append(allResults, results...)
} }
return allResults, nil return allResults, nil
} }

View File

@@ -27,8 +27,8 @@ func (f *failingLookupStore) UnindexRecord(collection, partition string, fields
f.inner.UnindexRecord(collection, partition, fields, record, lineIndex) f.inner.UnindexRecord(collection, partition, fields, record, lineIndex)
} }
func (f *failingLookupStore) Rebuild(collection, partition string, fields []string, records []any) error { func (f *failingLookupStore) Rebuild(collection, partition string, fields []string, records []any, lineIndexes []int) error {
return f.inner.Rebuild(collection, partition, fields, records) return f.inner.Rebuild(collection, partition, fields, records, lineIndexes)
} }
func (f *failingLookupStore) Clear(collection string) error { func (f *failingLookupStore) Clear(collection string) error {