chore: panic when unordered merge is closed abnormally (openGemini#616)

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

Author: fx408

engine/immutable/merge_performer.go

@@ -191,6 +191,7 @@ func (p *mergePerformer) Finish() (TSSPFile, error) {
 	}
 
 	p.Close()
+	p.Release()
 	return file, nil
 }
 
@@ -388,6 +389,9 @@ func (p *mergePerformer) WriteOriginal(fi *FileIterator) error {
 
 func (p *mergePerformer) Close() {
 	p.itr.Close()
+}
+
+func (p *mergePerformer) Release() {
 	p.itr = nil
 	p.ur = nil
 	p.sw = nil
@@ -548,15 +552,21 @@ func (c *MergePerformers) Pop() interface{} {
 	return v
 }
 
+func (c *MergePerformers) Done() {
+	c.wg.Done()
+}
+
 func (c *MergePerformers) Close() {
 	c.once.Do(func() {
 		close(c.signal)
 		c.closed = true
-		c.wg.Wait()
-
 		for _, item := range c.items {
 			item.Close()
 		}
+		c.wg.Wait()
+		for _, item := range c.items {
+			item.Release()
+		}
 		c.items = nil
 		c.ur = nil
 	})

engine/immutable/merge_tool.go

@@ -141,7 +141,7 @@ func (mt *mergeTool) execute(mst string, order, unordered *TSSPFiles) (*TSSPFile
 	var tempFiles []fileops.File
 
 	defer func() {
-		performers.wg.Done()
+		performers.Done()
 		performers.Close()
 		ur.Close()
 

engine/immutable/merge_tool_test.go

@@ -1137,5 +1137,43 @@ func BenchmarkChunkMeta_compress(b *testing.B) {
 		immutable.SetChunkMetaCompressMode(immutable.ChunkMetaCompressLZ4)
 		run(2000)
 	})
+}
+
+func TestMergePerformers_close(t *testing.T) {
+	var begin int64 = 1e12
+	defer beforeTest(t, 0)()
+
+	schemas := getDefaultSchemas()
+	mh := NewMergeTestHelper(immutable.NewTsStoreConfig())
+	defer mh.store.Close()
+	rg := newRecordGenerator(begin, defaultInterval, true)
+
+	for i := 0; i < 8; i++ {
+		mh.addRecord(uint64(100+i), rg.generate(schemas, 10))
+		require.NoError(t, mh.saveToOrder())
+	}
+
+	performers := immutable.NewMergePerformers(nil)
+	var tempFiles []*immutable.StreamWriteFile
+
+	defer func() {
+		for _, f := range tempFiles {
+			f.Close(true)
+		}
+	}()
+	for _, f := range mh.store.Order["mst"].Files() {
+		sw := mh.store.NewStreamWriteFile("mst")
+		require.NoError(t, sw.InitMergedFile(f))
+		sw.SetValidate(true)
+		tempFiles = append(tempFiles, sw)
+
+		p := immutable.NewMergePerformer(nil, nil)
+
+		itr := immutable.NewColumnIterator(immutable.NewFileIterator(f, nil))
+		p.Reset(sw, itr)
+		performers.Push(p)
+	}
 
+	performers.Done()
+	performers.Close()
 }

lib/util/lifted/influx/coordinator/statement_executor.go

@@ -201,23 +201,35 @@ func (e *StatementExecutor) Close() error {
 // ExecuteStatement executes the given statement with the given execution context.
 func (e *StatementExecutor) ExecuteStatement(stmt influxql.Statement, ctx *query.ExecutionContext, seq int) error {
 	e.MaxQueryParallel = int(atomic.LoadInt32(&syscontrol.QueryParallel))
+	stmtString := stmt.String()
+
 	// Select statements are handled separately so that they can be streamed.
 	if stmt, ok := stmt.(*influxql.SelectStatement); ok {
+		begin := time.Now()
 		err := e.retryExecuteSelectStatement(stmt, ctx, seq)
+		dur := time.Since(begin)
 		if err == nil {
+			if dur.Nanoseconds() > time.Second.Nanoseconds() {
+				e.StmtExecLogger.GetZapLogger().Warn("slow query",
+					zap.String("stmt", stmtString),
+					zap.Float64("duration", dur.Seconds()))
+			}
 			return nil
-		} else if errno.Equal(err, errno.DatabaseNotFound) ||
-			errno.Equal(err, errno.ErrMeasurementNotFound) {
-			e.StmtExecLogger.Error("execute select statement 400 error", zap.Any("stmt", stmt), zap.Error(err))
+		}
+
+		if errno.Equal(err, errno.DatabaseNotFound, errno.ErrMeasurementNotFound) {
+			e.StmtExecLogger.Error("execute select statement 400 error", zap.Any("stmt", stmtString),
+				zap.Error(err), zap.Float64("duration", dur.Seconds()))
 			atomic.AddInt64(&statistics.HandlerStat.Query400ErrorStmtCount, 1)
 		} else {
-			e.StmtExecLogger.Error("execute select statement 500 error", zap.Any("stmt", stmt), zap.Error(err))
+			e.StmtExecLogger.Error("execute select statement 500 error", zap.Any("stmt", stmtString),
+				zap.Error(err), zap.Float64("duration", dur.Seconds()))
 			atomic.AddInt64(&statistics.HandlerStat.QueryErrorStmtCount, 1)
 		}
 		return err
 	}
 
-	e.StmtExecLogger.Info("start execute statement", zap.Any("stmt", stmt))
+	e.StmtExecLogger.Info("start execute statement", zap.Any("stmt", stmtString))
 	var rows models.Rows
 	var messages []*query.Message
 	var err error