finish index feature

2026-04-07 11:49:42 +06:00
parent 8ba956d8c5
commit 15db6e81db
37 changed files with 1047 additions and 170 deletions
--- 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 {