futriis/internal/storage/compression.go

// /futriis/internal/storage/compression.go
// Пакет storage реализует простейшее сжатие для колонок с одинаковыми типами данных

package storage

import (
	"encoding/binary"
	"math"
	"strconv"
	"sync/atomic"
)

// CompressionType тип сжатия
type CompressionType int

const (
	NoCompression CompressionType = iota
	RLECompression                // Run-length encoding для повторяющихся значений
	DeltaCompression              // Дельта-сжатие для чисел
	DictionaryCompression         // Словарное сжатие для строк
)

// ColumnCompressor предоставляет сжатие для колонки
type ColumnCompressor struct {
	colType     string
	compType    CompressionType
	stats       struct {
		originalSize  int64
		compressedSize int64
		savings       int64
	}
}

// NewColumnCompressor создаёт новый компрессор для колонки
func NewColumnCompressor(colType string) *ColumnCompressor {
	cc := &ColumnCompressor{
		colType:  colType,
		compType: NoCompression,
	}
	
	// Автоматически выбираем тип сжатия на основе типа данных
	switch colType {
	case "int", "int64", "float64":
		cc.compType = DeltaCompression
	case "string":
		cc.compType = DictionaryCompression
	default:
		cc.compType = RLECompression
	}
	
	return cc
}

// Compress сжимает данные
func (cc *ColumnCompressor) Compress(data []interface{}) ([]byte, error) {
	var compressed []byte
	var err error
	
	switch cc.compType {
	case RLECompression:
		compressed, err = cc.rleCompress(data)
	case DeltaCompression:
		compressed, err = cc.deltaCompress(data)
	case DictionaryCompression:
		compressed, err = cc.dictionaryCompress(data)
	default:
		// Без сжатия - просто сериализуем
		compressed, err = cc.noCompress(data)
	}
	
	if err != nil {
		return nil, err
	}
	
	// Обновляем статистику
	originalSize := int64(len(data) * 8) // Примерная оценка
	compressedSize := int64(len(compressed))
	
	atomic.AddInt64(&cc.stats.originalSize, originalSize)
	atomic.AddInt64(&cc.stats.compressedSize, compressedSize)
	atomic.AddInt64(&cc.stats.savings, originalSize-compressedSize)
	
	return compressed, nil
}

// rleCompress реализует сжатие повторяющихся значений
func (cc *ColumnCompressor) rleCompress(data []interface{}) ([]byte, error) {
	if len(data) == 0 {
		return []byte{}, nil
	}
	
	result := make([]byte, 0)
	
	current := data[0]
	count := 1
	
	for i := 1; i < len(data); i++ {
		if data[i] == current {
			count++
		} else {
			// Записываем значение и счётчик
			result = append(result, []byte(encodeValue(current))...)
			result = append(result, byte(count))
			
			current = data[i]
			count = 1
		}
	}
	
	// Записываем последнее значение
	result = append(result, []byte(encodeValue(current))...)
	result = append(result, byte(count))
	
	return result, nil
}

// deltaCompress реализует дельта-сжатие для чисел
func (cc *ColumnCompressor) deltaCompress(data []interface{}) ([]byte, error) {
	if len(data) == 0 {
		return []byte{}, nil
	}
	
	result := make([]byte, 8) // Первое значение храним полностью
	
	// Преобразуем первое значение
	first, ok := data[0].(float64)
	if !ok {
		if i, ok := data[0].(int); ok {
			first = float64(i)
		} else {
			return cc.noCompress(data)
		}
	}
	
	binary.LittleEndian.PutUint64(result, math.Float64bits(first))
	
	// Для остальных храним дельты
	for i := 1; i < len(data); i++ {
		var curr float64
		switch v := data[i].(type) {
		case float64:
			curr = v
		case int:
			curr = float64(v)
		default:
			return cc.noCompress(data)
		}
		
		prev, _ := data[i-1].(float64)
		if iPrev, ok := data[i-1].(int); ok {
			prev = float64(iPrev)
		}
		
		delta := int16(curr - prev)
		deltaBytes := make([]byte, 2)
		binary.LittleEndian.PutUint16(deltaBytes, uint16(delta))
		result = append(result, deltaBytes...)
	}
	
	return result, nil
}

// dictionaryCompress реализует словарное сжатие для строк
func (cc *ColumnCompressor) dictionaryCompress(data []interface{}) ([]byte, error) {
	// Строим словарь уникальных значений
	dict := make(map[string]byte)
	values := make([]byte, len(data))
	
	nextCode := byte(0)
	
	for i, val := range data {
		str, ok := val.(string)
		if !ok {
			return cc.noCompress(data)
		}
		
		code, exists := dict[str]
		if !exists {
			code = nextCode
			dict[str] = code
			nextCode++
		}
		
		values[i] = code
	}
	
	// Кодируем: сначала словарь, затем значения
	result := make([]byte, 0)
	
	// Записываем размер словаря
	result = append(result, byte(len(dict)))
	
	// Записываем словарь
	for str, code := range dict {
		result = append(result, code)
		result = append(result, byte(len(str)))
		result = append(result, []byte(str)...)
	}
	
	// Записываем значения
	result = append(result, values...)
	
	return result, nil
}

// noCompress без сжатия
func (cc *ColumnCompressor) noCompress(data []interface{}) ([]byte, error) {
	result := make([]byte, 0)
	for _, val := range data {
		result = append(result, []byte(encodeValue(val))...)
	}
	return result, nil
}

// encodeValue кодирует значение в строку
func encodeValue(val interface{}) string {
	switch v := val.(type) {
	case string:
		return v
	case int:
		return strconv.Itoa(v)
	case int64:
		return strconv.FormatInt(v, 10)
	case float64:
		return strconv.FormatFloat(v, 'f', -1, 64)
	case bool:
		return strconv.FormatBool(v)
	default:
		return ""
	}
}

// GetStats возвращает статистику сжатия
func (cc *ColumnCompressor) GetStats() map[string]int64 {
	return map[string]int64{
		"original_size":   atomic.LoadInt64(&cc.stats.originalSize),
		"compressed_size": atomic.LoadInt64(&cc.stats.compressedSize),
		"savings":         atomic.LoadInt64(&cc.stats.savings),
	}
}