diff --git a/compaction/merge.go b/compaction/merge.go index ee6f6d88..635a1bf2 100644 --- a/compaction/merge.go +++ b/compaction/merge.go @@ -1,6 +1,7 @@ package compaction import ( + "encoding/json" "errors" "os" @@ -14,6 +15,10 @@ func Merge(filename string, params common.SealParams, srcs ...Source) (*sealed.P w := indexwriter.New(params) src := NewMergeSource(filename, srcs) + if err := createCompactionPlan(filename, srcs...); err != nil { + return nil, err + } + if err := createAndWrite( filename+consts.OffsetsTmpFileSuffix, filename+consts.OffsetsFileSuffix, @@ -83,6 +88,37 @@ func Merge(filename string, params common.SealParams, srcs ...Source) (*sealed.P return preloaded, nil } +func createCompactionPlan(filename string, srcs ...Source) error { + type plan struct { + Participants []string `json:"participants"` + } + + // Write which fractions are participating in compaction. + // + // Since we cannot remove all stale fractions and move + // the merged one into final state, we have to keep our + // intention on disk. + // + // Later on, it will help us to make correct decision + // whether we should drop participants (delete them from disk). + return createAndWrite( + filename+consts.CompactionPlan, + filename+consts.CompactionPlan, + func(f *os.File) error { + var p plan + + for i := range srcs { + p.Participants = append( + p.Participants, + srcs[i].Info().Name(), + ) + } + + return json.NewEncoder(f).Encode(p) + }, + ) +} + func mergeDocs(filename string, srcs ...Source) error { return createAndWrite( filename+consts.DocsTmpFileSuffix, diff --git a/compaction/merge_source.go b/compaction/merge_source.go index 00e6e708..20d0bd7b 100644 --- a/compaction/merge_source.go +++ b/compaction/merge_source.go @@ -69,11 +69,13 @@ func (s *MergeSource) prepareInfo() *common.Info { info := common.NewInfo(s.filename, 0, 0) var ( - from seq.MID = seq.MaxID.MID - to seq.MID = seq.MinID.MID + from seq.MID = seq.MaxID.MID + to seq.MID = seq.MinID.MID + creation uint64 = info.CreationTime ) for _, src := range s.sources { + creation = min(creation, src.Info().CreationTime) from = min(from, src.Info().From) to = max(to, src.Info().To) } @@ -81,6 +83,10 @@ func (s *MergeSource) prepareInfo() *common.Info { info.From, info.To = from, to info.SealingTime = info.CreationTime + // TODO(dkharms): This is workaround which will + // force compacted fraction to stay in the same bin. + info.CreationTime = creation + info.InitEmptyDistribution() return info } diff --git a/compaction/merge_source_test.go b/compaction/merge_source_test.go index c4cb0fcc..c9ea83be 100644 --- a/compaction/merge_source_test.go +++ b/compaction/merge_source_test.go @@ -7,6 +7,7 @@ import ( "math/rand" "slices" "testing" + "time" "github.com/stretchr/testify/require" @@ -29,6 +30,8 @@ func (m *mockSealingSource) Info() *common.Info { DocsTotal: uint32(len(m.ids)), DocsOnDisk: m.docsOnDisk, + CreationTime: uint64(time.Now().UnixMilli()), + From: slices.MinFunc(m.ids, func(x, y seq.ID) int { return cmp.Compare(x.MID, y.MID) }).MID, diff --git a/compaction/planner.go b/compaction/planner.go index c44b80ed..a76bca63 100644 --- a/compaction/planner.go +++ b/compaction/planner.go @@ -11,6 +11,7 @@ import ( "go.uber.org/zap" + "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/fracmanager" @@ -183,10 +184,13 @@ func (p *planner) pick() (task, bool) { } compactionResultTotal.WithLabelValues(bucketSize, "success").Inc() - // TODO(dkharms): Is it fine to substitute and delete? - // We need somehow substitute and delete atomically. + p.fm.SubstituteWithSealed(s, csnapshot) csnapshot.Destroy() + + // We have destroyed all sealed fractions which participated + // in compaction and now stale. So we can drop compaction plan. + util.RemoveFile(s.Info.Path + consts.CompactionPlan) }, }, true } diff --git a/config.example.yaml b/config.example.yaml index e4cf0a9a..1491645b 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -7,7 +7,7 @@ address: debug: :9200 storage: - frac_size: 16MiB + frac_size: 1MiB total_size: 1GiB compaction: diff --git a/consts/consts.go b/consts/consts.go index 3341aecd..422f229e 100644 --- a/consts/consts.go +++ b/consts/consts.go @@ -88,6 +88,7 @@ const ( RemoteFractionSuffix = ".remote" FracCacheFileSuffix = ".frac-cache" + CompactionPlan = ".compaction-plan" // tracing JaegerDebugKey = "jaeger-debug-id" diff --git a/fracmanager/frac_manifest.go b/fracmanager/frac_manifest.go index ae4adc4a..f899bacc 100644 --- a/fracmanager/frac_manifest.go +++ b/fracmanager/frac_manifest.go @@ -1,9 +1,13 @@ package fracmanager import ( + "encoding/json" "errors" "fmt" + "maps" + "os" "path/filepath" + "slices" "sort" "strings" @@ -37,6 +41,8 @@ type fracManifest struct { hasDocsDel bool // documents deletion marker hasSdocsDel bool // sorted documents deletion marker hasIndexDel bool // index deletion marker + + hasCompactionPlan bool } // hasAllIndexFiles reports whether all 5 split index files are present. @@ -79,6 +85,9 @@ func (m *fracManifest) AddExtension(ext string) error { case consts.IndexDelFileSuffix: m.hasIndexDel = true + case consts.CompactionPlan: + m.hasCompactionPlan = true + case consts.IndexTmpFileSuffix, consts.InfoTmpFileSuffix, consts.TokenTmpFileSuffix, consts.OffsetsTmpFileSuffix, consts.IDTmpFileSuffix, consts.LIDTmpFileSuffix, @@ -148,6 +157,11 @@ func removeMeta(m *fracManifest) { } } +func removeCompactionPlan(m *fracManifest) { + util.RemoveFile(m.basePath + consts.CompactionPlan) + m.hasCompactionPlan = false +} + func removeIndexFiles(m *fracManifest) { for _, suffix := range []string{ consts.InfoFileSuffix, @@ -233,7 +247,14 @@ func analyzeFiles(files []string) ([]*fracManifest, error) { // filterValid filters valid fractions and handles invalid ones // Removes partially deleted and unknown fractions func filterValid(ids []string, manifests map[string]*fracManifest) ([]*fracManifest, error) { + // We need to drop stale (compacted) fractions first. + ids, err := dropCompacted(ids, manifests) + if err != nil { + return nil, err + } + validated := make([]*fracManifest, 0, len(manifests)) + for _, id := range ids { manifest := manifests[id] if manifest == nil { @@ -242,6 +263,16 @@ func filterValid(ids []string, manifests map[string]*fracManifest) ([]*fracManif switch manifest.Stage() { case fracStageUnknown: + // Processing partially compacted fraction. + if manifest.hasCompactionPlan { + logger.Warn( + "dropping partially compacted fraction", + zap.String("base_path", manifest.basePath), + ) + removeAllFiles(manifest.basePath) + continue + } + logger.Error("unknown fraction stage", zap.Object("manifest", manifest)) fractionLoadErrors.Inc() removeAllFiles(manifest.basePath) @@ -256,9 +287,80 @@ func filterValid(ids []string, manifests map[string]*fracManifest) ([]*fracManif cleanupFrac(manifest) validated = append(validated, manifest) } + return validated, nil } +func dropCompacted(ids []string, manifests map[string]*fracManifest) ([]string, error) { + type plan struct { + Participants []string `json:"participants"` + } + + filtered := make(map[string]struct{}) + + for _, id := range ids { + filtered[id] = struct{}{} + } + + for _, id := range ids { + m := manifests[id] + if m == nil { + return nil, errors.New("inconsistent fraction file analysis") + } + + skip := !m.hasCompactionPlan || + m.Stage() != fracStageSealed + + if skip { + continue + } + + f, err := os.Open(m.basePath + consts.CompactionPlan) + if err != nil { + return nil, err + } + defer f.Close() //nolint + + var p plan + if err := json.NewDecoder(f).Decode(&p); err != nil { + // Well, we cannot decode compaction plan so let's drop whatever the result is. + + logger.Warn( + "dropping possibly correctly compacted fraction: cannot decode compaction plan", + zap.Error(err), + zap.String("base_path", m.basePath), + ) + + delete(filtered, id) + removeAllFiles(m.basePath) + continue + } + + for _, pname := range p.Participants { + pid := pname[len(fileBasePattern):] + + pm := manifests[pid] + if pm == nil { + // NOTE(dkharms): It is possible that compaction participants + // were dropped but the plan itself was not deleted. + continue + } + + logger.Warn( + "dropping fraction: it was merged into another one", + zap.Error(err), + zap.String("merged_base_path", m.basePath), + zap.String("participant_base_path", pm.basePath), + ) + + delete(filtered, pid) + removeAllFiles(pm.basePath) + } + } + + return slices.Collect(maps.Keys(filtered)), nil +} + // cleanupFrac performs cleanup of unnecessary files depending on fraction stage // Called after stage determination to optimize storage func cleanupFrac(m *fracManifest) { @@ -284,6 +386,7 @@ func cleanupRemoteFrac(m *fracManifest) { // Removes redundant files after finishing work with the fraction func cleanupSealedFrac(m *fracManifest) { removeMeta(m) + removeCompactionPlan(m) if m.hasSdocs { removeDocs(m) // remove orig docs, but keeping sorted } @@ -315,6 +418,7 @@ func removeAllFiles(basePath string) { consts.MetaFileSuffix, consts.WalFileSuffix, + consts.CompactionPlan, } { util.RemoveFile(basePath + suffix) }