From bc0bb0c36a3351a728e781f8e33f681876e7f5ea Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@victoriametrics.com>
Date: Fri, 27 Sep 2024 11:15:43 +0200
Subject: [PATCH] lib/logstorage: consistently sort stream contexts belonging
 to different streams by the minimum time seen in the matching logs

This should simplify debugging of stream_context output, since it remains stable over repeated requests.
---
 docs/VictoriaLogs/CHANGELOG.md        |  3 +++
 lib/logstorage/pipe_stream_context.go | 33 ++++++++++++++++++++++++++-
 lib/logstorage/storage_search_test.go | 10 ++++----
 3 files changed, 40 insertions(+), 6 deletions(-)

diff --git a/docs/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md
index b3191beb2..56efffff8 100644
--- a/docs/VictoriaLogs/CHANGELOG.md
+++ b/docs/VictoriaLogs/CHANGELOG.md
@@ -15,6 +15,9 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
 
 ## tip
 
+* BUGFIX: consistently return matching log streams sorted by time from [`stream_context` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#stream_context-pipe). Previously log streams could be returned in arbitrary order with every request. This could complicate using `stream_context` pipe.
+* BUGFIX: add missing `_msg="---"` delimiter between stream contexts belonging to different [log streams](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields). This should simplify investigating `stream_context` output for multiple matching log streams.
+
 ## [v0.30.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.30.0-victorialogs)
 
 Released at 2024-09-27
diff --git a/lib/logstorage/pipe_stream_context.go b/lib/logstorage/pipe_stream_context.go
index e9a6d790b..06a86c39d 100644
--- a/lib/logstorage/pipe_stream_context.go
+++ b/lib/logstorage/pipe_stream_context.go
@@ -530,7 +530,10 @@ func (pcp *pipeStreamContextProcessor) flush() error {
 		pcp: pcp,
 	}
 
-	for streamID, rows := range m {
+	// write output contexts in the ascending order of rows
+	streamIDs := getStreamIDsSortedByMinRowTimestamp(m)
+	for _, streamID := range streamIDs {
+		rows := m[streamID]
 		streamRowss, err := pcp.getStreamRowss(streamID, rows, stateSizeBudget)
 		if err != nil {
 			return err
@@ -557,6 +560,34 @@ func (pcp *pipeStreamContextProcessor) flush() error {
 	return nil
 }
 
+func getStreamIDsSortedByMinRowTimestamp(m map[string][]streamContextRow) []string {
+	type streamTimestamp struct {
+		streamID  string
+		timestamp int64
+	}
+	streamTimestamps := make([]streamTimestamp, 0, len(m))
+	for streamID, rows := range m {
+		minTimestamp := rows[0].timestamp
+		for _, r := range rows[1:] {
+			if r.timestamp < minTimestamp {
+				minTimestamp = r.timestamp
+			}
+		}
+		streamTimestamps = append(streamTimestamps, streamTimestamp{
+			streamID:  streamID,
+			timestamp: minTimestamp,
+		})
+	}
+	sort.Slice(streamTimestamps, func(i, j int) bool {
+		return streamTimestamps[i].timestamp < streamTimestamps[j].timestamp
+	})
+	streamIDs := make([]string, len(streamTimestamps))
+	for i := range streamIDs {
+		streamIDs[i] = streamTimestamps[i].streamID
+	}
+	return streamIDs
+}
+
 func newDelimiterRowFields(r *streamContextRow, streamID string) []Field {
 	return []Field{
 		{
diff --git a/lib/logstorage/storage_search_test.go b/lib/logstorage/storage_search_test.go
index fb7709fe9..4fd7ed0ac 100644
--- a/lib/logstorage/storage_search_test.go
+++ b/lib/logstorage/storage_search_test.go
@@ -662,7 +662,7 @@ func TestStorageRunQuery(t *testing.T) {
 			| stream_context before 0
 			| stats count() rows`, [][]Field{
 			{
-				{"rows", "33"},
+				{"rows", "66"},
 			},
 		})
 	})
@@ -671,7 +671,7 @@ func TestStorageRunQuery(t *testing.T) {
 			| stream_context before 0 after 0
 			| stats count() rows`, [][]Field{
 			{
-				{"rows", "33"},
+				{"rows", "66"},
 			},
 		})
 	})
@@ -680,7 +680,7 @@ func TestStorageRunQuery(t *testing.T) {
 			| stream_context before 1
 			| stats count() rows`, [][]Field{
 			{
-				{"rows", "66"},
+				{"rows", "99"},
 			},
 		})
 	})
@@ -689,7 +689,7 @@ func TestStorageRunQuery(t *testing.T) {
 			| stream_context after 1
 			| stats count() rows`, [][]Field{
 			{
-				{"rows", "66"},
+				{"rows", "99"},
 			},
 		})
 	})
@@ -698,7 +698,7 @@ func TestStorageRunQuery(t *testing.T) {
 			| stream_context before 1 after 1
 			| stats count() rows`, [][]Field{
 			{
-				{"rows", "99"},
+				{"rows", "132"},
 			},
 		})
 	})