feat: rotate fraction based on in-memory size

.seqbench/comparison.env

@@ -1,6 +1,6 @@
 GOGC=100
 
-SEQDB_STORAGE_FRAC_SIZE=16MiB
+SEQDB_STORAGE_FRAC_SIZE=1MiB
 SEQDB_STORAGE_TOTAL_SIZE=10GiB
 
 SEQDB_LIMITS_QUERY_RATE=1024

Makefile

@@ -56,7 +56,7 @@ test-deps:
 
 .PHONY: test
 test: test-deps
-	LOG_LEVEL=ERROR go test ./... -count 1
+	LOG_LEVEL=ERROR go test ./... -count 1 -v
 
 .bin-deps: export GOBIN := $(LOCAL_BIN)
 .bin-deps:

asyncsearcher/async_searcher.go

@@ -74,11 +74,11 @@ type AsyncSearcherConfig struct {
 }
 
 type fractionAcquirer interface {
-	Fractions() fracmanager.List
+	AcquireFractions() (_ fracmanager.List, release func())
 	AcquireFraction(name string) (_ frac.Fraction, release func(), ok bool)
 }
 
-func MustStartAsync(config AsyncSearcherConfig, mp MappingProvider, fracs fractionAcquirer) *AsyncSearcher {
+func MustStartAsync(config AsyncSearcherConfig, mp MappingProvider, fracProvider fractionAcquirer) *AsyncSearcher {
 	if config.DataDir == "" {
 		logger.Fatal("can't start async searcher: DataDir is empty")
 	}
@@ -107,7 +107,7 @@ func MustStartAsync(config AsyncSearcherConfig, mp MappingProvider, fracs fracti
 	for _, id := range notProcessedIDs {
 		asyncSearchActiveSearches.Add(1)
 		as.processWg.Add(1)
-		go as.processRequest(id, fracs)
+		go as.processRequest(id, fracProvider)
 	}
 
 	// set limit metrics that allow us to calculate alerts' thresholds
@@ -209,7 +209,7 @@ func (i *asyncSearchInfo) Status() AsyncSearchStatus {
 	return status
 }
 
-func (as *AsyncSearcher) StartSearch(r AsyncSearchRequest, fracs fractionAcquirer) error {
+func (as *AsyncSearcher) StartSearch(r AsyncSearchRequest, fracProvider fractionAcquirer) error {
 	if as.readOnly.Load() {
 		return fmt.Errorf("cannot start search on read-only mode")
 	}
@@ -240,14 +240,17 @@ func (as *AsyncSearcher) StartSearch(r AsyncSearchRequest, fracs fractionAcquire
 		return fmt.Errorf("retention time should be less than %s, got %s", maxRetention, r.Retention)
 	}
 
-	fracNames := fracs.Fractions().FilterInRange(r.Params.From, r.Params.To).Names()
+	fracs, release := fracProvider.AcquireFractions()
+	defer release()
+
+	fracNames := fracs.FilterInRange(r.Params.From, r.Params.To).Names()
 	if ok := as.saveSearchInfo(r, fracNames); !ok {
 		// Request was saved previously, skip it
 		return nil
 	}
 	asyncSearchActiveSearches.Add(1)
 	as.processWg.Add(1)
-	go as.processRequest(r.ID, fracs)
+	go as.processRequest(r.ID, fracProvider)
 	return nil
 }
 
@@ -301,17 +304,17 @@ func (as *AsyncSearcher) createDataDir() {
 	})
 }
 
-func (as *AsyncSearcher) processRequest(asyncSearchID string, fracs fractionAcquirer) {
+func (as *AsyncSearcher) processRequest(asyncSearchID string, fracProvider fractionAcquirer) {
 	defer as.processWg.Done()
 
 	as.rateLimit <- struct{}{}
 	defer func() { <-as.rateLimit }()
 
-	as.doSearch(asyncSearchID, fracs)
+	as.doSearch(asyncSearchID, fracProvider)
 	asyncSearchActiveSearches.Add(-1)
 }
 
-func (as *AsyncSearcher) doSearch(id string, fracs fractionAcquirer) {
+func (as *AsyncSearcher) doSearch(id string, fracProvider fractionAcquirer) {
 	qprPaths, err := as.findQPRs(id)
 	if err != nil {
 		panic(fmt.Errorf("can't find QPRs for id %q: %s", id, err))
@@ -353,7 +356,7 @@ func (as *AsyncSearcher) doSearch(id string, fracs fractionAcquirer) {
 		if as.shouldStopSearch(id) {
 			break
 		}
-		if err := as.acquireAndProcessFrac(fracInfo, info, fracs); err != nil {
+		if err := as.acquireAndProcessFrac(fracInfo, info, fracProvider); err != nil {
 			as.updateSearchInfo(id, func(info *asyncSearchInfo) {
 				info.Error = err.Error()
 			})
@@ -395,8 +398,8 @@ func compressQPR(qpr *seq.QPR, cb func(compressed []byte) error) error {
 	return nil
 }
 
-func (as *AsyncSearcher) acquireAndProcessFrac(fracInfo fracSearchState, searchInfo asyncSearchInfo, fracs fractionAcquirer) (err error) {
-	f, release, ok := fracs.AcquireFraction(fracInfo.Name)
+func (as *AsyncSearcher) acquireAndProcessFrac(fracInfo fracSearchState, searchInfo asyncSearchInfo, fracProvider fractionAcquirer) (err error) {
+	f, release, ok := fracProvider.AcquireFraction(fracInfo.Name)
 	if !ok { // oldest fracs may already be removed
 		logger.Info(
 			"async search: skip missing fraction",

asyncsearcher/async_searcher_test.go

@@ -49,8 +49,8 @@ func (fp fakeFractionProvider) AcquireFraction(name string) (frac.Fraction, func
 	return nil, func() {}, false
 }
 
-func (fp fakeFractionProvider) Fractions() fracmanager.List {
-	return fracmanager.List(fp)
+func (fp fakeFractionProvider) AcquireFractions() (fracmanager.List, func()) {
+	return fracmanager.List(fp), func() {}
 }
 
 func TestAsyncSearcherMaintain(t *testing.T) {

compaction/executor.go

@@ -0,0 +1,65 @@
+package compaction
+
+import (
+	"sync"
+
+	"go.uber.org/zap"
+
+	"github.com/ozontech/seq-db/frac"
+	"github.com/ozontech/seq-db/frac/common"
+	"github.com/ozontech/seq-db/frac/sealed"
+	"github.com/ozontech/seq-db/logger"
+)
+
+type Executor struct {
+	workers int
+	wg      sync.WaitGroup
+	p       *planner
+}
+
+// FIXME(dkharms): I need to pass here [common.SealParams].
+func NewExecutor(workers int, p *planner) *Executor {
+	e := Executor{workers: workers, p: p}
+	e.init()
+	return &e
+}
+
+func (e *Executor) Close() {
+	e.p.close()
+	e.wg.Wait()
+}
+
+func (e *Executor) init() {
+	for range e.workers {
+		e.wg.Go(func() {
+			for t := range e.p.tasks {
+				logger.Info(
+					"got new compaction task",
+					zap.Time("bin", t.bin),
+					zap.Any("snapshot", t.snapshot),
+				)
+				t.onComplete(e.compact(t))
+			}
+		})
+	}
+}
+
+func (e *Executor) compact(t task) (*sealed.PreloadedData, error) {
+	var (
+		names []string
+		srcs  []Source
+	)
+
+	for _, f := range t.snapshot.Fractions() {
+		names = append(names, f.Info().Name())
+		srcs = append(srcs, frac.NewSealedSource(f))
+	}
+
+	logger.Info(
+		"compacting fractions",
+		zap.Strings("names", names),
+	)
+
+	preloaded, err := Merge(t.filename, common.SealParams{}, srcs...)
+	return preloaded, err
+}

compaction/merge.go

@@ -0,0 +1,162 @@
+package compaction
+
+import (
+	"errors"
+	"os"
+
+	"github.com/ozontech/seq-db/consts"
+	"github.com/ozontech/seq-db/frac/common"
+	"github.com/ozontech/seq-db/frac/sealed"
+	"github.com/ozontech/seq-db/indexwriter"
+)
+
+func Merge(filename string, params common.SealParams, srcs ...Source) (*sealed.PreloadedData, error) {
+	writer := indexwriter.New(params)
+	src := NewMergeSource(filename, srcs)
+
+	if err := createAndWrite(
+		filename+consts.OffsetsTmpFileSuffix,
+		filename+consts.OffsetsFileSuffix,
+		func(f *os.File) error { return writer.WriteOffsetsFile(f, src) },
+	); err != nil {
+		return nil, err
+	}
+
+	if err := createAndWrite(
+		filename+consts.IDTmpFileSuffix,
+		filename+consts.IDFileSuffix,
+		func(f *os.File) error { return writer.WriteIDFile(f, src) },
+	); err != nil {
+		return nil, err
+	}
+
+	if err := createAndWriteBoth(
+		filename+consts.TokenTmpFileSuffix,
+		filename+consts.TokenFileSuffix,
+		filename+consts.LIDTmpFileSuffix,
+		filename+consts.LIDFileSuffix,
+		func(tf, lf *os.File) error { return writer.WriteTokenTriplet(tf, lf, src) },
+	); err != nil {
+		return nil, err
+	}
+
+	if err := createAndWrite(
+		filename+consts.InfoTmpFileSuffix,
+		filename+consts.InfoFileSuffix,
+		func(f *os.File) error { return writer.WriteInfoFile(f, src) },
+	); err != nil {
+		return nil, err
+	}
+
+	if err := mergeDocs(filename, srcs...); err != nil {
+		return nil, err
+	}
+
+	info := src.Info()
+	info.IndexOnDisk = 0
+
+	for _, suffix := range []string{
+		consts.InfoFileSuffix,
+		consts.TokenFileSuffix,
+		consts.OffsetsFileSuffix,
+		consts.IDFileSuffix,
+		consts.LIDFileSuffix,
+	} {
+		st, err := os.Stat(info.Path + suffix)
+		if err != nil {
+			return nil, err
+		}
+		info.IndexOnDisk += uint64(st.Size())
+	}
+
+	lidsTable := writer.LIDsTable()
+	preloaded := &sealed.PreloadedData{
+		Info:       info,
+		TokenTable: writer.TokenTable(),
+		BlocksData: sealed.BlocksData{
+			LIDsTable:     &lidsTable,
+			IDsTable:      writer.IDsTable(),
+			BlocksOffsets: src.BlockOffsets(),
+		},
+	}
+
+	return preloaded, nil
+}
+
+func mergeDocs(filename string, srcs ...Source) error {
+	return createAndWrite(
+		filename+consts.DocsTmpFileSuffix,
+		filename+consts.DocsFileSuffix,
+		func(f *os.File) error {
+			var docsSize uint64
+			for _, src := range srcs {
+				for loc, err := range src.DocBlock() {
+					if err != nil {
+						return err
+					}
+
+					payload, offset := loc.First, loc.Second
+					if _, err := f.WriteAt(payload, int64(offset+docsSize)); err != nil {
+						return err
+					}
+				}
+
+				docsSize += src.Info().DocsOnDisk
+			}
+
+			return nil
+		},
+	)
+}
+
+func syncAndClose(f *os.File) error {
+	if err := f.Sync(); err != nil {
+		f.Close()
+		return err
+	}
+	return f.Close()
+}
+
+func createAndWrite(
+	tmp, final string,
+	write func(*os.File) error,
+) error {
+	f, err := os.Create(tmp)
+	if err != nil {
+		return err
+	}
+
+	if err := errors.Join(write(f), syncAndClose(f)); err != nil {
+		return err
+	}
+
+	return os.Rename(tmp, final)
+}
+
+func createAndWriteBoth(
+	atmp, afinal,
+	btmp, bfinal string,
+	write func(*os.File, *os.File) error,
+) error {
+	a, err := os.Create(atmp)
+	if err != nil {
+		return err
+	}
+
+	b, err := os.Create(btmp)
+	if err != nil {
+		a.Close()
+		return err
+	}
+
+	writeErr := write(a, b)
+	if err := errors.Join(writeErr, syncAndClose(a), syncAndClose(b)); err != nil {
+		return err
+	}
+
+	if err := os.Rename(atmp, afinal); err != nil {
+		return err
+	}
+
+	return os.Rename(btmp, bfinal)
+}