feat(frac): new active fraction implementation based in mini-indexes

asyncsearcher/async_searcher_test.go

@@ -9,19 +9,18 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/ozontech/seq-db/frac"
-	"github.com/ozontech/seq-db/frac/common"
 	"github.com/ozontech/seq-db/frac/processor"
 	"github.com/ozontech/seq-db/mappingprovider"
 	"github.com/ozontech/seq-db/seq"
 )
 
 type fakeFrac struct {
 	frac.Fraction
-	info common.Info
+	info frac.Info
 	dp   fakeDP
 }
 
-func (f *fakeFrac) Info() *common.Info {
+func (f *fakeFrac) Info() *frac.Info {
 	return &f.info
 }
 
@@ -51,7 +50,7 @@ func TestAsyncSearcherMaintain(t *testing.T) {
 		Retention: time.Hour,
 	}
 	fracs := []frac.Fraction{
-		&fakeFrac{info: common.Info{Path: "1"}},
+		&fakeFrac{info: frac.Info{Path: "1"}},
 	}
 	r.NoError(as.StartSearch(req, fracs))
 

benchmarks/docker-compose-seqdb.yml

@@ -7,7 +7,7 @@ services:
         limits:
           cpus: "4"
           memory: "8GB"
-    image: ghcr.io/ozontech/seq-db:v0.61.0
+          image: 'gitlab-registry.ozon.ru/sre/images/seq-db:che@sha256:82d0dd34cb5d6db9e0450bc8d2cd1d9e29414ec2ba81dc8c4ae643dea6eb1bd0'
     ports:
       - '9002:9002'
     volumes:

cmd/distribution/main.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/ozontech/seq-db/cache"
 	"github.com/ozontech/seq-db/consts"
-	"github.com/ozontech/seq-db/frac/common"
+	"github.com/ozontech/seq-db/frac"
 	"github.com/ozontech/seq-db/frac/sealed"
 	"github.com/ozontech/seq-db/fracmanager"
 	"github.com/ozontech/seq-db/logger"
@@ -59,7 +59,7 @@ func readBlock(reader storage.IndexReader, blockIndex uint32) ([]byte, error) {
 	return data, nil
 }
 
-func loadInfo(path string) *common.Info {
+func loadInfo(path string) *frac.Info {
 	indexReader, f := getReader(path)
 	defer f.Close()
 
@@ -87,7 +87,7 @@ func loadInfo(path string) *common.Info {
 	return b.Info
 }
 
-func buildDist(dist *seq.MIDsDistribution, path string, _ *common.Info) {
+func buildDist(dist *seq.MIDsDistribution, path string, _ *frac.Info) {
 	blocksReader, f := getReader(path)
 	defer f.Close()
 

cmd/seq-db/seq-db.go

@@ -23,7 +23,6 @@ import (
 	"github.com/ozontech/seq-db/config"
 	"github.com/ozontech/seq-db/consts"
 	"github.com/ozontech/seq-db/frac"
-	"github.com/ozontech/seq-db/frac/common"
 	"github.com/ozontech/seq-db/fracmanager"
 	"github.com/ozontech/seq-db/logger"
 	"github.com/ozontech/seq-db/mappingprovider"
@@ -259,7 +258,8 @@ func startStore(
 			MaintenanceDelay:  0,
 			CacheGCDelay:      0,
 			CacheCleanupDelay: 0,
-			SealParams: common.SealParams{
+			MinSealFracSize:   uint64(cfg.Storage.TotalSize) * consts.DefaultMinSealPercent / 100,
+			SealParams: frac.SealParams{
 				IDsZstdLevel:           cfg.Compression.SealedZstdCompressionLevel,
 				LIDsZstdLevel:          cfg.Compression.SealedZstdCompressionLevel,
 				TokenListZstdLevel:     cfg.Compression.SealedZstdCompressionLevel,

config/shared.go

@@ -8,7 +8,7 @@ var (
 	ReaderWorkers int
 
 	CaseSensitive = false
-	SkipFsync     = false
+	SkipFsync     = true
 
 	MaxFetchSizeBytes = 4 * units.MiB
 

frac/active/active.go

@@ -0,0 +1,294 @@
+package active
+
+import (
+	"context"
+	"io"
+	"os"
+	"sync"
+	"time"
+
+	"github.com/ozontech/seq-db/cache"
+	"github.com/ozontech/seq-db/config"
+	"github.com/ozontech/seq-db/consts"
+	"github.com/ozontech/seq-db/frac"
+	"github.com/ozontech/seq-db/frac/active_old"
+	"github.com/ozontech/seq-db/frac/processor"
+	"github.com/ozontech/seq-db/logger"
+	"github.com/ozontech/seq-db/metric"
+	"github.com/ozontech/seq-db/metric/stopwatch"
+	"github.com/ozontech/seq-db/seq"
+	"github.com/ozontech/seq-db/storage"
+	"github.com/ozontech/seq-db/util"
+	"go.uber.org/zap"
+)
+
+type Active struct {
+	Config *frac.Config
+
+	BaseFileName string
+
+	indexer *Indexer
+
+	indexes *memIndexPool
+	merger  *mergeManager
+
+	docsFile   *os.File
+	docsReader storage.DocsReader
+	sortReader storage.DocsReader
+	docsCache  *cache.Cache[[]byte]
+	sortCache  *cache.Cache[[]byte]
+
+	metaFile   *os.File
+	metaReader storage.DocBlocksReader
+
+	writer *active_old.Writer
+}
+
+func New(
+	baseFileName string,
+	cfg *frac.Config,
+	workers int,
+	readLimiter *storage.ReadLimiter,
+	docsCache *cache.Cache[[]byte],
+	sortCache *cache.Cache[[]byte],
+) *Active {
+	docsFile, docsStats := util.MustOpenFile(baseFileName+consts.DocsFileSuffix, config.SkipFsync)
+	metaFile, metaStats := util.MustOpenFile(baseFileName+consts.MetaFileSuffix, config.SkipFsync)
+
+	info := frac.NewInfo(baseFileName, uint64(docsStats.Size()), uint64(metaStats.Size()))
+	indexes := NewIndexPool(info)
+
+	f := &Active{
+		BaseFileName: baseFileName,
+		Config:       cfg,
+		indexer:      NewIndexer(util.NewSemaphore(workers)),
+		merger:       newMergeManager(indexes, util.NewSemaphore(workers)),
+		indexes:      indexes,
+
+		docsFile:   docsFile,
+		docsCache:  docsCache,
+		sortCache:  sortCache,
+		docsReader: storage.NewDocsReader(readLimiter, docsFile, docsCache),
+		sortReader: storage.NewDocsReader(readLimiter, docsFile, sortCache),
+
+		metaFile:   metaFile,
+		metaReader: storage.NewDocBlocksReader(readLimiter, metaFile),
+
+		writer: active_old.NewWriter(docsFile, metaFile, docsStats.Size(), metaStats.Size(), config.SkipFsync),
+	}
+
+	logger.Info("active fraction created", zap.String("fraction", baseFileName))
+
+	return f
+}
+
+func (f *Active) Replay(ctx context.Context) error {
+
+	info := f.indexes.info
+
+	logger.Info("start replaying...", zap.String("name", info.Name()))
+
+	t := time.Now()
+
+	offset := uint64(0)
+	step := info.MetaOnDisk / 10
+	wg := sync.WaitGroup{}
+	next := step
+
+out:
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+			meta, metaSize, err := f.metaReader.ReadDocBlock(int64(offset))
+			if err == io.EOF {
+				if metaSize != 0 {
+					logger.Warn("last meta block is partially written, skipping it")
+				}
+				break out
+			}
+			if err != nil && err != io.EOF {
+				return err
+			}
+
+			if offset > next {
+				next += step
+				progress := float64(offset) / float64(info.MetaOnDisk) * 100
+				logger.Info("replaying batch, meta",
+					zap.String("name", info.Name()),
+					zap.Uint64("from", offset),
+					zap.Uint64("to", offset+metaSize),
+					zap.Uint64("target", info.MetaOnDisk),
+					util.ZapFloat64WithPrec("progress_percentage", progress, 2),
+				)
+			}
+			offset += metaSize
+
+			wg.Add(1)
+			f.indexer.Index(meta, func(idx *memIndex, err error) {
+				f.AddIndex(idx, 0, 0, err)
+				wg.Done()
+			})
+		}
+	}
+
+	wg.Wait()
+
+	tookSeconds := util.DurationToUnit(time.Since(t), "s")
+	throughputRaw := util.SizeToUnit(info.DocsRaw, "mb") / tookSeconds
+	throughputMeta := util.SizeToUnit(info.MetaOnDisk, "mb") / tookSeconds
+	logger.Info("active fraction replayed",
+		zap.String("name", info.Name()),
+		zap.Uint32("docs_total", info.DocsTotal),
+		util.ZapUint64AsSizeStr("docs_size", info.DocsOnDisk),
+		util.ZapFloat64WithPrec("took_s", tookSeconds, 1),
+		util.ZapFloat64WithPrec("throughput_raw_mb_sec", throughputRaw, 1),
+		util.ZapFloat64WithPrec("throughput_meta_mb_sec", throughputMeta, 1),
+	)
+	return nil
+}
+
+func (f *Active) Append(docs, meta []byte, wg *sync.WaitGroup) (err error) {
+	sw := stopwatch.New()
+	ma := sw.Start("append")
+	if err = f.writer.Write(docs, meta, sw); err != nil {
+		ma.Stop()
+		return err
+	}
+
+	mi := sw.Start("send_to_indexer")
+
+	f.indexer.Index(meta, func(idx *memIndex, err error) {
+		f.AddIndex(idx, uint64(len(docs)), uint64(len(meta)), err)
+		wg.Done()
+	})
+
+	mi.Stop()
+
+	ma.Stop()
+	sw.Export(bulkStagesSeconds)
+	return nil
+}
+
+func (f *Active) AddIndex(idx *memIndex, docsLen, metaLen uint64, err error) {
+	if err != nil {
+		logger.Fatal("bulk indexing error", zap.Error(err))
+	}
+	f.indexes.Add(idx, docsLen, metaLen)
+	f.merger.requestMerge()
+}
+
+func (f *Active) String() string {
+	return frac.FracToString(f, "active")
+}
+
+func (f *Active) Fetch(ctx context.Context, ids []seq.ID) ([][]byte, error) {
+	sw := stopwatch.New()
+	defer sw.Export(fetcherStagesSec)
+
+	t := sw.Start("total")
+
+	ss, release := f.indexes.Snapshot()
+	defer release()
+
+	if ss.info.DocsTotal == 0 { // it is empty active fraction state
+		return nil, nil
+	}
+
+	res := make([][]byte, len(ids))
+	for _, index := range ss.indexes {
+		fetchIndex := fetchIndex{index: index, docsReader: &f.docsReader}
+		if err := processor.IndexFetch(ids, sw, &fetchIndex, res); err != nil {
+			return nil, err
+		}
+	}
+	t.Stop()
+
+	return res, nil
+}
+
+func (f *Active) Search(ctx context.Context, params processor.SearchParams) (*seq.QPR, error) {
+	ss, release := f.indexes.Snapshot()
+	defer release()
+
+	if ss.info.DocsTotal == 0 { // it is empty active fraction state
+		metric.CountersTotal.WithLabelValues("empty_data_provider").Inc()
+		return &seq.QPR{Aggs: make([]seq.AggregatableSamples, len(params.AggQ))}, nil
+	}
+
+	aggLimits := processor.AggLimits(f.Config.Search.AggLimits)
+
+	// Limit the parameter range to data boundaries to prevent histogram overflow
+	params.From = max(params.From, ss.info.From)
+	params.To = min(params.To, ss.info.To)
+
+	sw := stopwatch.New()
+	defer sw.Export(getActiveSearchMetric(params))
+
+	t := sw.Start("total")
+	qprs := make([]*seq.QPR, 0, len(ss.indexes))
+	for _, index := range ss.indexes {
+		si := searchIndex{ctx: ctx, index: index}
+		qpr, err := processor.IndexSearch(ctx, params, &si, aggLimits, sw)
+		if err != nil {
+			return nil, err
+		}
+		qprs = append(qprs, qpr)
+	}
+	res := processor.MergeQPRs(qprs, params)
+	res.IDs.ApplyHint(ss.info.Name())
+	t.Stop()
+
+	return res, nil
+}
+
+func (f *Active) Info() *frac.Info {
+	return f.indexes.Info()
+}
+
+func (f *Active) Contains(id seq.MID) bool {
+	return f.Info().IsIntersecting(id, id)
+}
+
+func (f *Active) IsIntersecting(from, to seq.MID) bool {
+	return f.Info().IsIntersecting(from, to)
+}
+
+func (f *Active) Release() {
+	f.releaseMem()
+
+	if !f.Config.KeepMetaFile {
+		util.RemoveFile(f.metaFile.Name())
+	}
+
+	if !f.Config.SkipSortDocs {
+		// we use sorted docs in sealed fraction so we can remove original docs of active fraction
+		util.RemoveFile(f.docsFile.Name())
+	}
+
+}
+
+func (f *Active) Suicide() {
+	f.releaseMem()
+
+	util.RemoveFile(f.metaFile.Name())
+	util.RemoveFile(f.docsFile.Name())
+	util.RemoveFile(f.BaseFileName + consts.SdocsFileSuffix)
+}
+
+func (f *Active) releaseMem() {
+	f.writer.Stop()
+	f.merger.Stop()
+	f.indexes.Release()
+
+	f.docsCache.Release()
+	f.sortCache.Release()
+
+	if err := f.metaFile.Close(); err != nil {
+		logger.Error("can't close meta file", zap.String("frac", f.BaseFileName), zap.Error(err))
+	}
+	if err := f.docsFile.Close(); err != nil {
+		logger.Error("can't close docs file", zap.String("frac", f.BaseFileName), zap.Error(err))
+	}
+}