Merge branch 'master' into vmalert-proxy

Author: AndrewChubatiuk

docs/victorialogs/CHANGELOG.md

@@ -21,6 +21,10 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
 * FEATURE: proxy VMAlert requests at `/select/vmalert` path, when `-vmalert.proxyURL` flag is set. See [#8272](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8272).
 * FEATURE: add `-search.disableTenantResponseHeaders` flag to disable showing tenant information on VMUI for case, when AccountID and ProjectID headers are set by some proxy.
 
+* BUGFIX: [`/select/logsql/query` endpoint](https://docs.victoriametrics.com/victorialogs/querying/#querying-logs): properly optimize the query execution when the `limit` query arg is set, while the `offset` query arg isn't set. It wasn't addressed properly in [v1.33.0](#v1330). See [#620](https://github.com/VictoriaMetrics/VictoriaLogs/issues/620).
+* BUGFIX: [`/select/logsql/hits` endpoint](https://docs.victoriametrics.com/victorialogs/querying/#querying-hits-stats): take into account results from the [`union` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#union-pipe) while calculating log hits. See [#641](https://github.com/VictoriaMetrics/VictoriaLogs/issues/641).
+* BUGFIX: [`/select/logsql/hits` endpoint](https://docs.victoriametrics.com/victorialogs/querying/#querying-hits-stats): take into account results from the [`join` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#join-pipe) while calculating log hits.
+
 ## [v1.33.0](https://github.com/VictoriaMetrics/VictoriaLogs/releases/tag/v1.33.0)
 
 Released at 2025-09-10

lib/logstorage/net_query_runner_test.go

@@ -84,4 +84,9 @@ func TestSplitQueryToRemoteAndLocal(t *testing.T) {
 	f(`foo | unpack_syslog`, `foo | unpack_syslog`, ``)
 	f(`foo | unpack_words`, `foo | unpack_words`, ``)
 	f(`foo | unroll by (x)`, `foo | unroll by (x)`, ``)
+
+	// Special cases with 'offset 0'
+	// See https://github.com/VictoriaMetrics/VictoriaLogs/issues/620#issuecomment-3276624504
+	f(`foo | offset 0`, `foo`, ``)
+	f(`foo | offset 0 | limit 10`, `foo | limit 10`, `limit 10`)
 }

lib/logstorage/parser.go

@@ -511,7 +511,7 @@ func (q *Query) AddFacetsPipe(limit, maxValuesPerField, maxValueLen int, keepCon
 // AddCountByTimePipe adds '| stats by (_time:step offset off, field1, ..., fieldN) count() hits' to the end of q.
 func (q *Query) AddCountByTimePipe(step, off int64, fields []string) {
 	// Drop pipes from q, which modify or delete _time field, since they make impossible to calculate stats grouped by _time.
-	q.dropTimeModificationPipes()
+	q.dropPipesUnsafeForHits()
 
 	{
 		// add 'stats by (_time:step offset off, fields) count() hits'
@@ -541,10 +541,11 @@ func (q *Query) AddCountByTimePipe(step, off int64, fields []string) {
 	}
 }
 
-// dropTimeModificationPipes drops pipes from q, which modify
-func (q *Query) dropTimeModificationPipes() {
+// dropPipesUnsafeForHits drops trailing pipes from q, which are unsafe
+// for calculating hits grouped by _time.
+func (q *Query) dropPipesUnsafeForHits() {
 	for i, p := range q.pipes {
-		if !p.canReturnLastNResults() {
+		if !isPipeSafeForHits(p) {
 			// Drop the rest of the pipes, including the current pipe,
 			// since it modified or deletes the _time field.
 			q.pipes = q.pipes[:i]
@@ -553,6 +554,25 @@ func (q *Query) dropTimeModificationPipes() {
 	}
 }
 
+func isPipeSafeForHits(p pipe) bool {
+	if p.canReturnLastNResults() {
+		return true
+	}
+
+	switch t := p.(type) {
+	case *pipeUnion:
+		// Allow union pipes, but drop pipes unsafe for hits inside them.
+		// See https://github.com/VictoriaMetrics/VictoriaLogs/issues/641
+		t.q.dropPipesUnsafeForHits()
+		return true
+	case *pipeJoin:
+		// Allow join pipes, since they do not drop _time field.
+		return true
+	default:
+		return false
+	}
+}
+
 // Clone returns a copy of q at the given timestamp.
 func (q *Query) Clone(timestamp int64) *Query {
 	qStr := q.String()
@@ -790,7 +810,7 @@ func (q *Query) AddPipeOffsetLimit(offset, limit uint64) {
 	q.mustAppendPipe(limitStr)
 
 	// optimize the query, so the `offset` and `limit` pipes could be joined with the preceding `sort` pipe.
-	q.pipes = optimizeSortOffsetLimitPipes(q.pipes)
+	q.pipes = optimizeOffsetLimitPipes(q.pipes)
 }
 
 func (q *Query) mustAppendPipe(s string) {
@@ -807,7 +827,7 @@ func (q *Query) optimize() {
 }
 
 func (q *Query) optimizeNoSubqueries() {
-	q.pipes = optimizeSortOffsetLimitPipes(q.pipes)
+	q.pipes = optimizeOffsetLimitPipes(q.pipes)
 	q.pipes = optimizeUniqLimitPipes(q.pipes)
 	q.pipes = optimizeFilterPipes(q.pipes)
 
@@ -1323,15 +1343,28 @@ func removeStarFilters(f filter) filter {
 	return f
 }
 
-func optimizeSortOffsetLimitPipes(pipes []pipe) []pipe {
+func optimizeOffsetLimitPipes(pipes []pipe) []pipe {
 	for {
 		pipesLen := len(pipes)
 		pipes = optimizeSortOffsetPipes(pipes)
 		pipes = optimizeSortLimitPipes(pipes)
 		if len(pipes) == pipesLen {
-			return pipes
+			break
 		}
 	}
+
+	// Remove `offset 0` pipes.
+	i := 0
+	for i < len(pipes) {
+		po, ok := pipes[i].(*pipeOffset)
+		if !ok || po.offset != 0 {
+			i++
+			continue
+		}
+		pipes = append(pipes[:i], pipes[i+1:]...)
+	}
+
+	return pipes
 }
 
 func optimizeSortOffsetPipes(pipes []pipe) []pipe {

lib/logstorage/parser_test.go

@@ -355,7 +355,7 @@ func TestQuery_AddTimeFilter_StepPropagation(t *testing.T) {
 	})
 }
 
-func TestParseQuery_OptimizeSortOffsetLimitPipes(t *testing.T) {
+func TestParseQuery_OptimizeOffsetLimitPipes(t *testing.T) {
 	f := func(s, resultExpected string) {
 		t.Helper()
 
@@ -385,6 +385,11 @@ func TestParseQuery_OptimizeSortOffsetLimitPipes(t *testing.T) {
 	f(`* | sort by (x) | limit 30 | limit 20 | offset 4`, `* | sort by (x) offset 4 limit 16`)
 	f(`* | sort by (x) | limit 30 | limit 20 | offset 4 | offset 5`, `* | sort by (x) offset 9 limit 11`)
 	f(`* | sort by (x) | limit 30 | limit 20 | offset 4 | offset 5 | fields x`, `* | sort by (x) offset 9 limit 11 | fields x`)
+
+	// Verify the case without 'sort' pipe and with 'offset 0' pipes.
+	// See https://github.com/VictoriaMetrics/VictoriaLogs/issues/620#issuecomment-3276624504
+	f(`* | offset 0`, `*`)
+	f(`* | offset 0 | limit 10`, `* | limit 10`)
 }
 
 func TestParseQuery_OptimizeStarFilters(t *testing.T) {
@@ -3932,6 +3937,14 @@ func TestQuery_AddCountByTimePipe(t *testing.T) {
 	// pipes, which do not change _time field
 	f("* | extract 'abc<de>fg' | filter de:='qwer'", nsecsPerMinute, 0, nil, `* | extract "abc<de>fg" | filter de:=qwer | stats by (_time:1m) count(*) as hits | sort by (_time)`)
 
+	// union pipe is allowed. See https://github.com/VictoriaMetrics/VictoriaLogs/issues/641
+	f(`foo | union (bar)`, nsecsPerMinute, 0, nil, `foo | union (bar) | stats by (_time:1m) count(*) as hits | sort by (_time)`)
+	f(`foo | union (bar) | stats count()`, nsecsPerMinute, 0, nil, `foo | union (bar) | stats by (_time:1m) count(*) as hits | sort by (_time)`)
+	f(`foo | union (bar | stats count())`, nsecsPerMinute, 0, nil, `foo | union (bar) | stats by (_time:1m) count(*) as hits | sort by (_time)`)
+
+	// join pipe is allowed
+	f(`foo | join by (x) (y)`, nsecsPerMinute, 0, nil, `foo | join by (x) (y) | stats by (_time:1m) count(*) as hits | sort by (_time)`)
+
 	// pipes, which change _time field
 	f("* | extract 'abc<de>fg' | filter de:='qwer' | stats count()", nsecsPerMinute, 0, nil, `* | extract "abc<de>fg" | filter de:=qwer | stats by (_time:1m) count(*) as hits | sort by (_time)`)
 	f("* | extract 'abc<de>fg' | sort by (x)", nsecsPerMinute, 0, nil, `* | extract "abc<de>fg" | stats by (_time:1m) count(*) as hits | sort by (_time)`)

lib/logstorage/pipe_offset.go

@@ -19,6 +19,10 @@ func (po *pipeOffset) String() string {
 }
 
 func (po *pipeOffset) splitToRemoteAndLocal(_ int64) (pipe, []pipe) {
+	if po.offset == 0 {
+		// Special case - `offset 0` is safe to push to the remote side.
+		return po, nil
+	}
 	return nil, []pipe{po}
 }