Forward compatible snapshots format (#18746)

Cherry-picked this PR: #18226
and also changed snapshots format version from v1 to v0 in compressor
settings because new format will be used as default only since 3.4

Author: eastorski

db/seg/compress.go

@@ -77,6 +77,14 @@ type Cfg struct {
 
 	// arbitrary bytes set by user at start of the file
 	ExpectMetadata bool
+
+	// number of values on compressed page. if > 0 then page level compression is enabled
+	ValuesOnCompressedPage int
+}
+
+func (c Cfg) WithValuesOnCompressedPage(n int) Cfg {
+	c.ValuesOnCompressedPage = n
+	return c
 }
 
 var DefaultCfg = Cfg{
@@ -123,7 +131,10 @@ type Compressor struct {
 	logger           log.Logger
 	noFsync          bool // fsync is enabled by default, but tests can manually disable
 
-	metadata []byte
+	version             uint8
+	featureFlagBitmask  FeatureFlagBitmask
+	compPageValuesCount uint8
+	metadata            []byte
 }
 
 func NewCompressor(ctx context.Context, logPrefix, outputFile, tmpDir string, cfg Cfg, lvl log.Lvl, logger log.Logger) (*Compressor, error) {
@@ -151,7 +162,7 @@ func NewCompressor(ctx context.Context, logPrefix, outputFile, tmpDir string, cf
 		go extractPatternsInSuperstrings(ctx, superstrings, collector, cfg, wg, logger)
 	}
 	_, outputFileName := filepath.Split(outputFile)
-	return &Compressor{
+	cc := &Compressor{
 		Cfg:              cfg,
 		uncompressedFile: uncompressedFile,
 		outputFile:       outputFile,
@@ -164,7 +175,15 @@ func NewCompressor(ctx context.Context, logPrefix, outputFile, tmpDir string, cf
 		lvl:              lvl,
 		wg:               wg,
 		logger:           logger,
-	}, nil
+		version:          FileCompressionFormatV0, // for release 3.3 only, since 3.4 we use v1 format
+	}
+
+	if cfg.ValuesOnCompressedPage > 0 {
+		cc.featureFlagBitmask.Set(PageLevelCompressionEnabled)
+		cc.compPageValuesCount = uint8(cfg.ValuesOnCompressedPage)
+	}
+
+	return cc, nil
 }
 
 func (c *Compressor) Close() {
@@ -175,9 +194,10 @@ func (c *Compressor) Close() {
 	c.suffixCollectors = nil
 }
 
-func (c *Compressor) SetTrace(trace bool) { c.trace = trace }
-func (c *Compressor) FileName() string    { return c.outputFileName }
-func (c *Compressor) WorkersAmount() int  { return c.Workers }
+func (c *Compressor) SetTrace(trace bool)            { c.trace = trace }
+func (c *Compressor) FileName() string               { return c.outputFileName }
+func (c *Compressor) WorkersAmount() int             { return c.Workers }
+func (c *Compressor) GetValuesOnCompressedPage() int { return int(c.ValuesOnCompressedPage) }
 func (c *Compressor) SetMetadata(metadata []byte) {
 	if !c.ExpectMetadata {
 		panic("metadata not expected in compressor")
@@ -276,6 +296,19 @@ func (c *Compressor) Compress() error {
 	tmpFileName := cf.Name()
 	defer dir.RemoveFile(tmpFileName)
 	defer cf.Close()
+
+	if c.version == FileCompressionFormatV1 {
+		if _, err := cf.Write([]byte{c.version, byte(c.featureFlagBitmask)}); err != nil {
+			return err
+		}
+
+		if c.featureFlagBitmask.Has(PageLevelCompressionEnabled) {
+			if _, err := cf.Write([]byte{c.compPageValuesCount}); err != nil {
+				return err
+			}
+		}
+	}
+
 	if c.ExpectMetadata {
 		dataLen := uint32(len(c.metadata))
 		var dataLenB [4]byte

db/seg/decompress.go

@@ -121,19 +121,22 @@ func (e ErrCompressedFileCorrupted) Is(err error) bool {
 
 // Decompressor provides access to the superstrings in a file produced by a compressor
 type Decompressor struct {
-	f               *os.File
-	mmapHandle2     *[mmap.MaxMapSize]byte // mmap handle for windows (this is used to close mmap)
-	dict            *patternTable
-	posDict         *posTable
-	mmapHandle1     []byte // mmap handle for unix (this is used to close mmap)
-	data            []byte // slice of correct size for the decompressor to work with
-	wordsStart      uint64 // Offset of whether the superstrings actually start
-	size            int64
-	modTime         time.Time
-	wordsCount      uint64
-	emptyWordsCount uint64
-	hasMetadata     bool
-	metadata        []byte
+	f                   *os.File
+	mmapHandle2         *[mmap.MaxMapSize]byte // mmap handle for windows (this is used to close mmap)
+	dict                *patternTable
+	posDict             *posTable
+	mmapHandle1         []byte // mmap handle for unix (this is used to close mmap)
+	data                []byte // slice of correct size for the decompressor to work with
+	wordsStart          uint64 // Offset of whether the superstrings actually start
+	size                int64
+	modTime             time.Time
+	wordsCount          uint64
+	emptyWordsCount     uint64
+	hasMetadata         bool
+	metadata            []byte
+	version             uint8
+	featureFlagBitmask  FeatureFlagBitmask
+	compPageValuesCount uint8
 
 	serializedDictSize uint64
 	lenDictSize        uint64 // huffman encoded lengths
@@ -231,6 +234,21 @@ func NewDecompressorWithMetadata(compressedFilePath string, hasMetadata bool) (*
 	d.data = d.mmapHandle1[:d.size]
 	defer d.MadvNormal().DisableReadAhead() //speedup opening on slow drives
 
+	d.version = d.data[0]
+
+	if d.version == FileCompressionFormatV1 {
+		// 1st byte: version,
+		// 2nd byte: defines how exactly the file is compressed
+		// 3rd byte (otional): exists if PageLevelCompressionEnabled flag is enabled, and defines number of values on compressed page
+		d.featureFlagBitmask = FeatureFlagBitmask(d.data[1])
+		d.data = d.data[2:]
+	}
+
+	if d.featureFlagBitmask.Has(PageLevelCompressionEnabled) {
+		d.compPageValuesCount = d.data[0]
+		d.data = d.data[1:]
+	}
+
 	if hasMetadata {
 		metadataLen := binary.BigEndian.Uint32(d.data[:4])
 		d.metadata = d.data[4 : 4+metadataLen]
@@ -362,10 +380,11 @@ func NewDecompressorWithMetadata(compressedFilePath string, hasMetadata bool) (*
 	}
 	d.wordsStart = pos + dictSize
 
-	if d.Count() == 0 && dictSize == 0 && d.size > compressedMinSize {
+	if d.Count() == 0 && dictSize == 0 && d.size > d.calcCompressedMinSize() {
 		return nil, &ErrCompressedFileCorrupted{
 			FileName: fName, Reason: fmt.Sprintf("size %v but no words in it", datasize.ByteSize(d.size).HR())}
 	}
+
 	validationPassed = true
 	return d, nil
 }
@@ -466,10 +485,12 @@ func buildPosTable(depths []uint64, poss []uint64, table *posTable, code uint16,
 func (d *Decompressor) DataHandle() unsafe.Pointer {
 	return unsafe.Pointer(&d.data[0])
 }
-func (d *Decompressor) SerializedDictSize() uint64 { return d.serializedDictSize }
-func (d *Decompressor) SerializedLenSize() uint64  { return d.lenDictSize }
-func (d *Decompressor) DictWords() int             { return d.dictWords }
-func (d *Decompressor) DictLens() int              { return d.dictLens }
+func (d *Decompressor) SerializedDictSize() uint64      { return d.serializedDictSize }
+func (d *Decompressor) SerializedLenSize() uint64       { return d.lenDictSize }
+func (d *Decompressor) DictWords() int                  { return d.dictWords }
+func (d *Decompressor) DictLens() int                   { return d.dictLens }
+func (d *Decompressor) CompressedPageValuesCount() int  { return int(d.compPageValuesCount) }
+func (d *Decompressor) CompressionFormatVersion() uint8 { return d.version }
 
 func (d *Decompressor) Size() int64 {
 	return d.size
@@ -1082,3 +1103,15 @@ func (g *Getter) BinarySearch(seek []byte, count int, getOffset func(i uint64) (
 	}
 	return foundOffset, true
 }
+
+func (d *Decompressor) calcCompressedMinSize() int64 {
+	if d.version == FileCompressionFormatV0 {
+		return compressedMinSize
+	}
+
+	if d.featureFlagBitmask.Has(PageLevelCompressionEnabled) {
+		return compressedMinSize + 3 // 2 bytes always are used for bitmask and version + 1 optional for page level compression if enabled
+	}
+
+	return compressedMinSize + 2
+}

db/seg/seg_interface.go

@@ -24,6 +24,29 @@ const (
 	CompressVals FileCompression = 0b100
 )
 
+const (
+	FileCompressionFormatV0 = uint8(0)
+	FileCompressionFormatV1 = uint8(1)
+)
+
+type FeatureFlag uint8
+
+const (
+	PageLevelCompressionEnabled FeatureFlag = 1 << iota // 0b001
+	KeyCompressionEnabled                               // 0b010
+	ValCompressionEnabled                               // 0b100
+)
+
+type FeatureFlagBitmask uint8
+
+func (m FeatureFlagBitmask) Has(flag FeatureFlag) bool {
+	return m&FeatureFlagBitmask(flag) == FeatureFlagBitmask(flag)
+}
+
+func (m *FeatureFlagBitmask) Set(flag FeatureFlag) {
+	*m |= FeatureFlagBitmask(flag)
+}
+
 func ParseFileCompression(s string) (FileCompression, error) {
 	// Implementation would be here
 	return CompressNone, nil

db/seg/seg_paged_rw.go

@@ -227,8 +227,13 @@ func (g *PagedReader) Skip() (uint64, int) {
 	return offset, len(v)
 }
 
-func NewPagedWriter(parent CompressorI, pageSize int, compressionEnabled bool) *PagedWriter {
-	return &PagedWriter{parent: parent, pageSize: pageSize, compressionEnabled: compressionEnabled}
+func NewPagedWriter(parent CompressorI, compressionEnabled bool) *PagedWriter {
+
+	return &PagedWriter{
+		parent:             parent,
+		pageSize:           parent.GetValuesOnCompressedPage(),
+		compressionEnabled: compressionEnabled,
+	}
 }
 
 type CompressorI interface {
@@ -238,6 +243,7 @@ type CompressorI interface {
 	Count() int
 	FileName() string
 	SetMetadata(data []byte)
+	GetValuesOnCompressedPage() int
 }
 type PagedWriter struct {
 	parent             CompressorI

db/seg/seg_paged_rw_test.go

@@ -35,14 +35,14 @@ func prepareLoremDictOnPagedWriter(t *testing.T, pageSize int, pageCompression b
 	logger, require := log.New(), require.New(t)
 	tmpDir := t.TempDir()
 	file := filepath.Join(tmpDir, "compressed1")
-	cfg := DefaultCfg
+	cfg := DefaultCfg.WithValuesOnCompressedPage(pageSize)
 	cfg.MinPatternScore = 1
 	cfg.Workers = 1
 	c, err := NewCompressor(context.Background(), t.Name(), file, tmpDir, cfg, log.LvlDebug, logger)
 	require.NoError(err)
 	defer c.Close()
 
-	p := NewPagedWriter(NewWriter(c, CompressNone), pageSize, pageCompression)
+	p := NewPagedWriter(NewWriter(c, CompressNone), pageCompression)
 	for k, w := range loremStrings {
 		key := fmt.Sprintf("key %d", k)
 		val := fmt.Sprintf("%s %d", w, k)
@@ -97,25 +97,27 @@ func TestPagedReader(t *testing.T) {
 
 // multyBytesWriter is a writer for [][]byte, similar to bytes.Writer.
 type multyBytesWriter struct {
-	buffer [][]byte
+	buffer   [][]byte
+	pageSize int
 }
 
 func (w *multyBytesWriter) Write(p []byte) (n int, err error) {
 	w.buffer = append(w.buffer, common.Copy(p))
 	return len(p), nil
 }
-func (w *multyBytesWriter) Bytes() [][]byte    { return w.buffer }
-func (w *multyBytesWriter) FileName() string   { return "" }
-func (w *multyBytesWriter) Count() int         { return 0 }
-func (w *multyBytesWriter) Close()             {}
-func (w *multyBytesWriter) Compress() error    { return nil }
-func (w *multyBytesWriter) Reset()             { w.buffer = nil }
-func (w *multyBytesWriter) SetMetadata([]byte) {}
+func (w *multyBytesWriter) Bytes() [][]byte                { return w.buffer }
+func (w *multyBytesWriter) FileName() string               { return "" }
+func (w *multyBytesWriter) Count() int                     { return 0 }
+func (w *multyBytesWriter) Close()                         {}
+func (w *multyBytesWriter) Compress() error                { return nil }
+func (w *multyBytesWriter) Reset()                         { w.buffer = nil }
+func (w *multyBytesWriter) SetMetadata([]byte)             {}
+func (w *multyBytesWriter) GetValuesOnCompressedPage() int { return w.pageSize }
 
 func TestPage(t *testing.T) {
-	buf, require := &multyBytesWriter{}, require.New(t)
 	sampling := 2
-	w := NewPagedWriter(buf, sampling, false)
+	buf, require := &multyBytesWriter{pageSize: sampling}, require.New(t)
+	w := NewPagedWriter(buf, false)
 	for i := 0; i < sampling+1; i++ {
 		k, v := fmt.Sprintf("k %d", i), fmt.Sprintf("v %d", i)
 		require.NoError(w.Add([]byte(k), []byte(v)))
@@ -147,8 +149,8 @@ func TestPage(t *testing.T) {
 }
 
 func BenchmarkName(b *testing.B) {
-	buf := &multyBytesWriter{}
-	w := NewPagedWriter(buf, 16, false)
+	buf := &multyBytesWriter{pageSize: 16}
+	w := NewPagedWriter(buf, false)
 	for i := 0; i < 16; i++ {
 		w.Add([]byte{byte(i)}, []byte{10 + byte(i)})
 	}

db/state/history.go

@@ -298,11 +298,12 @@ func (h *History) buildVI(ctx context.Context, historyIdxPath string, hist, efHi
 				if err = rs.AddKey(histKey, valOffset); err != nil {
 					return err
 				}
-				if h.HistoryValuesOnCompressedPage == 0 {
+
+				if h.CompressorCfg.ValuesOnCompressedPage == 0 {
 					valOffset, _ = histReader.Skip()
 				} else {
 					i++
-					if i%h.HistoryValuesOnCompressedPage == 0 {
+					if i%h.CompressorCfg.ValuesOnCompressedPage == 0 {
 						valOffset, _ = histReader.Skip()
 					}
 				}
@@ -856,10 +857,12 @@ func (h *History) dataWriter(f *seg.Compressor) *seg.PagedWriter {
 	if !strings.Contains(f.FileName(), ".v") {
 		panic("assert: miss-use " + f.FileName())
 	}
-	return seg.NewPagedWriter(seg.NewWriter(f, h.Compression), h.HistoryValuesOnCompressedPage, true)
+	return seg.NewPagedWriter(seg.NewWriter(f, h.Compression), f.GetValuesOnCompressedPage() > 0)
+}
+func (ht *HistoryRoTx) dataReader(f *seg.Decompressor) *seg.Reader { return ht.h.dataReader(f) }
+func (ht *HistoryRoTx) datarWriter(f *seg.Compressor) *seg.PagedWriter {
+	return ht.h.dataWriter(f)
 }
-func (ht *HistoryRoTx) dataReader(f *seg.Decompressor) *seg.Reader     { return ht.h.dataReader(f) }
-func (ht *HistoryRoTx) datarWriter(f *seg.Compressor) *seg.PagedWriter { return ht.h.dataWriter(f) }
 
 func (h *History) isEmpty(tx kv.Tx) (bool, error) {
 	k, err := kv.FirstKey(tx, h.ValuesTable)
@@ -1137,7 +1140,13 @@ func (ht *HistoryRoTx) historySeekInFiles(key []byte, txNum uint64) ([]byte, boo
 		fmt.Printf("DomainGetAsOf(%s, %x, %d) -> %s, histTxNum=%d, isNil(v)=%t\n", ht.h.FilenameBase, key, txNum, g.FileName(), histTxNum, v == nil)
 	}
 
-	if ht.h.HistoryValuesOnCompressedPage > 1 {
+	compressedPageValuesCount := historyItem.src.decompressor.CompressedPageValuesCount()
+
+	if historyItem.src.decompressor.CompressionFormatVersion() == seg.FileCompressionFormatV0 {
+		compressedPageValuesCount = ht.h.HistoryValuesOnCompressedPage
+	}
+
+	if compressedPageValuesCount > 1 {
 		v, ht.snappyReadBuffer = seg.GetFromPage(historyKey, v, ht.snappyReadBuffer, true)
 	}
 	return v, true, nil
@@ -1400,9 +1409,15 @@ func (ht *HistoryRoTx) HistoryDump(fromTxNum, toTxNum int, dumpTo io.Writer) err
 					return fmt.Errorf("HistoryDump: failed to resolve offset in .vi %s file for key [%x]", viFile.Fullpath(), key)
 				}
 
+				compressedPageValuesCount := viFile.src.decompressor.CompressedPageValuesCount()
+
+				if viFile.src.decompressor.CompressionFormatVersion() == seg.FileCompressionFormatV0 {
+					compressedPageValuesCount = ht.h.HistoryValuesOnCompressedPage
+				}
+
 				vGetter := seg.NewPagedReader(
 					ht.statelessGetter(viFile.i),
-					ht.h.HistoryValuesOnCompressedPage,
+					compressedPageValuesCount,
 					true,
 				)