wip

lib/logstorage/block_result.go

@@ -486,6 +486,18 @@ func (br *blockResult) addTimeColumn() {
 }
 
 func (br *blockResult) addStreamColumn(bs *blockSearch) bool {
+	if !bs.prevStreamID.equal(&bs.bsw.bh.streamID) {
+		return br.addStreamColumnSlow(bs)
+	}
+
+	if len(bs.prevStream) == 0 {
+		return false
+	}
+	br.addConstColumn("_stream", bytesutil.ToUnsafeString(bs.prevStream))
+	return true
+}
+
+func (br *blockResult) addStreamColumnSlow(bs *blockSearch) bool {
 	bb := bbPool.Get()
 	defer bbPool.Put(bb)
 
@@ -496,6 +508,8 @@ func (br *blockResult) addStreamColumn(bs *blockSearch) bool {
 		// was recently registered and its tags aren't visible to search yet.
 		// The stream tags must become visible in a few seconds.
 		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6042
+		bs.prevStreamID = *streamID
+		bs.prevStream = bs.prevStream[:0]
 		return false
 	}
 
@@ -506,6 +520,9 @@ func (br *blockResult) addStreamColumn(bs *blockSearch) bool {
 
 	s := bytesutil.ToUnsafeString(bb.B)
 	br.addConstColumn("_stream", s)
+
+	bs.prevStreamID = *streamID
+	bs.prevStream = append(bs.prevStream[:0], s...)
 	return true
 }
 

lib/logstorage/block_search.go

@@ -114,6 +114,11 @@ type blockSearch struct {
 
 	// a is used for storing unmarshaled data in csh
 	a arena
+
+	// prevStreamID and prevStream are used for speeding up fetching _stream columns
+	// across sequential blocks belonging to the same stream.
+	prevStreamID streamID
+	prevStream   []byte
 }
 
 func (bs *blockSearch) reset() {