lib/logstorage: fix time ordering in running_stats and total_stats

docs/victorialogs/CHANGELOG.md

@@ -36,6 +36,7 @@ according to the following docs:
 
 * BUGFIX: [syslog data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/): avoid stamping [RFC3164](https://datatracker.ietf.org/doc/html/rfc3164) messages received right after the new year with the previous year (which could drop them under a short `-retentionPeriod`), and compute the year in `-syslog.timezone` instead of the server local timezone. See [#1556](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1556).
 * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): fix [`field_names` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#field_names-pipe) not respecting [`hidden_fields_filters`](https://docs.victoriametrics.com/victorialogs/querying/#hidden-fields). Fields listed in `hidden_fields_filters` still appeared in `field_names` output, leaking their existence to the caller. See [#1574](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1574/).
+* BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): fix the [`running_stats`](https://docs.victoriametrics.com/victorialogs/logsql/#running_stats-pipe) and [`total_stats`](https://docs.victoriametrics.com/victorialogs/logsql/#total_stats-pipe) pipes processing logs in the wrong time order. Previously, they compared `_time` values as text instead of comparing the actual timestamps. See [#1581](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1581).
 * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): fix a `vlselect` crash in [cluster mode](https://docs.victoriametrics.com/victorialogs/cluster/) on some valid queries. This happened when a [`math` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#math-pipe) operand had the same name as a function (such as `abs`, `round` or `floor`), or when `from` was used as a separator in the [`split` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#split-pipe). See [#1518](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1518).
 * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): allow [`filter` pipes](https://docs.victoriametrics.com/victorialogs/logsql/#filter-pipe) without the `filter` prefix when the filter starts with a non-word token or the `not` keyword, such as `... | !foo`, `... | {host="x"}`, `... | >5`, `... | =foo` or `... | not foo`. These were unintentionally rejected with `unexpected pipe` in [v1.51.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.51.0). A non-word token (and the `not` operator) cannot clash with a pipe name, so it is unambiguously a filter. See [#1522](https://github.com/VictoriaMetrics/VictoriaLogs/issues/1522).
 * BUGFIX: [LogsQL](https://docs.victoriametrics.com/victorialogs/logsql/): reject `field*` wildcards in the comma-separated (no parens) form of [`uniq`](https://docs.victoriametrics.com/victorialogs/logsql/#uniq-pipe), [`top`](https://docs.victoriametrics.com/victorialogs/logsql/#top-pipe) and [`unroll`](https://docs.victoriametrics.com/victorialogs/logsql/#unroll-pipe) pipes, consistently with their parens form, which already rejects wildcards. Previously a query such as `uniq by foo*` was accepted but could not be parsed back from its string representation. See [#1487](https://github.com/VictoriaMetrics/VictoriaLogs/pull/1487).

lib/logstorage/pipe_running_stats.go

@@ -103,16 +103,22 @@ func (ps *pipeRunningStats) isFixedOutputFieldsOrder() bool {
 }
 
 func (ps *pipeRunningStats) updateNeededFields(pf *prefixfilter.Filter) {
-	pfOrig := pf.Clone()
+	if pf.MatchNothing() {
+		return
+	}
 
+	pfOrig := pf.Clone()
 	for _, f := range ps.funcs {
 		pf.AddDenyFilter(f.resultName)
 		if pfOrig.MatchString(f.resultName) {
 			f.f.updateNeededFields(pf)
 		}
 	}
 
-	// byFields are needed unconditionally, since the output depends on them.
+	// The _time field is needed when the caller needs output fields, since running stats depend on row order.
+	pf.AddAllowFilter("_time")
+
+	// byFields are needed when the caller needs output fields, since the output depends on them.
 	for _, bf := range ps.byFields {
 		pf.AddAllowFilter(bf)
 	}
@@ -159,9 +165,21 @@ type pipeRunningStatsProcessor struct {
 	stateSizeBudget atomic.Int64
 }
 
+type pipeRunningStatsRow struct {
+	fields       []Field
+	timestampStr string
+	timestamp    int64
+	isTime       bool
+}
+
+const (
+	pipeRunningStatsRowSize    = int(unsafe.Sizeof(pipeRunningStatsRow{}))
+	pipeRunningStatsRowPtrSize = int(unsafe.Sizeof((*pipeRunningStatsRow)(nil)))
+)
+
 type pipeRunningStatsProcessorShard struct {
 	// rows tracks all the rows collected by the shard.
-	rows [][]Field
+	rows []pipeRunningStatsRow
 
 	columnValues [][]string
 
@@ -170,6 +188,10 @@ type pipeRunningStatsProcessorShard struct {
 
 func (shard *pipeRunningStatsProcessorShard) writeBlock(br *blockResult) {
 	cs := br.getColumns()
+	var timestamps []int64
+	if br.getColumnByName("_time").isTime {
+		timestamps = br.getTimestamps()
+	}
 
 	columnValues := slicesutil.SetLength(shard.columnValues, len(cs))
 	for i, c := range cs {
@@ -181,17 +203,33 @@ func (shard *pipeRunningStatsProcessorShard) writeBlock(br *blockResult) {
 		fields := make([]Field, len(cs))
 		shard.stateSizeBudget -= int(unsafe.Sizeof(fields[0])) * len(fields)
 
+		timestamp := ""
 		for j, c := range cs {
 			v := columnValues[j][rowIdx]
+			vCopy := strings.Clone(v)
 			fields[j] = Field{
 				Name:  strings.Clone(c.name),
-				Value: strings.Clone(v),
+				Value: vCopy,
 			}
 			shard.stateSizeBudget -= len(c.name) + len(v)
+			if c.name == "_time" {
+				timestamp = vCopy
+			}
 		}
 
-		shard.rows = append(shard.rows, fields)
-		shard.stateSizeBudget -= int(unsafe.Sizeof(fields))
+		row := pipeRunningStatsRow{
+			timestampStr: timestamp,
+			fields:       fields,
+		}
+		if timestamps != nil {
+			row.timestamp = timestamps[rowIdx]
+			row.isTime = true
+		} else {
+			row.timestamp, row.isTime = TryParseTimestampRFC3339Nano(timestamp)
+		}
+		shard.rows = append(shard.rows, row)
+		// Account for row metadata and the pointer added during flush.
+		shard.stateSizeBudget -= pipeRunningStatsRowSize + pipeRunningStatsRowPtrSize
 	}
 }
 
@@ -233,25 +271,17 @@ func (psp *pipeRunningStatsProcessor) flush() error {
 		return string(key)
 	}
 
-	type rowWithTimestamp struct {
-		timestamp string
-		fields    []Field
-	}
-
-	m := make(map[string][]rowWithTimestamp)
+	m := make(map[string][]*pipeRunningStatsRow)
 	shards := psp.shards.All()
 	for _, shard := range shards {
-		for _, row := range shard.rows {
+		for i := range shard.rows {
 			if needStop(psp.stopCh) {
 				return nil
 			}
 
-			key := getKeyForRow(row)
-			timestamp := getFieldValueByName(row, "_time")
-			m[key] = append(m[key], rowWithTimestamp{
-				timestamp: timestamp,
-				fields:    row,
-			})
+			row := &shard.rows[i]
+			key := getKeyForRow(row.fields)
+			m[key] = append(m[key], row)
 		}
 	}
 
@@ -271,7 +301,7 @@ func (psp *pipeRunningStatsProcessor) flush() error {
 	for _, key := range keys {
 		rows := m[key]
 		sort.Slice(rows, func(i, j int) bool {
-			return rows[i].timestamp < rows[j].timestamp
+			return lessRunningStatsRow(rows[i], rows[j])
 		})
 
 		if needStop(psp.stopCh) {
@@ -314,6 +344,16 @@ func (psp *pipeRunningStatsProcessor) flush() error {
 	return nil
 }
 
+func lessRunningStatsRow(a, b *pipeRunningStatsRow) bool {
+	if a.isTime != b.isTime {
+		return a.isTime
+	}
+	if a.isTime && a.timestamp != b.timestamp {
+		return a.timestamp < b.timestamp
+	}
+	return a.timestampStr < b.timestampStr
+}
+
 type pipeRunningStatsWriter struct {
 	ppNext pipeProcessor
 

lib/logstorage/pipe_running_stats_test.go

@@ -266,6 +266,32 @@ func TestPipeRunningStats(t *testing.T) {
 			{"min_c", ""},
 		},
 	})
+
+	// timestamps with a different number of digits after the decimal point
+	f("running_stats count() as running_count", [][]Field{
+		{
+			{"_time", "2026-03-31T12:00:45.990844Z"},
+		},
+		{
+			{"_time", "2026-03-31T12:00:45.99Z"},
+		},
+		{
+			{"_time", "2026-03-31T12:00:45.999324Z"},
+		},
+	}, [][]Field{
+		{
+			{"_time", "2026-03-31T12:00:45.99Z"},
+			{"running_count", "1"},
+		},
+		{
+			{"_time", "2026-03-31T12:00:45.990844Z"},
+			{"running_count", "2"},
+		},
+		{
+			{"_time", "2026-03-31T12:00:45.999324Z"},
+			{"running_count", "3"},
+		},
+	})
 }
 
 func TestPipeRunningStatsUpdateNeededFields(t *testing.T) {
@@ -274,6 +300,10 @@ func TestPipeRunningStatsUpdateNeededFields(t *testing.T) {
 		expectPipeNeededFields(t, s, allowFilters, denyFilters, allowFiltersExpected, denyFiltersExpected)
 	}
 
+	// no fields are needed
+	f("running_stats count() r1", "", "", "", "")
+	f("running_stats by (b1,b2) count(f1,f2) r1", "", "", "", "")
+
 	// all the needed fields
 	f("running_stats count() r1", "*", "", "*", "r1")
 	f("running_stats count(*) r1", "*", "", "*", "r1")
@@ -302,18 +332,18 @@ func TestPipeRunningStatsUpdateNeededFields(t *testing.T) {
 	f("running_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "*", "r1,r3", "*", "r1,r2,r3")
 
 	// needed fields do not intersect with stats fields
-	f("running_stats count() r1", "r2", "", "r2", "")
-	f("running_stats count(*) r1", "r2", "", "r2", "")
-	f("running_stats count(f1,f2) r1", "r2", "", "r2", "")
-	f("running_stats count(f1,f2) r1, sum(f3,f4) r2", "r3", "", "r3", "")
-	f("running_stats by (b1,b2) count(f1,f2) r1", "r2", "", "b1,b2,r2", "")
-	f("running_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "r3", "", "b1,b2,r3", "")
+	f("running_stats count() r1", "r2", "", "_time,r2", "")
+	f("running_stats count(*) r1", "r2", "", "_time,r2", "")
+	f("running_stats count(f1,f2) r1", "r2", "", "_time,r2", "")
+	f("running_stats count(f1,f2) r1, sum(f3,f4) r2", "r3", "", "_time,r3", "")
+	f("running_stats by (b1,b2) count(f1,f2) r1", "r2", "", "_time,b1,b2,r2", "")
+	f("running_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "r3", "", "_time,b1,b2,r3", "")
 
 	// needed fields intersect with stats fields
-	f("running_stats count() r1", "r1,r2", "", "r2", "")
-	f("running_stats count(*) r1", "r1,r2", "", "r2", "")
-	f("running_stats count(f1,f2) r1", "r1,r2", "", "f1,f2,r2", "")
-	f("running_stats count(f1,f2) r1, sum(f3,f4) r2", "r1,r3", "", "f1,f2,r3", "")
-	f("running_stats by (b1,b2) count(f1,f2) r1", "r1,r2", "", "b1,b2,f1,f2,r2", "")
-	f("running_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "r1,r3", "", "b1,b2,f1,f2,r3", "")
+	f("running_stats count() r1", "r1,r2", "", "_time,r2", "")
+	f("running_stats count(*) r1", "r1,r2", "", "_time,r2", "")
+	f("running_stats count(f1,f2) r1", "r1,r2", "", "_time,f1,f2,r2", "")
+	f("running_stats count(f1,f2) r1, sum(f3,f4) r2", "r1,r3", "", "_time,f1,f2,r3", "")
+	f("running_stats by (b1,b2) count(f1,f2) r1", "r1,r2", "", "_time,b1,b2,f1,f2,r2", "")
+	f("running_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "r1,r3", "", "_time,b1,b2,f1,f2,r3", "")
 }

lib/logstorage/pipe_total_stats_test.go

@@ -257,6 +257,41 @@ func TestPipeTotalStats(t *testing.T) {
 			{"min_c", ""},
 		},
 	})
+
+	// timestamps with a different number of digits after the decimal point
+	f("total_stats first(a) as first_a, last(a) as last_a", [][]Field{
+		{
+			{"_time", "2026-03-31T12:00:45.990844Z"},
+			{"a", "middle"},
+		},
+		{
+			{"_time", "2026-03-31T12:00:45.99Z"},
+			{"a", "first"},
+		},
+		{
+			{"_time", "2026-03-31T12:00:45.999324Z"},
+			{"a", "last"},
+		},
+	}, [][]Field{
+		{
+			{"_time", "2026-03-31T12:00:45.99Z"},
+			{"a", "first"},
+			{"first_a", "first"},
+			{"last_a", "last"},
+		},
+		{
+			{"_time", "2026-03-31T12:00:45.990844Z"},
+			{"a", "middle"},
+			{"first_a", "first"},
+			{"last_a", "last"},
+		},
+		{
+			{"_time", "2026-03-31T12:00:45.999324Z"},
+			{"a", "last"},
+			{"first_a", "first"},
+			{"last_a", "last"},
+		},
+	})
 }
 
 func TestPipeTotalStatsUpdateNeededFields(t *testing.T) {
@@ -293,18 +328,18 @@ func TestPipeTotalStatsUpdateNeededFields(t *testing.T) {
 	f("total_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "*", "r1,r3", "*", "r1,r2,r3")
 
 	// needed fields do not intersect with stats fields
-	f("total_stats count() r1", "r2", "", "r2", "")
-	f("total_stats count(*) r1", "r2", "", "r2", "")
-	f("total_stats count(f1,f2) r1", "r2", "", "r2", "")
-	f("total_stats count(f1,f2) r1, sum(f3,f4) r2", "r3", "", "r3", "")
-	f("total_stats by (b1,b2) count(f1,f2) r1", "r2", "", "b1,b2,r2", "")
-	f("total_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "r3", "", "b1,b2,r3", "")
+	f("total_stats count() r1", "r2", "", "_time,r2", "")
+	f("total_stats count(*) r1", "r2", "", "_time,r2", "")
+	f("total_stats count(f1,f2) r1", "r2", "", "_time,r2", "")
+	f("total_stats count(f1,f2) r1, sum(f3,f4) r2", "r3", "", "_time,r3", "")
+	f("total_stats by (b1,b2) count(f1,f2) r1", "r2", "", "_time,b1,b2,r2", "")
+	f("total_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "r3", "", "_time,b1,b2,r3", "")
 
 	// needed fields intersect with stats fields
-	f("total_stats count() r1", "r1,r2", "", "r2", "")
-	f("total_stats count(*) r1", "r1,r2", "", "r2", "")
-	f("total_stats count(f1,f2) r1", "r1,r2", "", "f1,f2,r2", "")
-	f("total_stats count(f1,f2) r1, sum(f3,f4) r2", "r1,r3", "", "f1,f2,r3", "")
-	f("total_stats by (b1,b2) count(f1,f2) r1", "r1,r2", "", "b1,b2,f1,f2,r2", "")
-	f("total_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "r1,r3", "", "b1,b2,f1,f2,r3", "")
+	f("total_stats count() r1", "r1,r2", "", "_time,r2", "")
+	f("total_stats count(*) r1", "r1,r2", "", "_time,r2", "")
+	f("total_stats count(f1,f2) r1", "r1,r2", "", "_time,f1,f2,r2", "")
+	f("total_stats count(f1,f2) r1, sum(f3,f4) r2", "r1,r3", "", "_time,f1,f2,r3", "")
+	f("total_stats by (b1,b2) count(f1,f2) r1", "r1,r2", "", "_time,b1,b2,f1,f2,r2", "")
+	f("total_stats by (b1,b2) count(f1,f2) r1, count(f1,f3) r2", "r1,r3", "", "_time,b1,b2,f1,f2,r3", "")
 }