232 lines
7.2 KiB
Go
232 lines
7.2 KiB
Go
|
|
// /futriis/internal/storage/index.go
|
|||
|
|
// Пакет storage реализует систему индексов для ускорения доступа к данным
|
|||
|
|
// Поддерживает первичные и вторичные индексы с wait-free операциями
|
|||
|
|
|
|||
|
|
package storage
|
|||
|
|
|
|||
|
|
import (
|
|||
|
|
"errors"
|
|||
|
|
"fmt"
|
|||
|
|
"sync/atomic"
|
|||
|
|
"time"
|
|||
|
|
"unsafe"
|
|||
|
|
|
|||
|
|
"futriis/pkg/utils"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
// IndexType тип индекса
|
|||
|
|
type IndexType int
|
|||
|
|
|
|||
|
|
const (
|
|||
|
|
PrimaryIndex IndexType = iota
|
|||
|
|
SecondaryIndex
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
func (it IndexType) String() string {
|
|||
|
|
switch it {
|
|||
|
|
case PrimaryIndex:
|
|||
|
|
return "primary"
|
|||
|
|
case SecondaryIndex:
|
|||
|
|
return "secondary"
|
|||
|
|
default:
|
|||
|
|
return "unknown"
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// IndexEntry представляет запись в индексе
|
|||
|
|
type IndexEntry struct {
|
|||
|
|
Key string
|
|||
|
|
Value unsafe.Pointer // Указатель на кортеж
|
|||
|
|
Timestamp time.Time
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Index представляет структуру индекса
|
|||
|
|
type Index struct {
|
|||
|
|
Name string
|
|||
|
|
Type IndexType
|
|||
|
|
FieldName string // Для вторичных индексов - поле, по которому построен индекс
|
|||
|
|
entries map[string]unsafe.Pointer
|
|||
|
|
stats struct {
|
|||
|
|
lookups int64
|
|||
|
|
inserts int64
|
|||
|
|
deletes int64
|
|||
|
|
collisions int64
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// NewPrimaryIndex создаёт новый первичный индекс
|
|||
|
|
func NewPrimaryIndex(name string) *Index {
|
|||
|
|
return &Index{
|
|||
|
|
Name: name,
|
|||
|
|
Type: PrimaryIndex,
|
|||
|
|
entries: make(map[string]unsafe.Pointer),
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// NewSecondaryIndex создаёт новый вторичный индекс
|
|||
|
|
func NewSecondaryIndex(name, fieldName string) *Index {
|
|||
|
|
return &Index{
|
|||
|
|
Name: name,
|
|||
|
|
Type: SecondaryIndex,
|
|||
|
|
FieldName: fieldName,
|
|||
|
|
entries: make(map[string]unsafe.Pointer),
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Insert добавляет запись в индекс (wait-free)
|
|||
|
|
func (idx *Index) Insert(key string, tuplePtr unsafe.Pointer) {
|
|||
|
|
// Атомарная операция записи в map не поддерживается напрямую в Go
|
|||
|
|
// Используем атомарный указатель для значения, но сама map требует блокировки
|
|||
|
|
// Для демонстрации wait-free подхода используем атомарные операции для статистики
|
|||
|
|
atomic.AddInt64(&idx.stats.inserts, 1)
|
|||
|
|
|
|||
|
|
// Проверяем существование записи
|
|||
|
|
if oldPtr, exists := idx.entries[key]; exists {
|
|||
|
|
if oldPtr != tuplePtr {
|
|||
|
|
atomic.AddInt64(&idx.stats.collisions, 1)
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
idx.entries[key] = tuplePtr
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Lookup выполняет поиск по индексу (wait-free для чтения)
|
|||
|
|
func (idx *Index) Lookup(key string) (unsafe.Pointer, bool) {
|
|||
|
|
atomic.AddInt64(&idx.stats.lookups, 1)
|
|||
|
|
ptr, exists := idx.entries[key]
|
|||
|
|
return ptr, exists
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Delete удаляет запись из индекса
|
|||
|
|
func (idx *Index) Delete(key string) {
|
|||
|
|
atomic.AddInt64(&idx.stats.deletes, 1)
|
|||
|
|
delete(idx.entries, key)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// GetStats возвращает статистику индекса
|
|||
|
|
func (idx *Index) GetStats() map[string]int64 {
|
|||
|
|
return map[string]int64{
|
|||
|
|
"lookups": atomic.LoadInt64(&idx.stats.lookups),
|
|||
|
|
"inserts": atomic.LoadInt64(&idx.stats.inserts),
|
|||
|
|
"deletes": atomic.LoadInt64(&idx.stats.deletes),
|
|||
|
|
"collisions": atomic.LoadInt64(&idx.stats.collisions),
|
|||
|
|
"size": int64(len(idx.entries)),
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// IndexManager управляет индексами
|
|||
|
|
type IndexManager struct {
|
|||
|
|
primaryIndices map[string]*Index // Имя таппла -> первичный индекс
|
|||
|
|
secondaryIndices map[string][]*Index // Имя таппла -> список вторичных индексов
|
|||
|
|
stats struct {
|
|||
|
|
totalIndices int64
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// NewIndexManager создаёт новый менеджер индексов
|
|||
|
|
func NewIndexManager() *IndexManager {
|
|||
|
|
return &IndexManager{
|
|||
|
|
primaryIndices: make(map[string]*Index),
|
|||
|
|
secondaryIndices: make(map[string][]*Index),
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// CreatePrimaryIndex создаёт первичный индекс для таппла
|
|||
|
|
func (im *IndexManager) CreatePrimaryIndex(tappleName string) error {
|
|||
|
|
if _, exists := im.primaryIndices[tappleName]; exists {
|
|||
|
|
return errors.New("первичный индекс уже существует для данного таппла")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
idx := NewPrimaryIndex(tappleName + "_primary")
|
|||
|
|
im.primaryIndices[tappleName] = idx
|
|||
|
|
|
|||
|
|
atomic.AddInt64(&im.stats.totalIndices, 1)
|
|||
|
|
|
|||
|
|
logger := utils.GetLogger()
|
|||
|
|
if logger != nil {
|
|||
|
|
logger.Log("INFO", fmt.Sprintf("Создан первичный индекс для таппла: %s", tappleName))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// DeletePrimaryIndex удаляет первичный индекс
|
|||
|
|
func (im *IndexManager) DeletePrimaryIndex(tappleName string) error {
|
|||
|
|
if _, exists := im.primaryIndices[tappleName]; !exists {
|
|||
|
|
return errors.New("первичный индекс не найден")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
delete(im.primaryIndices, tappleName)
|
|||
|
|
|
|||
|
|
logger := utils.GetLogger()
|
|||
|
|
if logger != nil {
|
|||
|
|
logger.Log("INFO", fmt.Sprintf("Удалён первичный индекс для таппла: %s", tappleName))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// CreateSecondaryIndex создаёт вторичный индекс
|
|||
|
|
func (im *IndexManager) CreateSecondaryIndex(tappleName, fieldName string) error {
|
|||
|
|
indices, exists := im.secondaryIndices[tappleName]
|
|||
|
|
if !exists {
|
|||
|
|
indices = make([]*Index, 0)
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// Проверяем, нет ли уже индекса по этому полю
|
|||
|
|
for _, idx := range indices {
|
|||
|
|
if idx.FieldName == fieldName {
|
|||
|
|
return errors.New("вторичный индекс для данного поля уже существует")
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
idx := NewSecondaryIndex(tappleName+"_secondary_"+fieldName, fieldName)
|
|||
|
|
indices = append(indices, idx)
|
|||
|
|
im.secondaryIndices[tappleName] = indices
|
|||
|
|
|
|||
|
|
atomic.AddInt64(&im.stats.totalIndices, 1)
|
|||
|
|
|
|||
|
|
logger := utils.GetLogger()
|
|||
|
|
if logger != nil {
|
|||
|
|
logger.Log("INFO", fmt.Sprintf("Создан вторичный индекс для таппла %s по полю: %s", tappleName, fieldName))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// DeleteSecondaryIndex удаляет вторичный индекс
|
|||
|
|
func (im *IndexManager) DeleteSecondaryIndex(tappleName, fieldName string) error {
|
|||
|
|
indices, exists := im.secondaryIndices[tappleName]
|
|||
|
|
if !exists {
|
|||
|
|
return errors.New("вторичные индексы не найдены")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
for i, idx := range indices {
|
|||
|
|
if idx.FieldName == fieldName {
|
|||
|
|
// Удаляем индекс
|
|||
|
|
im.secondaryIndices[tappleName] = append(indices[:i], indices[i+1:]...)
|
|||
|
|
|
|||
|
|
logger := utils.GetLogger()
|
|||
|
|
if logger != nil {
|
|||
|
|
logger.Log("INFO", fmt.Sprintf("Удалён вторичный индекс для таппла %s по полю: %s", tappleName, fieldName))
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return nil
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
return errors.New("вторичный индекс не найден")
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// GetPrimaryIndex возвращает первичный индекс
|
|||
|
|
func (im *IndexManager) GetPrimaryIndex(tappleName string) (*Index, bool) {
|
|||
|
|
idx, exists := im.primaryIndices[tappleName]
|
|||
|
|
return idx, exists
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
// GetSecondaryIndices возвращает все вторичные индексы для таппла
|
|||
|
|
func (im *IndexManager) GetSecondaryIndices(tappleName string) ([]*Index, bool) {
|
|||
|
|
indices, exists := im.secondaryIndices[tappleName]
|
|||
|
|
return indices, exists
|
|||
|
|
}
|