change module name 22 очистка кэша при вставке и измении и удалении

pkg/linedb/cache.go

@@ -1,6 +1,7 @@
 package linedb
 
 import (
+	"strings"
 	"sync"
 	"time"
 )
@@ -94,6 +95,37 @@ func (c *RecordCache) Clear() {
 	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 удаляет из кэша только те записи, в данных которых
 // встречается хотя бы один из переданных id. Если ids пуст — ничего не делает.
 func (c *RecordCache) ClearEntriesContainingIDs(ids []any) {

pkg/linedb/line_db.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -137,12 +138,10 @@ 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 {
+				for _, adapter := range db.partitionStorageAdaptersOrdered(baseName) {
-					if !strings.HasPrefix(name, baseName+"_") {
+					name := adapter.GetCollectionName()
-						continue
-					}
 					records, lineIdx, err := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{})
 					if err != nil {
 						continue
@@ -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)
@@ -374,12 +392,7 @@ func (db *LineDb) Insert(data any, collectionName string, options LineDbAdapterO
 		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)
-		}
-	}
 
 	return nil
 }
@@ -512,10 +525,7 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li
 			return nil, err
 		}
 		if used {
-			if db.cacheExternal != nil {
+			db.clearCacheForCollection(collectionName)
-				ids := extractIDsFromRecords(result)
-				db.cacheExternal.ClearEntriesContainingIDs(ids)
-			}
 			return result, nil
 		}
 	}
@@ -534,10 +544,7 @@ func (db *LineDb) Update(data any, collectionName string, filter any, options Li
 			}
 		}
 	}
-	if db.cacheExternal != nil {
+	db.clearCacheForCollection(collectionName)
-		ids := extractIDsFromRecords(result)
-		db.cacheExternal.ClearEntriesContainingIDs(ids)
-	}
 	return result, nil
 }
 
@@ -578,10 +585,7 @@ func (db *LineDb) Delete(data any, collectionName string, options LineDbAdapterO
 			return nil, err
 		}
 		if used {
-			if db.cacheExternal != nil {
+			db.clearCacheForCollection(collectionName)
-				ids := extractIDsFromRecords(result)
-				db.cacheExternal.ClearEntriesContainingIDs(ids)
-			}
 			return result, nil
 		}
 	}
@@ -600,10 +604,7 @@ func (db *LineDb) Delete(data any, collectionName string, options LineDbAdapterO
 			}
 		}
 	}
-	if db.cacheExternal != nil {
+	db.clearCacheForCollection(collectionName)
-		ids := extractIDsFromRecords(result)
-		db.cacheExternal.ClearEntriesContainingIDs(ids)
-	}
 	return result, nil
 }
 
@@ -740,15 +741,21 @@ func (db *LineDb) ClearCache(collectionName string, options LineDbAdapterOptions
 		if collectionName == "" {
 			db.cacheExternal.Clear()
 		} else {
-			// Очищаем только записи для конкретной коллекции
+			db.cacheExternal.ClearCollection(collectionName)
-			// Это упрощенная реализация
-			db.cacheExternal.Clear()
 		}
 	}
 
 	return nil
 }
 
+// clearCacheForCollection сбрасывает кэш запросов по логической коллекции (включая ключи партиций base_*).
+func (db *LineDb) clearCacheForCollection(collectionName string) {
+	if db.cacheExternal == nil || collectionName == "" {
+		return
+	}
+	db.cacheExternal.ClearCollection(collectionName)
+}
+
 // Close закрывает базу данных
 func (db *LineDb) Close() {
 	db.mutex.Lock()
@@ -798,10 +805,8 @@ func (db *LineDb) indexRebuildTimerLoop(interval time.Duration) {
 					}
 					if db.isCollectionPartitioned(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{})
 							if err != nil {
 								continue
@@ -1336,17 +1341,28 @@ func (db *LineDb) isCollectionPartitioned(collectionName string) bool {
 	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+"_") {
-			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) {
@@ -1356,6 +1372,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]
@@ -1526,24 +1550,6 @@ func (db *LineDb) compareIDs(a, b any) bool {
 	return a == b
 }
 
-// extractIDsFromRecords извлекает id из списка записей (map[string]any).
-func extractIDsFromRecords(records []any) []any {
-	if len(records) == 0 {
-		return nil
-	}
-	ids := make([]any, 0, len(records))
-	for _, rec := range records {
-		m, ok := rec.(map[string]any)
-		if !ok {
-			continue
-		}
-		if id, exists := m["id"]; exists && id != nil {
-			ids = append(ids, id)
-		}
-	}
-	return ids
-}
-
 func (db *LineDb) generateCacheKey(filter any, collectionName string) string {
 	// Упрощенная реализация генерации ключа кэша
 	return fmt.Sprintf("%s:%v", collectionName, filter)
@@ -1609,130 +1615,87 @@ 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
-	}
 	dataMap, err := db.toMap(data)
 	if err != nil {
 		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 {
+	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 {
-			if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
+			return nil, err
-				upd, used, terr := db.tryIndexUpdate(adapter, filter, dataMap, collectionName, partitionName, opts.IndexedFields, options)
+		}
-				if terr != nil {
+		allResults = append(allResults, results...)
-					return nil, terr
+		if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
-				}
+			records, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{InTransaction: true})
-				if used {
+			if readErr == nil {
-					allResults = append(allResults, upd...)
+				_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
-					continue
-				}
-			}
-			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, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{InTransaction: true})
-				if readErr == nil {
-					_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
-				}
 			}
 		}
 	}
 
+	db.clearCacheForCollection(collectionName)
 	return allResults, nil
 }
 
 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 {
+	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 {
-			if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
+			return nil, err
-				del, used, derr := db.tryIndexDelete(adapter, data, collectionName, partitionName, opts.IndexedFields, options)
+		}
-				if derr != nil {
+		allResults = append(allResults, results...)
-					return nil, derr
+		if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
-				}
+			records, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{InTransaction: true})
-				if used {
+			if readErr == nil {
-					allResults = append(allResults, del...)
+				_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
-					continue
-				}
-			}
-			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, lineIdx, readErr := adapter.ReadWithPhysicalLineIndexes(LineDbAdapterOptions{InTransaction: true})
-				if readErr == nil {
-					_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
-				}
 			}
 		}
 	}
 
+	db.clearCacheForCollection(collectionName)
 	return allResults, nil
 }
 
 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
-	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
 }