3 Commits

Author SHA1 Message Date
40c6935b4f change module name 21 2026-04-09 16:33:06 +06:00
0b1deabbaa change module name 20 2026-04-09 15:55:09 +06:00
fafe0d4a66 change module name 10 2026-04-09 15:06:02 +06:00
5 changed files with 389 additions and 139 deletions

View File

@@ -33,8 +33,9 @@ type IndexStore interface {
UnindexRecord(collection, partition string, fields []string, record map[string]any, lineIndex int)
// Rebuild перестраивает вклад в индекс для одной партиции (или всей коллекции, если одна партиция).
// partition — имя партиции; records — записи, позиция в срезе = lineIndex.
Rebuild(collection, partition string, fields []string, records []any) error
// lineIndexes[i] — 0-based номер физической строки в файле для records[i] (после пустых/удалённых строк).
// Если lineIndexes == nil или len != len(records), используются плотные индексы 0..len-1 (устаревшее поведение).
Rebuild(collection, partition string, fields []string, records []any, lineIndexes []int) error
// Clear очищает индекс коллекции (все партиции).
Clear(collection string) error
@@ -164,7 +165,7 @@ func (s *InMemoryIndexStore) Lookup(collection, field, value string) ([]IndexPos
}
// 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 == "" {
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)
if !ok {
continue
@@ -205,7 +210,7 @@ func (s *InMemoryIndexStore) Rebuild(collection, partition string, fields []stri
if s.index[key] == nil {
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

View File

@@ -133,7 +133,7 @@ func (s *MemcachedIndexStore) Lookup(collection, field, value string) ([]IndexPo
}
// 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 == "" {
partition = DefaultPartition
}
@@ -155,7 +155,11 @@ func (s *MemcachedIndexStore) Rebuild(collection, partition string, fields []str
// встречаются в records. Для остальных value старые позиции этой partition останутся.
// Значит нужен merge: для каждого value в records делаем Get, убираем старые Position с этой partition, добавляем новые.
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)
if !ok {
continue
@@ -165,7 +169,7 @@ func (s *MemcachedIndexStore) Rebuild(collection, partition string, fields []str
if byFieldValue[field] == nil {
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()
}
// 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.
// Ожидается, что файл нормализован и каждая строка имеет длину allocSize байт (включая \n),
// как это обеспечивает Init/normalizeExistingFile/rewriteFile.
@@ -662,6 +789,86 @@ func (j *JSONLFile) BlankLinesAtPositions(lineIndexes []int, options LineDbAdapt
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).
// Используется для точечного индексирования после Write.
func (j *JSONLFile) LineCount() (int, error) {

View File

@@ -4,6 +4,7 @@ import (
"fmt"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
@@ -137,26 +138,24 @@ func (db *LineDb) Init(force bool, initOptions *LineDbInitOptions) error {
continue
}
if db.isCollectionPartitioned(opt.CollectionName) {
// Партиции: обходим все существующие партиции (могут быть созданы ранее)
// Основной файл + все base_* (пустой partition id → основной файл)
baseName := opt.CollectionName
for name, adapter := range db.adapters {
if !strings.HasPrefix(name, baseName+"_") {
continue
}
records, err := adapter.Read(LineDbAdapterOptions{})
for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
name := adapter.GetCollectionName()
records, lineIdx, err := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
if err != nil {
continue
}
_ = db.indexStore.Rebuild(opt.CollectionName, name, opt.IndexedFields, records)
_ = db.indexStore.Rebuild(opt.CollectionName, name, opt.IndexedFields, records, lineIdx)
}
} else {
adapter := db.adapters[opt.CollectionName]
if adapter != nil {
records, err := adapter.Read(LineDbAdapterOptions{})
records, lineIdx, err := adapter.ReadWithPhysicalLineIndexes(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 {
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)
}
}
@@ -249,7 +248,9 @@ func (db *LineDb) seedLastIDPartitioned(dbFolder, baseName string) error {
return nil
}
// Read читает все записи из коллекции
// Read читает все записи из коллекции.
// Для партиционированной логической коллекции (имя без суффикса _*) читает основной файл
// и все файлы партиций base_* подряд. Запрос по имени одной партиции (base_part) читает только её файл.
func (db *LineDb) Read(collectionName string, options LineDbAdapterOptions) ([]any, error) {
db.mutex.RLock()
defer db.mutex.RUnlock()
@@ -258,6 +259,23 @@ func (db *LineDb) Read(collectionName string, options LineDbAdapterOptions) ([]a
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]
if !exists {
return nil, fmt.Errorf("collection %s not found", collectionName)
@@ -365,8 +383,9 @@ 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 opts != nil && len(opts.IndexedFields) > 0 {
// Точечное IndexRecord в Write (LineCount → стартовая строка для новых записей)
writeOptions.DoIndexing = true
}
}
if err := db.Write(resultDataArray, collectionName, writeOptions); err != nil {
@@ -380,20 +399,6 @@ func (db *LineDb) Insert(data any, collectionName string, options LineDbAdapterO
}
}
// Обновляем индекс (полная пересборка по коллекции) — только для непартиционированных
if db.indexStore != nil {
opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 && !db.isCollectionPartitioned(collectionName) {
adapter, exists := db.adapters[collectionName]
if exists {
allRecords, readErr := adapter.Read(LineDbAdapterOptions{})
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords)
}
}
}
}
return nil
}
@@ -520,7 +525,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 && !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 {
return nil, err
}
@@ -541,9 +546,9 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li
if db.indexStore != nil {
opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 {
allRecords, readErr := adapter.Read(LineDbAdapterOptions{})
allRecords, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords)
_ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords, lineIdx)
}
}
}
@@ -586,7 +591,7 @@ func (db *LineDb) Delete(data any, collectionName string, options LineDbAdapterO
opts := db.getCollectionOptions(collectionName)
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
result, used, err := db.tryIndexDelete(adapter, data, collectionName, opts.IndexedFields, options)
result, used, err := db.tryIndexDelete(adapter, data, collectionName, DefaultPartition, opts.IndexedFields, options)
if err != nil {
return nil, err
}
@@ -607,9 +612,9 @@ func (db *LineDb) Delete(data any, collectionName string, options LineDbAdapterO
if db.indexStore != nil {
opts := db.getCollectionOptions(collectionName)
if opts != nil && len(opts.IndexedFields) > 0 {
allRecords, readErr := adapter.Read(LineDbAdapterOptions{})
allRecords, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords)
_ = db.indexStore.Rebuild(collectionName, DefaultPartition, opts.IndexedFields, allRecords, lineIdx)
}
}
}
@@ -792,6 +797,8 @@ func (db *LineDb) Close() {
// indexRebuildTimerLoop выполняет полный ребилд всех индексов с заданным интервалом (IndexRebuildTimer).
// Работает в отдельной горутине; при каждом тике берёт db.mutex.Lock() на время ребилда.
// Перед ребилдом коллекции: если в файле есть пустые слоты (след точечного удаления), вызывается
// CompactFile — перезапись без «дыр», затем индексирование по сжатому файлу.
// Останавливается при закрытии db.indexRebuildDone.
func (db *LineDb) indexRebuildTimerLoop(interval time.Duration) {
ticker := time.NewTicker(interval)
@@ -809,22 +816,31 @@ func (db *LineDb) indexRebuildTimerLoop(interval time.Duration) {
}
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{})
for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
name := adapter.GetCollectionName()
dirty, err := adapter.HasBlankSlots(LineDbAdapterOptions{})
if err != nil {
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 {
adapter := db.adapters[opt.CollectionName]
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 {
_ = db.indexStore.Rebuild(opt.CollectionName, DefaultPartition, opt.IndexedFields, records)
_ = db.indexStore.Rebuild(opt.CollectionName, DefaultPartition, opt.IndexedFields, records, lineIdx)
}
}
}
@@ -965,8 +981,23 @@ func (db *LineDb) tryIndexLookupPartitioned(filter any, collectionName string, o
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).
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)
if !ok || len(filterMap) == 0 {
return nil, false, nil
@@ -990,7 +1021,7 @@ func (db *LineDb) tryIndexUpdate(adapter *JSONLFile, filter any, dataMap map[str
}
var indexes []int
for _, p := range posList {
if p.Partition == DefaultPartition {
if p.Partition == partition {
indexes = append(indexes, p.LineIndex)
}
}
@@ -1033,21 +1064,26 @@ func (db *LineDb) tryIndexUpdate(adapter *JSONLFile, filter any, dataMap map[str
if err := adapter.WriteAtLineIndexes(updated, toUpdatePos, opt); err != nil {
return nil, false, err
}
for i, rec := range toUpdate {
if m, ok := rec.(map[string]any); ok {
db.indexStore.UnindexRecord(collectionName, DefaultPartition, indexedFields, m, toUpdatePos[i])
for i := range toUpdate {
oldM, ok1 := toUpdate[i].(map[string]any)
newM, ok2 := updated[i].(map[string]any)
if !ok1 || !ok2 {
continue
}
if indexedFieldKeysEqual(oldM, newM, indexedFields) {
continue
}
for i, rec := range updated {
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
}
// 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)
if !ok || len(filterMap) == 0 {
return nil, false, nil
@@ -1071,7 +1107,7 @@ func (db *LineDb) tryIndexDelete(adapter *JSONLFile, filter any, collectionName
}
var indexes []int
for _, p := range posList {
if p.Partition == DefaultPartition {
if p.Partition == partition {
indexes = append(indexes, p.LineIndex)
}
}
@@ -1101,7 +1137,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, DefaultPartition, indexedFields, m, toDelPos[i])
db.indexStore.UnindexRecord(collectionName, partition, indexedFields, m, toDelPos[i])
}
}
return toDel, true, nil
@@ -1316,17 +1352,28 @@ func (db *LineDb) isCollectionPartitioned(collectionName string) bool {
return exists
}
func (db *LineDb) getPartitionFiles(collectionName string) ([]string, error) {
baseName := db.getBaseCollectionName(collectionName)
var files []string
for name, filename := range db.collections {
// partitionStorageAdaptersOrdered — для логического имени партиционированной коллекции baseName:
// сначала адаптер основного файла (baseName.jsonl), затем все baseName_* в лексикографическом порядке.
// Пустой ключ партиции пишет в основной файл; эти записи учитываются здесь первым адаптером.
func (db *LineDb) partitionStorageAdaptersOrdered(baseName string) []*JSONLFile {
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+"_") {
files = append(files, filename)
partNames = append(partNames, name)
}
}
return files, nil
sort.Strings(partNames)
for _, name := range partNames {
out = append(out, db.adapters[name])
}
return out
}
func (db *LineDb) getPartitionAdapter(data any, collectionName string) (*JSONLFile, error) {
@@ -1336,6 +1383,14 @@ func (db *LineDb) getPartitionAdapter(data any, collectionName string) (*JSONLFi
}
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)
adapter, exists := db.adapters[partitionName]
@@ -1589,35 +1644,35 @@ 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)
dataMap, err := db.toMap(data)
if err != nil {
return nil, err
return nil, fmt.Errorf("invalid update data format: %w", err)
}
opts := db.getCollectionOptions(collectionName)
baseName := db.getBaseCollectionName(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
for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
partitionName := adapter.GetCollectionName()
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
upd, used, terr := db.tryIndexUpdate(adapter, filter, dataMap, collectionName, partitionName, opts.IndexedFields, options)
if terr != nil {
return nil, terr
}
if used {
allResults = append(allResults, upd...)
continue
}
}
if adapter != nil {
results, err := adapter.Update(data, filter, options)
if err != nil {
return nil, err
}
allResults = append(allResults, results...)
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
records, readErr := adapter.Read(LineDbAdapterOptions{InTransaction: true})
records, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{InTransaction: true})
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records)
}
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
}
}
}
@@ -1626,35 +1681,31 @@ 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)
baseName := db.getBaseCollectionName(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
for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
partitionName := adapter.GetCollectionName()
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
del, used, derr := db.tryIndexDelete(adapter, data, collectionName, partitionName, opts.IndexedFields, options)
if derr != nil {
return nil, derr
}
if used {
allResults = append(allResults, del...)
continue
}
}
if adapter != nil {
results, err := adapter.Delete(data, options)
if err != nil {
return nil, err
}
allResults = append(allResults, results...)
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
records, readErr := adapter.Read(LineDbAdapterOptions{InTransaction: true})
records, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{InTransaction: true})
if readErr == nil {
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records)
}
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
}
}
}
@@ -1663,32 +1714,15 @@ func (db *LineDb) deletePartitioned(data any, collectionName string, options Lin
}
func (db *LineDb) readByFilterPartitioned(filter any, collectionName string, options LineDbAdapterOptions) ([]any, error) {
// Получаем все партиции
partitionFiles, err := db.getPartitionFiles(collectionName)
if err != nil {
return nil, err
}
baseName := db.getBaseCollectionName(collectionName)
var allResults []any
for _, filename := range partitionFiles {
// Находим адаптер по имени файла
var adapter *JSONLFile
for name, adapterFile := range db.collections {
if adapterFile == filename {
adapter = db.adapters[name]
break
}
}
if adapter != nil {
for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
results, err := adapter.ReadByFilter(filter, options)
if err != nil {
return nil, err
}
allResults = append(allResults, results...)
}
}
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)
}
func (f *failingLookupStore) Rebuild(collection, partition string, fields []string, records []any) error {
return f.inner.Rebuild(collection, partition, fields, records)
func (f *failingLookupStore) Rebuild(collection, partition string, fields []string, records []any, lineIndexes []int) error {
return f.inner.Rebuild(collection, partition, fields, records, lineIndexes)
}
func (f *failingLookupStore) Clear(collection string) error {