diff --git a/docs/victorialogs/CHANGELOG.md b/docs/victorialogs/CHANGELOG.md index 6196803026..9c2b207323 100644 --- a/docs/victorialogs/CHANGELOG.md +++ b/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). diff --git a/lib/logstorage/pipe_running_stats.go b/lib/logstorage/pipe_running_stats.go index 76c9a2009b..4832d2a6e3 100644 --- a/lib/logstorage/pipe_running_stats.go +++ b/lib/logstorage/pipe_running_stats.go @@ -103,8 +103,11 @@ 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) { @@ -112,7 +115,10 @@ func (ps *pipeRunningStats) updateNeededFields(pf *prefixfilter.Filter) { } } - // 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 diff --git a/lib/logstorage/pipe_running_stats_test.go b/lib/logstorage/pipe_running_stats_test.go index 14c82bdc16..f263004d37 100644 --- a/lib/logstorage/pipe_running_stats_test.go +++ b/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", "") } diff --git a/lib/logstorage/pipe_total_stats_test.go b/lib/logstorage/pipe_total_stats_test.go index 52406cd745..1a783140b8 100644 --- a/lib/logstorage/pipe_total_stats_test.go +++ b/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", "") }