From c13d7fce36f639a85b2d0dcf68c7a9df701a6ae5 Mon Sep 17 00:00:00 2001 From: Daniil Porokhnin Date: Fri, 10 Jul 2026 15:51:31 +0300 Subject: [PATCH] chore: remove compatibility with `.meta` file --- cmd/seq-db/seq-db.go | 2 +- consts/consts.go | 3 +- docs/en/13-architecture.md | 4 +- docs/ru/13-architecture.md | 4 +- frac/active.go | 126 +++++------------------------------ frac/active_writer.go | 21 ++---- frac/config.go | 2 +- frac/fraction_test.go | 2 +- frac/legacy_meta_writer.go | 25 ------- fracmanager/frac_manifest.go | 19 ++---- storage/wal_block.go | 37 ---------- storage/wal_block_test.go | 36 ---------- storeapi/grpc_bulk.go | 6 -- 13 files changed, 35 insertions(+), 252 deletions(-) delete mode 100644 frac/legacy_meta_writer.go diff --git a/cmd/seq-db/seq-db.go b/cmd/seq-db/seq-db.go index 25b65f329..738febdcb 100644 --- a/cmd/seq-db/seq-db.go +++ b/cmd/seq-db/seq-db.go @@ -284,7 +284,7 @@ func startStore( }, }, SkipSortDocs: !cfg.DocsSorting.Enabled, - KeepMetaFile: false, + KeepWalFile: false, }, OffloadingEnabled: cfg.Offloading.Enabled, OffloadingRetention: cfg.Offloading.Retention, diff --git a/consts/consts.go b/consts/consts.go index 3341aecd9..1aeb1fb11 100644 --- a/consts/consts.go +++ b/consts/consts.go @@ -52,8 +52,7 @@ const ( BrokenDir = ".broken" // known extensions - MetaFileSuffix = ".meta" - WalFileSuffix = ".wal" + WalFileSuffix = ".wal" DocsFileSuffix = ".docs" DocsTmpFileSuffix = "._docs" diff --git a/docs/en/13-architecture.md b/docs/en/13-architecture.md index 37967c2b7..cfd7af65d 100644 --- a/docs/en/13-architecture.md +++ b/docs/en/13-architecture.md @@ -24,7 +24,7 @@ seq-db store keeps all document data in three file types: | File type | Purpose | |-----------|------------------------------------------------| | `.docs` | Stores compressed batches of raw log documents | -| `.meta` | Tokenized metadata stream (used for recovery) | +| `.wal` | Tokenized metadata stream (used for recovery) | | `.index` | On-disk inverted index | @@ -38,7 +38,7 @@ Read more about file types and their internal structure [here](internal/fraction A write operation is acknowledged only after the payload is safely persisted: ``` -write, fsync # .meta file +write, fsync # .wal file write, fsync # .data file ``` That is, two write system calls followed by two fsync diff --git a/docs/ru/13-architecture.md b/docs/ru/13-architecture.md index dc6fa7dfc..7f87f9d42 100644 --- a/docs/ru/13-architecture.md +++ b/docs/ru/13-architecture.md @@ -21,7 +21,7 @@ seq-db store хранит все данные документов в трех | Тип файла | Назначение | |-----------|--------------------------------------------------------| | `.docs` | Хранит сжатые батчи сырых документов логов | -| `.meta` | Токенизированный поток метаданных (для восстановления) | +| `.wal` | Токенизированный поток метаданных (для восстановления) | | `.index` | Дисковый обратный индекс. | Поскольку набор данных хранится в этих трех типах файлов, перемещение или восстановление шарда выполняется просто: достаточно скопировать (`cp` / `rsync`) директорию на целевой узел и запустить под. @@ -32,7 +32,7 @@ seq-db store хранит все данные документов в трех Операция записи подтверждается только после того, как полезная нагрузка гарантированно сохранена: ``` -write, fsync # файл .meta +write, fsync # файл .wal write, fsync # файл .data ``` То есть, два системных вызова `write`, за которыми следуют два вызова `fsync` — это гарантирует, что данные переживут аварию узла или его перезапуск до того, как клиент получит ответ об успехе. diff --git a/frac/active.go b/frac/active.go index 2907d9c5e..33f2b750e 100644 --- a/frac/active.go +++ b/frac/active.go @@ -3,7 +3,6 @@ package frac import ( "context" "fmt" - "io" "os" "path/filepath" "sync" @@ -52,9 +51,8 @@ type Active struct { docsCache *cache.Cache[[]byte] sortCache *cache.Cache[[]byte] - metaFile *os.File - metaReader *storage.DocBlocksReader - walReader *storage.WalReader + walFile *os.File + walReader *storage.WalReader writer *ActiveWriter indexer *ActiveIndexer @@ -72,8 +70,7 @@ func NewActive( skipMaskProvider skipMaskProvider, ) *Active { docsFile, docsStats := mustOpenFile(baseFileName+consts.DocsFileSuffix, config.SkipFsync) - - metaFile, writer, metaReader, walReader, metaSize := mustOpenMetaWriter(baseFileName, readLimiter, docsFile, docsStats) + walFile, writer, walReader, walSize := mustOpenWalWriter(baseFileName, readLimiter, docsFile, docsStats) f := &Active{ TokenList: NewActiveTokenList(config.IndexWorkers), @@ -89,15 +86,14 @@ func NewActive( docsReader: storage.NewDocsReader(readLimiter, docsFile, docsCache), sortReader: storage.NewDocsReader(readLimiter, docsFile, sortCache), - metaFile: metaFile, - metaReader: metaReader, - walReader: walReader, + walFile: walFile, + walReader: walReader, indexer: activeIndexer, writer: writer, BaseFileName: baseFileName, - info: common.NewInfo(baseFileName, uint64(docsStats.Size()), metaSize), + info: common.NewInfo(baseFileName, uint64(docsStats.Size()), walSize), Config: cfg, skipMaskProvider: skipMaskProvider, @@ -112,34 +108,21 @@ func NewActive( return f } -func mustOpenMetaWriter( +func mustOpenWalWriter( baseFileName string, readLimiter *storage.ReadLimiter, docsFile *os.File, docsStats os.FileInfo, -) (*os.File, *ActiveWriter, *storage.DocBlocksReader, *storage.WalReader, uint64) { - legacyMetaFileName := baseFileName + consts.MetaFileSuffix - - if _, err := os.Stat(legacyMetaFileName); err == nil { - // .meta file exists - metaFile, metaStats := mustOpenFile(legacyMetaFileName, config.SkipFsync) - metaSize := uint64(metaStats.Size()) - metaReader := storage.NewDocBlocksReader(readLimiter, metaFile) - writer := NewActiveWriterLegacy(docsFile, metaFile, docsStats.Size(), metaStats.Size(), config.SkipFsync) - logger.Info("using legacy meta file format", zap.String("fraction", baseFileName)) - return metaFile, writer, &metaReader, nil, metaSize - } - - logger.Info("using new WAL format", zap.String("fraction", baseFileName)) +) (*os.File, *ActiveWriter, *storage.WalReader, uint64) { walFileName := baseFileName + consts.WalFileSuffix - metaFile, metaStats := mustOpenFile(walFileName, config.SkipFsync) - metaSize := uint64(metaStats.Size()) - writer := NewActiveWriter(docsFile, metaFile, docsStats.Size(), metaStats.Size(), config.SkipFsync) - walReader, err := storage.NewWalReader(readLimiter, metaFile, baseFileName) + walFile, walStats := mustOpenFile(walFileName, config.SkipFsync) + walSize := uint64(walStats.Size()) + writer := NewActiveWriter(docsFile, walFile, docsStats.Size(), walStats.Size(), config.SkipFsync) + walReader, err := storage.NewWalReader(readLimiter, walFile, baseFileName) if err != nil { logger.Fatal("failed to initialize WAL reader", zap.String("fraction", baseFileName), zap.Error(err)) } - return metaFile, writer, nil, walReader, metaSize + return walFile, writer, walReader, walSize } func mustOpenFile(name string, skipFsync bool) (*os.File, os.FileInfo) { @@ -161,13 +144,6 @@ func mustOpenFile(name string, skipFsync bool) (*os.File, os.FileInfo) { } func (f *Active) Replay(ctx context.Context) error { - if f.metaReader != nil { - return f.replayMetaFileLegacy(ctx) - } - return f.replayWalFile(ctx) -} - -func (f *Active) replayWalFile(ctx context.Context) error { logger.Info("start replaying WAL file...", zap.String("name", f.info.Name())) t := time.Now() @@ -258,72 +234,6 @@ func (f *Active) backupCorruptedFiles() error { return nil } -// replayMetaFileLegacy replays legacy *.meta files. Only basic corruption detection support is implemented -func (f *Active) replayMetaFileLegacy(ctx context.Context) error { - logger.Info("start replaying...", zap.String("name", f.info.Name())) - - t := time.Now() - - offset := uint64(0) - step := f.info.MetaOnDisk / 10 - next := step - - sw := stopwatch.New() - wg := sync.WaitGroup{} - -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(f.info.MetaOnDisk) * 100 - logger.Info("replaying batch, meta", - zap.String("name", f.info.Name()), - zap.Uint64("from", offset), - zap.Uint64("to", offset+metaSize), - zap.Uint64("target", f.info.MetaOnDisk), - util.ZapFloat64WithPrec("progress_percentage", progress, 2), - ) - } - offset += metaSize - - wg.Add(1) - - walBlock := storage.PackDocBlockToWalBlock(meta) - f.indexer.Index(f, walBlock, &wg, sw) - } - } - - wg.Wait() - - tookSeconds := util.DurationToUnit(time.Since(t), "s") - throughputRaw := util.SizeToUnit(f.info.DocsRaw, "mb") / tookSeconds - throughputMeta := util.SizeToUnit(f.info.MetaOnDisk, "mb") / tookSeconds - logger.Info("active fraction replayed", - zap.String("name", f.info.Name()), - zap.Uint32("docs_total", f.info.DocsTotal), - util.ZapUint64AsSizeStr("docs_size", f.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 -} - var bulkStagesSeconds = promauto.NewHistogramVec(prometheus.HistogramOpts{ Namespace: "seq_db_store", Subsystem: "bulk", @@ -466,8 +376,8 @@ func (f *Active) IsIntersecting(from, to seq.MID) bool { func (f *Active) Release() { f.releaseMem() - if !f.Config.KeepMetaFile { - util.RemoveFile(f.metaFile.Name()) + if !f.Config.KeepWalFile { + util.RemoveFile(f.walFile.Name()) } if !f.Config.SkipSortDocs { @@ -479,7 +389,7 @@ func (f *Active) Release() { func (f *Active) Suicide() { f.releaseMem() - util.RemoveFile(f.metaFile.Name()) + util.RemoveFile(f.walFile.Name()) util.RemoveFile(f.docsFile.Name()) util.RemoveFile(f.BaseFileName + consts.SdocsFileSuffix) } @@ -491,8 +401,8 @@ func (f *Active) releaseMem() { 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.walFile.Close(); err != nil { + logger.Error("can't close wal 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)) diff --git a/frac/active_writer.go b/frac/active_writer.go index 279746f0d..80450f503 100644 --- a/frac/active_writer.go +++ b/frac/active_writer.go @@ -9,27 +9,14 @@ import ( type ActiveWriter struct { docs *storage.FileWriter - meta MetaWriter -} - -type MetaWriter interface { - Write(data storage.WalBlock, sw *stopwatch.Stopwatch) (int64, error) - Stop() + wal *storage.WalWriter } // NewActiveWriter creates a writer for *.wal files func NewActiveWriter(docsFile, walFile *os.File, docsOffset, walOffset int64, skipFsync bool) *ActiveWriter { return &ActiveWriter{ docs: storage.NewFileWriter(docsFile, docsOffset, skipFsync), - meta: storage.NewWalWriter(walFile, walOffset, skipFsync), - } -} - -// NewActiveWriterLegacy creates a writer for *.meta files -func NewActiveWriterLegacy(docsFile, metaFile *os.File, docsOffset, metaOffset int64, skipFsync bool) *ActiveWriter { - return &ActiveWriter{ - docs: storage.NewFileWriter(docsFile, docsOffset, skipFsync), - meta: NewLegacyMetaWriter(storage.NewFileWriter(metaFile, metaOffset, skipFsync)), + wal: storage.NewWalWriter(walFile, walOffset, skipFsync), } } @@ -45,7 +32,7 @@ func (a *ActiveWriter) Write(docs storage.DocBlock, meta storage.WalBlock, sw *s meta.SetDocsOffset(uint64(offset)) m = sw.Start("write_meta") - _, err = a.meta.Write(meta, sw) + _, err = a.wal.Write(meta, sw) m.Stop() return err @@ -53,5 +40,5 @@ func (a *ActiveWriter) Write(docs storage.DocBlock, meta storage.WalBlock, sw *s func (a *ActiveWriter) Stop() { a.docs.Stop() - a.meta.Stop() + a.wal.Stop() } diff --git a/frac/config.go b/frac/config.go index 3b1c1e979..e91392aaf 100644 --- a/frac/config.go +++ b/frac/config.go @@ -4,7 +4,7 @@ type Config struct { Search SearchConfig SkipSortDocs bool - KeepMetaFile bool + KeepWalFile bool } type SearchConfig struct { diff --git a/frac/fraction_test.go b/frac/fraction_test.go index 180ec4e24..8b0b5cfaf 100644 --- a/frac/fraction_test.go +++ b/frac/fraction_test.go @@ -2516,7 +2516,7 @@ func (s *ActiveReplayedFractionTestSuite) SetupTest() { s.SetupTestCommon() // Setting this flags allows to keep meta and docs files on disk after Active.Release() is called s.config.SkipSortDocs = true - s.config.KeepMetaFile = true + s.config.KeepWalFile = true s.insertDocuments = func(bulks ...[]string) { if s.fraction != nil { diff --git a/frac/legacy_meta_writer.go b/frac/legacy_meta_writer.go deleted file mode 100644 index b4c97e6fe..000000000 --- a/frac/legacy_meta_writer.go +++ /dev/null @@ -1,25 +0,0 @@ -package frac - -import ( - "github.com/ozontech/seq-db/metric/stopwatch" - "github.com/ozontech/seq-db/storage" -) - -// LegacyMetaWriter is MetaWriter for the legacy *.meta files. Converts new storage.WalBlock block type to -// storage.DocBlock -type LegacyMetaWriter struct { - fw *storage.FileWriter -} - -func NewLegacyMetaWriter(fw *storage.FileWriter) *LegacyMetaWriter { - return &LegacyMetaWriter{fw: fw} -} - -func (l *LegacyMetaWriter) Write(data storage.WalBlock, sw *stopwatch.Stopwatch) (int64, error) { - docBlock := storage.PackWalBlockToDocBlock(data, nil) - return l.fw.Write(docBlock, sw) -} - -func (l *LegacyMetaWriter) Stop() { - l.fw.Stop() -} diff --git a/fracmanager/frac_manifest.go b/fracmanager/frac_manifest.go index ae4adc4a6..f0f3c15e2 100644 --- a/fracmanager/frac_manifest.go +++ b/fracmanager/frac_manifest.go @@ -20,8 +20,7 @@ import ( type fracManifest struct { basePath string // base path to fraction files (without extension) hasDocs bool // presence of main documents file - hasMeta bool // presence of meta-information (legacy WAL format) - hasWal bool // presence of WAL with meta (new WAL format) + hasWal bool // presence of WAL with meta hasIndex bool // presence of index file hasSdocs bool // presence of sorted documents hasRemote bool // presence of remote fraction @@ -50,8 +49,6 @@ func (m *fracManifest) AddExtension(ext string) error { switch ext { case consts.DocsFileSuffix: m.hasDocs = true - case consts.MetaFileSuffix: - m.hasMeta = true case consts.WalFileSuffix: m.hasWal = true case consts.SdocsFileSuffix: @@ -114,7 +111,7 @@ func (m *fracManifest) Stage() fracStage { if (m.hasAllIndexFiles() || m.hasIndex) && (m.hasSdocs || m.hasDocs) { return fracStageSealed } - if (m.hasMeta || m.hasWal) && m.hasDocs { + if m.hasWal && m.hasDocs { return fracStageActive } if m.hasDocsDel || m.hasSdocsDel || m.hasIndexDel { @@ -137,11 +134,7 @@ func removeSdocs(m *fracManifest) { } } -func removeMeta(m *fracManifest) { - if m.hasMeta { - util.RemoveFile(m.basePath + consts.MetaFileSuffix) - m.hasMeta = false - } +func removeWal(m *fracManifest) { if m.hasWal { util.RemoveFile(m.basePath + consts.WalFileSuffix) m.hasWal = false @@ -274,7 +267,7 @@ func cleanupFrac(m *fracManifest) { // cleanupRemoteFrac cleans files for remote fractions // Removes local file copies since data is stored remotely func cleanupRemoteFrac(m *fracManifest) { - removeMeta(m) + removeWal(m) removeDocs(m) removeSdocs(m) removeIndexFiles(m) @@ -283,7 +276,7 @@ func cleanupRemoteFrac(m *fracManifest) { // cleanupSealedFrac cleans files for sealed fractions // Removes redundant files after finishing work with the fraction func cleanupSealedFrac(m *fracManifest) { - removeMeta(m) + removeWal(m) if m.hasSdocs { removeDocs(m) // remove orig docs, but keeping sorted } @@ -313,7 +306,6 @@ func removeAllFiles(basePath string) { consts.IDFileSuffix, consts.IDTmpFileSuffix, consts.LIDFileSuffix, consts.LIDTmpFileSuffix, - consts.MetaFileSuffix, consts.WalFileSuffix, } { util.RemoveFile(basePath + suffix) @@ -351,7 +343,6 @@ func extractSuffix(filename string) string { func (f *fracManifest) MarshalLogObject(enc zapcore.ObjectEncoder) error { enc.AddString("basePath", f.basePath) enc.AddBool("hasDocs", f.hasDocs) - enc.AddBool("hasMeta", f.hasMeta) enc.AddBool("hasWal", f.hasWal) enc.AddBool("hasIndex", f.hasIndex) enc.AddBool("hasSdocs", f.hasSdocs) diff --git a/storage/wal_block.go b/storage/wal_block.go index d84b7587f..968ba6f24 100644 --- a/storage/wal_block.go +++ b/storage/wal_block.go @@ -163,43 +163,6 @@ func PackWalBlock(payload []byte, dst WalBlock) WalBlock { return dst } -// PackWalBlockToDocBlock converts WalBlock to legacy DocBlock. -func PackWalBlockToDocBlock(walBlock WalBlock, dst DocBlock) DocBlock { - dst = append(dst[:0], make([]byte, DocBlockHeaderLen)...) - dst = append(dst, walBlock.Payload()...) - - dst.CalcLen() - dst.SetRawLen(uint64(walBlock.RawLen())) - dst.SetCodec(walBlock.Codec()) - dst.SetExt2(walBlock.DocsOffset()) - - return dst -} - -// PackDocBlockToWalBlock converts DocBlock to WalBlock in place without copying payload. -// docBlock will be invalid after packing -func PackDocBlockToWalBlock(docBlock DocBlock) WalBlock { - rawLen := uint32(docBlock.RawLen()) - codec := docBlock.Codec() - docsOffset := docBlock.GetExt2() - payloadLen := uint32(len(docBlock) - DocBlockHeaderLen) - - const headerDiff = DocBlockHeaderLen - WalBlockHeaderLen - mb := WalBlock(docBlock[headerDiff:]) - - mb[offsetWalBlockMagic] = WalBlockMagic - mb.SetVersion(WalBlockCurrentVersion) - mb.SetLen(payloadLen) - mb.SetRawLen(rawLen) - mb.SetCodec(codec) - // write docs offset directly since SetDocsOffset recalculates header checksum - binary.LittleEndian.PutUint64(mb[offsetWalBlockDocsOffset:], docsOffset) - mb.CalcPayloadChecksum() - mb.CalcHeaderChecksum() - - return mb -} - // DecompressTo always put the result in `dst` regardless of whether unpacking is required // or part of the WalBlock can be enough. // diff --git a/storage/wal_block_test.go b/storage/wal_block_test.go index c18b2a7ed..9e033fc23 100644 --- a/storage/wal_block_test.go +++ b/storage/wal_block_test.go @@ -4,7 +4,6 @@ import ( "testing" "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" ) func TestCorruptionDetection(t *testing.T) { @@ -37,38 +36,3 @@ func TestCorruptionDetection(t *testing.T) { block = append(block, 1, 2, 3) assert.False(t, block.IsCorrect()) } - -func TestConvertDocToWalBlock(t *testing.T) { - payload := []byte("test test payload") - - docBlock := CompressDocBlock(payload, nil, 1) - docBlock.SetExt2(11111) - - walBlock := PackDocBlockToWalBlock(docBlock) - - assert.Equal(t, CodecZSTD, walBlock.Codec()) - assert.Equal(t, uint32(len(payload)), walBlock.RawLen()) - assert.Equal(t, uint64(11111), walBlock.DocsOffset()) - assert.True(t, walBlock.IsCorrect()) - - decompressed, err := walBlock.DecompressTo(nil) - require.NoError(t, err) - assert.Equal(t, payload, decompressed) -} - -func TestConvertMetaToDocBlock(t *testing.T) { - payload := []byte("test payload data") - - walBlock := CompressWalBlock(payload, nil, 1) - walBlock.SetDocsOffset(22222) - - docBlock := PackWalBlockToDocBlock(walBlock, nil) - - assert.Equal(t, CodecZSTD, docBlock.Codec()) - assert.Equal(t, uint64(len(payload)), docBlock.RawLen()) - assert.Equal(t, uint64(22222), docBlock.GetExt2()) - - decompressed, err := docBlock.DecompressTo(nil) - require.NoError(t, err) - assert.Equal(t, payload, decompressed) -} diff --git a/storeapi/grpc_bulk.go b/storeapi/grpc_bulk.go index 9616a0909..67239edd4 100644 --- a/storeapi/grpc_bulk.go +++ b/storeapi/grpc_bulk.go @@ -14,7 +14,6 @@ import ( "github.com/ozontech/seq-db/logger" "github.com/ozontech/seq-db/metric" "github.com/ozontech/seq-db/pkg/storeapi" - "github.com/ozontech/seq-db/storage" "github.com/ozontech/seq-db/tracing" ) @@ -65,11 +64,6 @@ func (g *GrpcV1) doBulk(ctx context.Context, req *storeapi.BulkRequest) error { start := time.Now() - if !storage.IsWalBlock(req.Metas) { - // We use WalBlock now for metas everywhere, DocBlock might be sent by legacy proxy service. - req.Metas = storage.PackDocBlockToWalBlock(req.Metas) - } - err := g.fracManager.Append(ctx, req.Docs, req.Metas) if err != nil {