simplify intcomp usage, refine tests

Author: cheb0

cmd/index_analyzer/main.go

@@ -104,6 +104,7 @@ func analyzeIndex(
 	if err := b.Unpack(readBlock()); err != nil {
 		logger.Fatal("error unpacking block info", zap.Error(err))
 	}
+	ver := b.Info.BinaryDataVer
 
 	docsCount := int(b.Info.DocsTotal)
 
@@ -162,7 +163,7 @@ func analyzeIndex(
 		}
 
 		block := &lids.Block{}
-		if err := block.Unpack(data, &lids.UnpackBuffer{}); err != nil {
+		if err := block.Unpack(data, ver, &lids.UnpackBuffer{}); err != nil {
 			logger.Fatal("error unpacking lids block", zap.Error(err))
 		}
 

config/frac_version.go

@@ -9,7 +9,7 @@ const (
 	BinaryDataV1
 	// BinaryDataV2 - MIDs stored in nanoseconds
 	BinaryDataV2
-	// BinaryDataV3 - MIDs and LIDs encoded in bitpack, variable LID block size
+	// BinaryDataV3 - delta bitpack encoded MIDs and LIDs
 	BinaryDataV3
 )
 

consts/consts.go

@@ -12,7 +12,7 @@ const (
 	DummyMID = 0
 
 	IDsPerBlock      = int(4 * units.KiB)
-	LIDBlockCap      = int(4 * units.KiB)
+	LIDBlockCap      = int(64 * units.KiB)
 	RegularBlockSize = int(16 * units.KiB)
 
 	DefaultMaintenanceDelay  = time.Second

frac/fraction_test.go

@@ -1636,8 +1636,8 @@ func (s *FractionTestSuite) TestFractionInfo() {
 			"index on disk doesn't match. actual value: %d", info.IndexOnDisk)
 	case *Remote:
 		s.Require().Equal(uint64(0), info.MetaOnDisk, "meta on disk doesn't match. actual value")
-		s.Require().True(info.IndexOnDisk > uint64(1400) && info.IndexOnDisk < uint64(1500),
-			"index on disk doesn't match. actual value: %d", info.MetaOnDisk)
+		s.Require().True(info.IndexOnDisk > uint64(1400) && info.IndexOnDisk < uint64(1550),
+			"index on disk doesn't match. actual value: %d", info.IndexOnDisk)
 	default:
 		s.Require().Fail("unsupported fraction type")
 	}

frac/remote.go

@@ -147,7 +147,7 @@ func (f *Remote) createDataProvider(ctx context.Context) (*sealedDataProvider, e
 		docsReader:       &f.docsReader,
 		blocksOffsets:    f.blocksData.BlocksOffsets,
 		lidsTable:        f.blocksData.LIDsTable,
-		lidsLoader:       lids.NewLoader(&f.indexReader, f.indexCache.LIDs),
+		lidsLoader:       lids.NewLoader(f.info.BinaryDataVer, &f.indexReader, f.indexCache.LIDs),
 		tokenBlockLoader: token.NewBlockLoader(f.BaseFileName, &f.indexReader, f.indexCache.Tokens),
 		tokenTableLoader: token.NewTableLoader(f.BaseFileName, &f.indexReader, f.indexCache.TokenTable),
 

frac/sealed.go

@@ -323,7 +323,7 @@ func (f *Sealed) createDataProvider(ctx context.Context) *sealedDataProvider {
 		docsReader:       &f.docsReader,
 		blocksOffsets:    f.blocksData.BlocksOffsets,
 		lidsTable:        f.blocksData.LIDsTable,
-		lidsLoader:       lids.NewLoader(&f.indexReader, f.indexCache.LIDs),
+		lidsLoader:       lids.NewLoader(f.info.BinaryDataVer, &f.indexReader, f.indexCache.LIDs),
 		tokenBlockLoader: token.NewBlockLoader(f.BaseFileName, &f.indexReader, f.indexCache.Tokens),
 		tokenTableLoader: token.NewTableLoader(f.BaseFileName, &f.indexReader, f.indexCache.TokenTable),
 

frac/sealed/lids/block.go

@@ -2,9 +2,11 @@ package lids
 
 import (
 	"encoding/binary"
+	"errors"
+	"math"
 	"unsafe"
 
-	"github.com/ozontech/seq-db/consts"
+	"github.com/ozontech/seq-db/config"
 	"github.com/ozontech/seq-db/packer"
 )
 
@@ -23,40 +25,18 @@ func (b *Block) getLIDs(i int) []uint32 {
 	return b.LIDs[b.Offsets[i]:b.Offsets[i+1]]
 }
 
-func (b *Block) Pack(dst []byte) []byte {
+func (b *Block) Pack(dst []byte, tmp []uint32) []byte {
 	// TODO store next flags into a single byte
 	// write b.IsLastLID as a dedicated uint32 in the header of block
-	switch b.IsLastLID {
-	case true:
+	if b.IsLastLID {
 		dst = binary.LittleEndian.AppendUint32(dst, 1)
-	case false:
+	} else {
 		dst = binary.LittleEndian.AppendUint32(dst, 0)
 	}
 
-	fullBlock := len(b.LIDs) == consts.LIDBlockCap
-	switch fullBlock {
-	case true:
-		dst = binary.LittleEndian.AppendUint32(dst, 1)
-	case false:
-		dst = binary.LittleEndian.AppendUint32(dst, 0)
-	}
+	dst = packer.CompressDeltaBitpackUint32(dst, b.Offsets, tmp)
+	dst = packer.CompressDeltaBitpackUint32(dst, b.LIDs, tmp)
 
-	if len(b.LIDs) == consts.LIDBlockCap {
-		offsetPacker := packer.NewBitpacker(dst, 128)
-		offsetPacker.Append(b.Offsets)
-		dst = offsetPacker.Close()
-		lidPacker := packer.NewBitpacker(dst, 128)
-		dst = lidPacker.Append4kBlock(b.LIDs)
-	} else {
-		lidPacker := packer.NewBitpacker(dst, 128)
-		sep := []uint32{0}
-		last := b.getCount() - 1
-		for i := 0; i <= last; i++ {
-			lidPacker.Append(b.getLIDs(i))
-			lidPacker.Append(sep)
-		}
-		dst = lidPacker.Close()
-	}
 	return dst
 }
 
@@ -68,63 +48,75 @@ func (b *Block) GetSizeBytes() int {
 	return blockSize + uint32Size*cap(b.LIDs) + uint32Size*cap(b.Offsets)
 }
 
-// TODO add support of the previous versions
-func (b *Block) Unpack(data []byte, buf *UnpackBuffer) error {
-	unpacker := packer.NewBytesUnpacker(data)
+func (b *Block) Unpack(data []byte, fracVer config.BinaryDataVersion, buf *UnpackBuffer) error {
 	buf.Reset()
 
+	if fracVer >= config.BinaryDataV3 {
+		return b.unpackBitpack(data, buf)
+	}
+
+	return b.unpackVarint(data, buf)
+}
+
+func (b *Block) unpackBitpack(data []byte, buf *UnpackBuffer) error {
 	// read IsLastLID from a dedicated uint32
-	isLastLIDValue := unpacker.GetUint32()
-	switch isLastLIDValue {
-	case 1:
-		b.IsLastLID = true
-	case 0:
-		b.IsLastLID = false
+	if len(data) < 4 {
+		return errors.New("lids block decode error: truncated IsLastLID header")
 	}
+	isLastLIDValue := binary.LittleEndian.Uint32(data[:4])
+	b.IsLastLID = isLastLIDValue == 1
+	data = data[4:]
+
+	var err error
+	var values []uint32
 
-	fullBlock := unpacker.GetUint32()
-	switch fullBlock {
-	case 1:
-		// block has exactly consts.LIDBlockCap LIDs
-		decompressedChunk := buf.decompressed
-		compressedChunk := buf.compressed
-		offsetUnpacker := packer.NewBitpackUnpacker(unpacker, decompressedChunk, compressedChunk)
-		for {
-			offsetChunk, ok := offsetUnpacker.NextChunk()
-			if !ok {
-				break
-			}
-			b.Offsets = append(b.Offsets, offsetChunk...)
+	data, values, err = packer.DecompressDeltaBitpackUint32(data, buf.compressed, buf.decompressed)
+	if err != nil {
+		return err
+	}
+	b.Offsets = append([]uint32{}, values...)
+
+	data, values, err = packer.DecompressDeltaBitpackUint32(data, buf.compressed, buf.decompressed)
+	if err != nil {
+		return err
+	}
+	b.LIDs = append([]uint32{}, values...)
+	return nil
+}
+
+func (b *Block) unpackVarint(data []byte, buf *UnpackBuffer) error {
+	var lid, offset uint32
+
+	b.IsLastLID = true
+
+	buf.offsets = append(buf.offsets, 0) // first offset is always zero
+
+	unpacker := packer.NewBytesUnpacker(data)
+	for unpacker.Len() > 0 {
+		delta, err := unpacker.GetVarint()
+		if err != nil {
+			return err
 		}
+		lid += uint32(delta)
 
-		lidUnpacker := packer.NewBitpackUnpacker(unpacker, decompressedChunk, compressedChunk)
-		b.LIDs = lidUnpacker.AllocateAndRead4kChunk()
-	case 0:
-		decompressedChunk := buf.decompressed
-		compressedChunk := buf.compressed
-		buf.offsets = append(buf.offsets, 0)
-
-		bitpackUnpacker := packer.NewBitpackUnpacker(unpacker, decompressedChunk, compressedChunk)
-		pos := 0
-		for {
-			chunk, ok := bitpackUnpacker.NextChunk()
-			if !ok {
-				break
-			}
-
-			for _, lid := range chunk {
-				if pos > 0 && lid == 0 {
-					b.LIDs = append(b.LIDs, buf.lids...)
-					buf.lids = buf.lids[:0]
-					buf.offsets = append(buf.offsets, uint32(pos))
-				} else {
-					buf.lids = append(buf.lids, lid)
-					pos++
-				}
-			}
+		if lid == math.MaxUint32 { // end of LIDs of current TID, see `Block.Pack()` method
+			offset = uint32(len(buf.lids))
+			buf.offsets = append(buf.offsets, offset)
+			lid -= uint32(delta)
+			continue
 		}
 
-		b.Offsets = append([]uint32{}, buf.offsets...)
+		buf.lids = append(buf.lids, lid)
 	}
+
+	if int(offset) < len(buf.lids) {
+		b.IsLastLID = false
+		buf.offsets = append(buf.offsets, uint32(len(buf.lids)))
+	}
+
+	// copy from buffer
+	b.LIDs = append([]uint32{}, buf.lids...)
+	b.Offsets = append([]uint32{}, buf.offsets...)
+
 	return nil
 }