Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 0b1deabbaa | |||
| fafe0d4a66 | |||
| a1bbd9177a | |||
| 7894474f2d |
@@ -25,7 +25,7 @@ package main
|
|||||||
import (
|
import (
|
||||||
"log"
|
"log"
|
||||||
"time"
|
"time"
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
type User struct {
|
type User struct {
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestCustomJSONSerialization(t *testing.T) {
|
func TestCustomJSONSerialization(t *testing.T) {
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// User представляет пользователя
|
// User представляет пользователя
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Item — структура для вставки (LineDB поддерживает struct и map)
|
// Item — структура для вставки (LineDB поддерживает struct и map)
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// testDeleteOneRecordByID тестирует удаление одной записи по ID
|
// testDeleteOneRecordByID тестирует удаление одной записи по ID
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// testSingleInsert тестирует вставку одной записи
|
// testSingleInsert тестирует вставку одной записи
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// User представляет пользователя
|
// User представляет пользователя
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import (
|
|||||||
"path/filepath"
|
"path/filepath"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Пример работы с партициями + индексируемой коллекцией.
|
// Пример работы с партициями + индексируемой коллекцией.
|
||||||
|
|||||||
@@ -20,7 +20,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
|||||||
@@ -9,7 +9,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
func main() {
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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) {
|
||||||
|
|||||||
@@ -4,7 +4,9 @@ import (
|
|||||||
"sync"
|
"sync"
|
||||||
)
|
)
|
||||||
|
|
||||||
// LastIDManager управляет последними ID для коллекций
|
// LastIDManager управляет последними ID для коллекций.
|
||||||
|
// Ключ — полное имя коллекции (как в NextID/Insert), без обрезки по «_»:
|
||||||
|
// иначе user_data и events_A ломались бы на первый сегмент.
|
||||||
type LastIDManager struct {
|
type LastIDManager struct {
|
||||||
lastIDs map[string]int
|
lastIDs map[string]int
|
||||||
mutex sync.RWMutex
|
mutex sync.RWMutex
|
||||||
@@ -23,53 +25,32 @@ func GetLastIDManagerInstance() *LastIDManager {
|
|||||||
return lastIDManagerInstance
|
return lastIDManagerInstance
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetLastID получает последний ID для коллекции
|
// GetLastID получает последний ID для коллекции (ключ — полное имя коллекции).
|
||||||
func (l *LastIDManager) GetLastID(filename string) int {
|
func (l *LastIDManager) GetLastID(collectionKey string) int {
|
||||||
l.mutex.RLock()
|
l.mutex.RLock()
|
||||||
defer l.mutex.RUnlock()
|
defer l.mutex.RUnlock()
|
||||||
|
|
||||||
baseFileName := l.getBaseFileName(filename)
|
return l.lastIDs[collectionKey]
|
||||||
return l.lastIDs[baseFileName]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetLastID устанавливает последний ID для коллекции
|
// SetLastID устанавливает последний ID для коллекции, если id больше текущего.
|
||||||
func (l *LastIDManager) SetLastID(filename string, id int) {
|
func (l *LastIDManager) SetLastID(collectionKey string, id int) {
|
||||||
l.mutex.Lock()
|
l.mutex.Lock()
|
||||||
defer l.mutex.Unlock()
|
defer l.mutex.Unlock()
|
||||||
|
|
||||||
baseFileName := l.getBaseFileName(filename)
|
currentID := l.lastIDs[collectionKey]
|
||||||
currentID := l.lastIDs[baseFileName]
|
|
||||||
if currentID < id {
|
if currentID < id {
|
||||||
l.lastIDs[baseFileName] = id
|
l.lastIDs[collectionKey] = id
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// IncrementLastID увеличивает последний ID для коллекции
|
// IncrementLastID увеличивает последний ID для коллекции
|
||||||
func (l *LastIDManager) IncrementLastID(filename string) int {
|
func (l *LastIDManager) IncrementLastID(collectionKey string) int {
|
||||||
l.mutex.Lock()
|
l.mutex.Lock()
|
||||||
defer l.mutex.Unlock()
|
defer l.mutex.Unlock()
|
||||||
|
|
||||||
baseFileName := l.getBaseFileName(filename)
|
currentID := l.lastIDs[collectionKey]
|
||||||
currentID := l.lastIDs[baseFileName]
|
|
||||||
newID := currentID + 1
|
newID := currentID + 1
|
||||||
l.lastIDs[baseFileName] = newID
|
l.lastIDs[collectionKey] = newID
|
||||||
return newID
|
return newID
|
||||||
}
|
}
|
||||||
|
|
||||||
// getBaseFileName извлекает базовое имя файла
|
|
||||||
func (l *LastIDManager) getBaseFileName(filename string) string {
|
|
||||||
if idx := l.findPartitionSeparator(filename); idx != -1 {
|
|
||||||
return filename[:idx]
|
|
||||||
}
|
|
||||||
return filename
|
|
||||||
}
|
|
||||||
|
|
||||||
// findPartitionSeparator находит разделитель партиции
|
|
||||||
func (l *LastIDManager) findPartitionSeparator(filename string) int {
|
|
||||||
for i, char := range filename {
|
|
||||||
if char == '_' {
|
|
||||||
return i
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return -1
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -143,20 +143,20 @@ func (db *LineDb) Init(force bool, initOptions *LineDbInitOptions) error {
|
|||||||
if !strings.HasPrefix(name, baseName+"_") {
|
if !strings.HasPrefix(name, baseName+"_") {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
records, err := adapter.Read(LineDbAdapterOptions{})
|
records, lineIdx, err := adapter.ReadWithPhysicalLineIndexes(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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -171,6 +171,81 @@ func (db *LineDb) Init(force bool, initOptions *LineDbInitOptions) error {
|
|||||||
go db.indexRebuildTimerLoop(interval)
|
go db.indexRebuildTimerLoop(interval)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if err := db.seedLastIDsFromData(dbFolder); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// seedLastIDsFromData выставляет LastIDManager по максимальному id в существующих данных
|
||||||
|
// (после рестарта процесса счётчик не начинается с 1).
|
||||||
|
func (db *LineDb) seedLastIDsFromData(dbFolder string) error {
|
||||||
|
for i, opt := range db.initOptions.Collections {
|
||||||
|
collectionName := opt.CollectionName
|
||||||
|
if collectionName == "" {
|
||||||
|
collectionName = fmt.Sprintf("collection_%d", i+1)
|
||||||
|
}
|
||||||
|
if db.isCollectionPartitioned(collectionName) {
|
||||||
|
if err := db.seedLastIDPartitioned(dbFolder, collectionName); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
adapter := db.adapters[collectionName]
|
||||||
|
if adapter == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
records, err := adapter.Read(LineDbAdapterOptions{})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("seed last id: read %s: %w", collectionName, err)
|
||||||
|
}
|
||||||
|
db.lastIDManager.SetLastID(collectionName, db.GetMaxID(records))
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// seedLastIDPartitioned — максимальный id по базовому файлу и всем events_*.jsonl на диске.
|
||||||
|
// NextID для партиций вызывается с логическим именем (например events), ключ тот же.
|
||||||
|
func (db *LineDb) seedLastIDPartitioned(dbFolder, baseName string) error {
|
||||||
|
maxID := 0
|
||||||
|
if a := db.adapters[baseName]; a != nil {
|
||||||
|
recs, err := a.Read(LineDbAdapterOptions{})
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("seed last id: read base %s: %w", baseName, err)
|
||||||
|
}
|
||||||
|
if m := db.GetMaxID(recs); m > maxID {
|
||||||
|
maxID = m
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pattern := filepath.Join(dbFolder, baseName+"_*.jsonl")
|
||||||
|
matches, err := filepath.Glob(pattern)
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("seed last id: glob %s: %w", pattern, err)
|
||||||
|
}
|
||||||
|
for _, path := range matches {
|
||||||
|
base := filepath.Base(path)
|
||||||
|
partName := strings.TrimSuffix(base, ".jsonl")
|
||||||
|
var adapter *JSONLFile
|
||||||
|
if existing, ok := db.adapters[partName]; ok {
|
||||||
|
adapter = existing
|
||||||
|
} else {
|
||||||
|
adapter = NewJSONLFile(path, "", JSONLFileOptions{CollectionName: partName})
|
||||||
|
if err := adapter.Init(false, LineDbAdapterOptions{}); err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
db.adapters[partName] = adapter
|
||||||
|
db.collections[partName] = path
|
||||||
|
}
|
||||||
|
recs, err := adapter.Read(LineDbAdapterOptions{})
|
||||||
|
if err != nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if m := db.GetMaxID(recs); m > maxID {
|
||||||
|
maxID = m
|
||||||
|
}
|
||||||
|
}
|
||||||
|
db.lastIDManager.SetLastID(baseName, maxID)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -290,8 +365,9 @@ 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 {
|
||||||
@@ -305,20 +381,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
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -445,7 +507,7 @@ 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
|
||||||
}
|
}
|
||||||
@@ -466,9 +528,9 @@ 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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -511,7 +573,7 @@ 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
|
||||||
}
|
}
|
||||||
@@ -532,9 +594,9 @@ 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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -717,6 +779,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)
|
||||||
@@ -738,18 +802,29 @@ func (db *LineDb) indexRebuildTimerLoop(interval time.Duration) {
|
|||||||
if !strings.HasPrefix(name, baseName+"_") {
|
if !strings.HasPrefix(name, baseName+"_") {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
records, err := adapter.Read(LineDbAdapterOptions{})
|
dirty, err := adapter.HasBlankSlots(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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -890,8 +965,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
|
||||||
@@ -915,7 +1005,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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -958,21 +1048,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) {
|
||||||
|
continue
|
||||||
}
|
}
|
||||||
for i, rec := range updated {
|
db.indexStore.UnindexRecord(collectionName, partition, indexedFields, oldM, toUpdatePos[i])
|
||||||
if m, ok := rec.(map[string]any); ok {
|
db.indexStore.IndexRecord(collectionName, partition, indexedFields, newM, toUpdatePos[i])
|
||||||
db.indexStore.IndexRecord(collectionName, DefaultPartition, indexedFields, m, 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
|
||||||
@@ -996,7 +1091,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)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1026,7 +1121,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
|
||||||
@@ -1285,8 +1380,9 @@ func (db *LineDb) GetMaxID(records []any) int {
|
|||||||
for _, record := range records {
|
for _, record := range records {
|
||||||
if recordMap, ok := record.(map[string]any); ok {
|
if recordMap, ok := record.(map[string]any); ok {
|
||||||
if id, ok := recordMap["id"]; ok {
|
if id, ok := recordMap["id"]; ok {
|
||||||
if idInt, ok := id.(int); ok && idInt > maxID {
|
n := idToIntForMax(id)
|
||||||
maxID = idInt
|
if n > maxID {
|
||||||
|
maxID = n
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1294,6 +1390,21 @@ func (db *LineDb) GetMaxID(records []any) int {
|
|||||||
return maxID
|
return maxID
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func idToIntForMax(id any) int {
|
||||||
|
switch v := id.(type) {
|
||||||
|
case int:
|
||||||
|
return v
|
||||||
|
case int64:
|
||||||
|
return int(v)
|
||||||
|
case float64:
|
||||||
|
return int(v)
|
||||||
|
case float32:
|
||||||
|
return int(v)
|
||||||
|
default:
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (db *LineDb) matchesFilter(record any, filter any, strictCompare bool) bool {
|
func (db *LineDb) matchesFilter(record any, filter any, strictCompare bool) bool {
|
||||||
if recordMap, ok := record.(map[string]any); ok {
|
if recordMap, ok := record.(map[string]any); ok {
|
||||||
if filterMap, ok := filter.(map[string]any); ok {
|
if filterMap, ok := filter.(map[string]any); ok {
|
||||||
@@ -1383,10 +1494,32 @@ func (db *LineDb) isInvalidID(id any) bool {
|
|||||||
if id == nil {
|
if id == nil {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
if idNum, ok := id.(int); ok {
|
switch v := id.(type) {
|
||||||
return idNum <= -1
|
case int:
|
||||||
}
|
return v <= -1
|
||||||
|
case int8:
|
||||||
|
return v <= -1
|
||||||
|
case int16:
|
||||||
|
return v <= -1
|
||||||
|
case int32:
|
||||||
|
return v <= -1
|
||||||
|
case int64:
|
||||||
|
return v <= -1
|
||||||
|
case uint, uint8, uint16, uint32, uint64:
|
||||||
return false
|
return false
|
||||||
|
case float32:
|
||||||
|
if v != v { // NaN
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
return float64(v) <= -1
|
||||||
|
case float64:
|
||||||
|
if v != v { // NaN
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
return v <= -1
|
||||||
|
default:
|
||||||
|
return false
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (db *LineDb) compareIDs(a, b any) bool {
|
func (db *LineDb) compareIDs(a, b any) bool {
|
||||||
@@ -1480,6 +1613,10 @@ func (db *LineDb) updatePartitioned(data any, collectionName string, filter any,
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
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)
|
opts := db.getCollectionOptions(collectionName)
|
||||||
|
|
||||||
var allResults []any
|
var allResults []any
|
||||||
@@ -1495,15 +1632,25 @@ func (db *LineDb) updatePartitioned(data any, collectionName string, filter any,
|
|||||||
}
|
}
|
||||||
|
|
||||||
if adapter != nil {
|
if adapter != nil {
|
||||||
|
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
|
||||||
|
}
|
||||||
|
}
|
||||||
results, err := adapter.Update(data, filter, options)
|
results, err := adapter.Update(data, filter, options)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
allResults = append(allResults, results...)
|
allResults = append(allResults, results...)
|
||||||
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
|
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 {
|
if readErr == nil {
|
||||||
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records)
|
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1532,15 +1679,25 @@ func (db *LineDb) deletePartitioned(data any, collectionName string, options Lin
|
|||||||
}
|
}
|
||||||
|
|
||||||
if adapter != nil {
|
if adapter != nil {
|
||||||
|
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
|
||||||
|
}
|
||||||
|
}
|
||||||
results, err := adapter.Delete(data, options)
|
results, err := adapter.Delete(data, options)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
allResults = append(allResults, results...)
|
allResults = append(allResults, results...)
|
||||||
if db.indexStore != nil && opts != nil && len(opts.IndexedFields) > 0 {
|
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 {
|
if readErr == nil {
|
||||||
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records)
|
_ = db.indexStore.Rebuild(collectionName, partitionName, opts.IndexedFields, records, lineIdx)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// mockMemcached — in-memory реализация MemcachedClient для тестов.
|
// mockMemcached — in-memory реализация MemcachedClient для тестов.
|
||||||
@@ -111,7 +111,7 @@ func TestIndexEncodedCollectionCache(t *testing.T) {
|
|||||||
}
|
}
|
||||||
initOptions := &linedb.LineDbInitOptions{
|
initOptions := &linedb.LineDbInitOptions{
|
||||||
CacheSize: 100,
|
CacheSize: 100,
|
||||||
CacheTTL: time.Minute*10,
|
CacheTTL: time.Minute * 10,
|
||||||
DBFolder: "./data/test-linedb-index-enc",
|
DBFolder: "./data/test-linedb-index-enc",
|
||||||
Collections: []linedb.JSONLFileOptions{
|
Collections: []linedb.JSONLFileOptions{
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
func setupPointUpdateDB(t *testing.T) (*linedb.LineDb, func()) {
|
func setupPointUpdateDB(t *testing.T) (*linedb.LineDb, func()) {
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestLineDbBasic(t *testing.T) {
|
func TestLineDbBasic(t *testing.T) {
|
||||||
|
|||||||
@@ -5,7 +5,7 @@ import (
|
|||||||
"path/filepath"
|
"path/filepath"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
const dbDir = "./data/nonpartitioned"
|
const dbDir = "./data/nonpartitioned"
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import (
|
|||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
// failingLookupStore — IndexStore, у которого Lookup всегда возвращает ошибку (для проверки FailOnFailureIndexRead).
|
// failingLookupStore — IndexStore, у которого Lookup всегда возвращает ошибку (для проверки FailOnFailureIndexRead).
|
||||||
@@ -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 {
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"linedb/pkg/linedb"
|
"direct-dev.ru/gitea/GiteaAdmin/elowdb-go/pkg/linedb"
|
||||||
)
|
)
|
||||||
|
|
||||||
const dbDir = "./data/partitioned"
|
const dbDir = "./data/partitioned"
|
||||||
|
|||||||
Reference in New Issue
Block a user