New sealer

Author: eguguchkin

bytespool/writer.go

@@ -29,15 +29,18 @@ func AcquireWriterSize(out io.Writer, size int) *Writer {
 	}
 }
 
-func FlushReleaseWriter(w *Writer) error {
-	err := w.Flush()
-	if err != nil {
-		return err
-	}
+func ReleaseWriter(w *Writer) {
 	Release(w.Buf)
 	w.Buf = nil
 	w.out = nil
 	writerPool.Put(w)
+}
+
+func FlushReleaseWriter(w *Writer) error {
+	if err := w.Flush(); err != nil {
+		return err
+	}
+	ReleaseWriter(w)
 	return nil
 }
 

cmd/distribution/main.go

@@ -12,7 +12,8 @@ import (
 	"github.com/ozontech/seq-db/cache"
 	"github.com/ozontech/seq-db/consts"
 	"github.com/ozontech/seq-db/disk"
-	"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/fracmanager"
 	"github.com/ozontech/seq-db/logger"
 	"github.com/ozontech/seq-db/seq"
@@ -58,7 +59,7 @@ func readBlock(reader disk.IndexReader, blockIndex uint32) ([]byte, error) {
 	return data, nil
 }
 
-func loadInfo(path string) *frac.Info {
+func loadInfo(path string) *common.Info {
 	indexReader, f := getReader(path)
 	result, err := readBlock(indexReader, 0)
 	if err != nil {
@@ -69,7 +70,7 @@ func loadInfo(path string) *frac.Info {
 		logger.Fatal("seq-db index file header corrupted", zap.String("file", path))
 	}
 
-	b := frac.BlockInfo{}
+	b := sealed.BlockInfo{}
 	err = b.Unpack(result)
 	if err != nil {
 		logger.Fatal("can't unpack info bloc of index file", zap.String("file", path), zap.Error(err))
@@ -84,7 +85,7 @@ func loadInfo(path string) *frac.Info {
 	return b.Info
 }
 
-func buildDist(dist *seq.MIDsDistribution, path string, _ *frac.Info) {
+func buildDist(dist *seq.MIDsDistribution, path string, _ *common.Info) {
 	blocksReader, _ := getReader(path)
 
 	// skip tokens

cmd/index_analyzer/main.go

@@ -12,7 +12,7 @@ import (
 	"github.com/alecthomas/units"
 
 	"github.com/ozontech/seq-db/disk"
-	"github.com/ozontech/seq-db/frac"
+	"github.com/ozontech/seq-db/frac/sealed"
 	"github.com/ozontech/seq-db/frac/sealed/lids"
 	"github.com/ozontech/seq-db/frac/sealed/token"
 	"github.com/ozontech/seq-db/fracmanager"
@@ -92,8 +92,11 @@ func analyzeIndex(
 	}
 
 	// load info
-	b := frac.BlockInfo{}
-	_ = b.Unpack(readBlock())
+	var b sealed.BlockInfo
+	if err := b.Unpack(readBlock()); err != nil {
+		logger.Fatal("error unpacking block info", zap.Error(err))
+	}
+
 	docsCount := int(b.Info.DocsTotal)
 
 	// load tokens

cmd/seq-db/seq-db.go

@@ -22,6 +22,7 @@ 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"
@@ -253,7 +254,7 @@ func startStore(
 			MaintenanceDelay:  0,
 			CacheGCDelay:      0,
 			CacheCleanupDelay: 0,
-			SealParams: frac.SealParams{
+			SealParams: common.SealParams{
 				IDsZstdLevel:           cfg.Compression.SealedZstdCompressionLevel,
 				LIDsZstdLevel:          cfg.Compression.SealedZstdCompressionLevel,
 				TokenListZstdLevel:     cfg.Compression.SealedZstdCompressionLevel,

consts/consts.go

@@ -11,10 +11,9 @@ const (
 	// DummyMID is used in aggregations when we do not need to build time series.
 	DummyMID = 0
 
-	IDsBlockSize     = int(4 * units.KiB)
-	RegularBlockSize = int(16 * units.KiB)
 	IDsPerBlock      = int(4 * units.KiB)
 	LIDBlockCap      = int(64 * units.KiB)
+	RegularBlockSize = int(16 * units.KiB)
 
 	DefaultMaintenanceDelay  = time.Second
 	DefaultCacheGCDelay      = 1 * time.Second

disk/block_former.go

@@ -22,12 +22,12 @@ func NewEmptyIndexBlockHeader() IndexBlockHeader {
 	return make(IndexBlockHeader, IndexBlockHeaderSize)
 }
 
-func NewIndexBlockHeader(pos int64, ext1, ext2 uint64, origBuff, finalBuf []byte, codec Codec) IndexBlockHeader {
+func NewIndexBlockHeader(pos int64, ext1, ext2 uint64, size, rawSize int, codec Codec) IndexBlockHeader {
 	header := NewEmptyIndexBlockHeader()
 	header.SetExt1(ext1)
 	header.SetExt2(ext2)
-	header.SetLen(uint32(len(finalBuf)))
-	header.SetRawLen(uint32(len(origBuff)))
+	header.SetLen(uint32(size))
+	header.SetRawLen(uint32(rawSize))
 	header.SetCodec(codec)
 	header.SetPos(uint64(pos))
 	return header