perf: optimizing the unordered file merge self (openGemini#607)

Signed-off-by: fx408 <[email protected]>

Author: fx408

engine/immutable/chunk_iterators.go

@@ -20,18 +20,15 @@ import (
 	"container/heap"
 	"sync/atomic"
 
-	"github.com/openGemini/openGemini/lib/fileops"
 	Log "github.com/openGemini/openGemini/lib/logger"
 	"github.com/openGemini/openGemini/lib/record"
-	"go.uber.org/zap"
 )
 
 type ChunkIterator struct {
 	*FileIterator
 	ctx    *ReadContext
 	id     uint64
 	fields record.Schemas
-	rec    *record.Record
 	merge  *record.Record
 	log    *Log.Logger
 }
@@ -165,7 +162,6 @@ func NewChunkIterator(r *FileIterator) *ChunkIterator {
 		FileIterator: r,
 		ctx:          NewReadContext(true),
 		merge:        allocRecord(),
-		rec:          allocRecord(),
 	}
 
 	return itr
@@ -177,7 +173,6 @@ func (c *ChunkIterator) WithLog(log *Log.Logger) {
 
 func (c *ChunkIterator) Close() {
 	c.FileIterator.Close()
-	freeRecord(c.rec)
 	freeRecord(c.merge)
 	c.ctx.Release()
 	c.ctx = nil
@@ -208,47 +203,40 @@ func (c *ChunkIterator) Next() bool {
 		c.fields[i].Type = int(cm.ty)
 	}
 
-	if c.err = c.read(); c.err != nil {
+	if c.err = c.readRecord(); c.err != nil {
 		return false
 	}
 
 	return true
 }
 
-func (c *ChunkIterator) read() error {
+func (c *ChunkIterator) readRecord() error {
 	var err error
 	c.id = c.curtChunkMeta.sid
 	cMeta := c.curtChunkMeta
 
 	c.merge.Reset()
 	c.merge.SetSchema(c.fields)
 	c.merge.ReserveColVal(len(c.fields))
-	timeMeta := cMeta.timeMeta()
-	for i := range timeMeta.entries {
-		c.rec.Reset()
-		c.rec.SetSchema(c.fields)
-		c.rec.ReserveColVal(len(c.fields))
-		c.rec.ReserveColumnRows(8)
-		record.CheckRecord(c.rec)
-
-		c.rec, err = c.r.ReadAt(cMeta, i, c.rec, c.ctx, fileops.IO_PRIORITY_LOW_READ)
-		if err != nil {
-			c.log.Error("read segment error", zap.String("file", c.r.Path()), zap.Error(err))
-			return err
-		}
-
-		c.segPos++
 
-		record.CheckRecord(c.rec)
-		c.merge.Merge(c.rec)
-		record.CheckRecord(c.merge)
+	buf, err := c.readData(cMeta.offset, cMeta.size)
+	if err != nil {
+		return err
 	}
 
-	if c.segPos >= len(timeMeta.entries) {
-		c.curtChunkMeta = nil
-		c.chunkUsed++
+	c.merge.Reset()
+	c.merge.SetSchema(c.fields)
+	c.merge.ReserveColVal(len(c.fields))
+
+	err = decodeRecord(c.ctx, buf, cMeta, c.merge)
+	if err != nil {
+		return err
 	}
 
+	record.CheckRecord(c.merge)
+	c.curtChunkMeta = nil
+	c.chunkUsed++
+
 	return nil
 }
 
@@ -259,3 +247,34 @@ func (c *ChunkIterator) GetSeriesID() uint64 {
 func (c *ChunkIterator) GetRecord() *record.Record {
 	return c.merge
 }
+
+func decodeRecord(ctx *ReadContext, chunkData []byte, cm *ChunkMeta, dst *record.Record) error {
+	var err error
+
+	schema := dst.Schema
+	swap := &ctx.col
+
+	for i := 0; i < schema.Len(); i++ {
+		ref := &schema[i]
+		colMeta := &cm.colMeta[i]
+		col := dst.Column(i)
+
+		for n := range colMeta.entries {
+			buf := columnData(chunkData, cm.offset, colMeta.entries[n].offset, colMeta.entries[n].size)
+
+			if ref.Name == record.TimeField {
+				err = appendTimeColumnData(buf, swap, ctx, false)
+			} else {
+				err = decodeColumnData(ref, buf, swap, ctx, false)
+			}
+
+			if err != nil {
+				return err
+			}
+
+			col.AppendColVal(swap, ref.Type, 0, swap.Len)
+		}
+	}
+
+	return nil
+}

engine/immutable/merge_self.go

@@ -0,0 +1,116 @@
+/*
+Copyright 2024 Huawei Cloud Computing Technologies Co., Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package immutable
+
+import (
+	"container/heap"
+	"sync"
+
+	"github.com/openGemini/openGemini/lib/config"
+	"github.com/openGemini/openGemini/lib/logger"
+	"github.com/openGemini/openGemini/lib/record"
+)
+
+type MergeSelf struct {
+	once   sync.Once
+	signal chan struct{}
+	mts    *MmsTables
+	lg     *logger.Logger
+}
+
+func NewMergeSelf(mts *MmsTables, lg *logger.Logger) *MergeSelf {
+	return &MergeSelf{
+		signal: make(chan struct{}),
+		mts:    mts,
+		lg:     lg,
+	}
+}
+
+func (m *MergeSelf) Merge(mst string, files []TSSPFile) (TSSPFile, error) {
+	builder := m.createMsBuilder(mst, files[0].FileName())
+	sh := record.NewColumnSortHelper()
+	defer sh.Release()
+
+	itrs := m.createIterators(files)
+
+	for {
+		sid, rec, err := itrs.Next()
+		if err != nil {
+			builder.Reset()
+			return nil, err
+		}
+
+		if rec == nil || sid == 0 {
+			break
+		}
+
+		record.CheckRecord(rec)
+		rec = sh.Sort(rec)
+		itrs.merged = rec
+		builder, err = builder.WriteRecord(sid, rec, nil)
+		if err != nil {
+			builder.Reset()
+			return nil, err
+		}
+	}
+
+	itrs.Close()
+
+	merged, err := builder.NewTSSPFile(true)
+	return merged, err
+}
+
+func (m *MergeSelf) createIterators(files []TSSPFile) *ChunkIterators {
+	var dropping int64 = 0
+	itrs := &ChunkIterators{
+		dropping:      &dropping,
+		closed:        m.signal,
+		stopCompMerge: m.signal,
+		itrs:          make([]*ChunkIterator, 0, len(files)),
+		merged:        &record.Record{},
+	}
+	itrs.WithLog(m.lg)
+
+	for _, f := range files {
+		fi := NewFileIterator(f, m.lg)
+		itr := NewChunkIterator(fi)
+		itr.WithLog(m.lg)
+		ok := itr.Next()
+		if !ok || itr.err != nil {
+			itr.Close()
+			continue
+		}
+		itrs.itrs = append(itrs.itrs, itr)
+	}
+
+	heap.Init(itrs)
+	return itrs
+}
+
+func (m *MergeSelf) createMsBuilder(mst string, fileName TSSPFileName) *MsBuilder {
+	fileName.merge++
+	fileName.lock = m.mts.lock
+	builder := NewMsBuilder(m.mts.path, mst, m.mts.lock, m.mts.Conf,
+		0, fileName, 0, nil, 0, config.TSSTORE, nil, m.mts.shardId)
+	return builder
+}
+
+func (m *MergeSelf) Stop() {
+	m.once.Do(func() {
+		close(m.signal)
+	})
+}

engine/immutable/merge_self_test.go

@@ -0,0 +1,104 @@
+/*
+Copyright 2024 Huawei Cloud Computing Technologies Co., Ltd.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package immutable_test
+
+import (
+	"testing"
+
+	"github.com/openGemini/openGemini/engine/immutable"
+	"github.com/openGemini/openGemini/lib/config"
+	"github.com/openGemini/openGemini/lib/logger"
+	"github.com/openGemini/openGemini/lib/record"
+	"github.com/openGemini/openGemini/lib/util/lifted/vm/protoparser/influx"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMergeSelf(t *testing.T) {
+	var begin int64 = 1e12
+	defer beforeTest(t, 0)()
+	conf := config.GetStoreConfig()
+	conf.Merge.MergeSelfOnly = true
+
+	mh := NewMergeTestHelper(immutable.NewTsStoreConfig())
+	defer mh.store.Close()
+	rg := newRecordGenerator(begin, defaultInterval, true)
+
+	mh.addRecord(100, rg.generate(getDefaultSchemas(), 10))
+	require.NoError(t, mh.saveToOrder())
+
+	schema := getDefaultSchemas()
+	for i := 0; i < 14; i++ {
+		rg.setBegin(begin + int64(i))
+		mh.addRecord(uint64(100+i%2), rg.generate(schema, 10))
+		require.NoError(t, mh.saveToUnordered())
+	}
+
+	schema = append(schema, record.Field{Type: influx.Field_Type_Int, Name: "int_1"})
+	rg.setBegin(begin + 100)
+	mh.addRecord(100, rg.generate(schema, 10))
+	require.NoError(t, mh.saveToUnordered())
+
+	schema = append(getDefaultSchemas(), record.Field{Type: influx.Field_Type_Float, Name: "float_1"})
+	rg.setBegin(begin + 101)
+	mh.addRecord(100, rg.generate(schema, 10))
+	require.NoError(t, mh.saveToUnordered())
+
+	conf.Merge.MergeSelfOnly = false
+	err := mh.store.MergeOutOfOrder(1, false, true)
+	require.NoError(t, err)
+	mh.store.Wait()
+
+	require.NoError(t, mh.mergeAndCompact(true))
+	require.NoError(t, compareRecords(mh.readExpectRecord(), mh.readMergedRecord()))
+}
+
+func TestMergeSelf_Stop(t *testing.T) {
+	var begin int64 = 1e12
+	defer beforeTest(t, 0)()
+	conf := config.GetStoreConfig()
+	conf.Merge.MergeSelfOnly = true
+
+	mh := NewMergeTestHelper(immutable.NewTsStoreConfig())
+	defer mh.store.Close()
+	rg := newRecordGenerator(begin, defaultInterval, true)
+
+	mh.addRecord(100, rg.generate(getDefaultSchemas(), 10))
+	require.NoError(t, mh.saveToOrder())
+
+	schema := getDefaultSchemas()
+	for i := 0; i < 8; i++ {
+		rg.setBegin(begin + 1)
+		mh.addRecord(100, rg.generate(schema, 10))
+		require.NoError(t, mh.saveToUnordered())
+	}
+
+	var files []immutable.TSSPFile
+	for _, f := range mh.store.OutOfOrder["mst"].Files() {
+		files = append(files, f)
+	}
+
+	ms := immutable.NewMergeSelf(mh.store, logger.NewLogger(0))
+	ms.Stop()
+	_, err := ms.Merge("mst", files)
+	require.EqualError(t, err, "compact stopped")
+
+	ms = immutable.NewMergeSelf(mh.store, logger.NewLogger(0))
+	files[1], files[0] = files[0], files[1]
+	_ = files[1].Close()
+	_, err = ms.Merge("mst", files)
+	require.NoError(t, err)
+}