Repository: VictoriaMetrics/VictoriaMetrics
Pull request #7487 (2024-11-08 04:22:43)

docs: [VictoriaLogs] FAQ add logs without msg field
### Describe Your Changes

Add FAQs to VictoriaLogs:
- I want to ingest logs without message field, is that possible?
- What if my logs have multiple message fields candidates

Preview: https://github.com/VictoriaMetrics/VictoriaMetrics/blob/docs/VL-FAQ-empty-msg/docs/VictoriaLogs/FAQ.md#i-want-to-ingest-logs-without-message-field-is-that-possible

### Checklist

The following checks are **mandatory**:

- [x] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).



Patch:
diff --git a/app/vlinsert/insertutils/common_params.go b/app/vlinsert/insertutils/common_params.go
index 0077a19a0..c357d3f87 100644
--- a/app/vlinsert/insertutils/common_params.go
+++ b/app/vlinsert/insertutils/common_params.go
@@ -112,19 +112,24 @@ func getExtraFields(r *http.Request) ([]logstorage.Field, error) {
 }
 
 // GetCommonParamsForSyslog returns common params needed for parsing syslog messages and storing them to the given tenantID.
-func GetCommonParamsForSyslog(tenantID logstorage.TenantID) *CommonParams {
+func GetCommonParamsForSyslog(tenantID logstorage.TenantID, streamFields, ignoreFields []string, extraFields []logstorage.Field) *CommonParams {
 	// See https://docs.victoriametrics.com/victorialogs/logsql/#unpack_syslog-pipe
+	if streamFields == nil {
+		streamFields = []string{
+			"hostname",
+			"app_name",
+			"proc_id",
+		}
+	}
 	cp := &CommonParams{
 		TenantID:  tenantID,
 		TimeField: "timestamp",
 		MsgFields: []string{
 			"message",
 		},
-		StreamFields: []string{
-			"hostname",
-			"app_name",
-			"proc_id",
-		},
+		StreamFields: streamFields,
+		IgnoreFields: ignoreFields,
+		ExtraFields:  extraFields,
 	}
 
 	return cp
diff --git a/app/vlinsert/loki/loki_json.go b/app/vlinsert/loki/loki_json.go
index 5b178ef40..768691c11 100644
--- a/app/vlinsert/loki/loki_json.go
+++ b/app/vlinsert/loki/loki_json.go
@@ -57,7 +57,7 @@ func handleJSON(r *http.Request, w http.ResponseWriter) {
 	n, err := parseJSONRequest(data, lmp)
 	lmp.MustClose()
 	if err != nil {
-		httpserver.Errorf(w, r, "cannot parse Loki json request: %s", err)
+		httpserver.Errorf(w, r, "cannot parse Loki json request: %s; data=%s", err, data)
 		return
 	}
 
@@ -85,7 +85,7 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor) (int, er
 
 	streamsV := v.Get("streams")
 	if streamsV == nil {
-		return 0, fmt.Errorf("missing `streams` item in the parsed JSON: %q", v)
+		return 0, fmt.Errorf("missing `streams` item in the parsed JSON")
 	}
 	streams, err := streamsV.Array()
 	if err != nil {
@@ -125,7 +125,7 @@ func parseJSONRequest(data []byte, lmp insertutils.LogMessageProcessor) (int, er
 		// populate messages from `values` array
 		linesV := stream.Get("values")
 		if linesV == nil {
-			return rowsIngested, fmt.Errorf("missing `values` item in the parsed JSON %q", stream)
+			return rowsIngested, fmt.Errorf("missing `values` item in the parsed `stream` object %q", stream)
 		}
 		lines, err := linesV.Array()
 		if err != nil {
diff --git a/app/vlinsert/syslog/syslog.go b/app/vlinsert/syslog/syslog.go
index 83bf6944c..7211204ab 100644
--- a/app/vlinsert/syslog/syslog.go
+++ b/app/vlinsert/syslog/syslog.go
@@ -3,11 +3,13 @@ package syslog
 import (
 	"bufio"
 	"crypto/tls"
+	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
 	"io"
 	"net"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -35,10 +37,25 @@ var (
 	syslogTimezone = flag.String("syslog.timezone", "Local", "Timezone to use when parsing timestamps in RFC3164 syslog messages. Timezone must be a valid IANA Time Zone. "+
 		"For example: America/New_York, Europe/Berlin, Etc/GMT+3 . See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/")
 
-	syslogTenantIDTCP = flagutil.NewArrayString("syslog.tenantID.tcp", "TenantID for logs ingested via the corresponding -syslog.listenAddr.tcp. "+
-		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/")
-	syslogTenantIDUDP = flagutil.NewArrayString("syslog.tenantID.udp", "TenantID for logs ingested via the corresponding -syslog.listenAddr.udp. "+
-		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/")
+	streamFieldsTCP = flagutil.NewArrayString("syslog.streamFields.tcp", "Fields to use as log stream labels for logs ingested via the corresponding -syslog.listenAddr.tcp. "+
+		`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#stream-fields`)
+	streamFieldsUDP = flagutil.NewArrayString("syslog.streamFields.udp", "Fields to use as log stream labels for logs ingested via the corresponding -syslog.listenAddr.udp. "+
+		`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#stream-fields`)
+
+	ignoreFieldsTCP = flagutil.NewArrayString("syslog.ignoreFields.tcp", "Fields to ignore at logs ingested via the corresponding -syslog.listenAddr.tcp. "+
+		`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#dropping-fields`)
+	ignoreFieldsUDP = flagutil.NewArrayString("syslog.ignoreFields.udp", "Fields to ignore at logs ingested via the corresponding -syslog.listenAddr.udp. "+
+		`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#dropping-fields`)
+
+	extraFieldsTCP = flagutil.NewArrayString("syslog.extraFields.tcp", "Fields to add to logs ingested via the corresponding -syslog.listenAddr.tcp. "+
+		`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#adding-extra-fields`)
+	extraFieldsUDP = flagutil.NewArrayString("syslog.extraFields.udp", "Fields to add to logs ingested via the corresponding -syslog.listenAddr.udp. "+
+		`See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#adding-extra-fields`)
+
+	tenantIDTCP = flagutil.NewArrayString("syslog.tenantID.tcp", "TenantID for logs ingested via the corresponding -syslog.listenAddr.tcp. "+
+		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#multitenancy")
+	tenantIDUDP = flagutil.NewArrayString("syslog.tenantID.udp", "TenantID for logs ingested via the corresponding -syslog.listenAddr.udp. "+
+		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#multitenancy")
 
 	listenAddrTCP = flagutil.NewArrayString("syslog.listenAddr.tcp", "Comma-separated list of TCP addresses to listen to for Syslog messages. "+
 		"See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/")
@@ -150,7 +167,7 @@ func runUDPListener(addr string, argIdx int) {
 		logger.Fatalf("cannot start UDP syslog server at %q: %s", addr, err)
 	}
 
-	tenantIDStr := syslogTenantIDUDP.GetOptionalArg(argIdx)
+	tenantIDStr := tenantIDUDP.GetOptionalArg(argIdx)
 	tenantID, err := logstorage.ParseTenantID(tenantIDStr)
 	if err != nil {
 		logger.Fatalf("cannot parse -syslog.tenantID.udp=%q for -syslog.listenAddr.udp=%q: %s", tenantIDStr, addr, err)
@@ -161,9 +178,27 @@ func runUDPListener(addr string, argIdx int) {
 
 	useLocalTimestamp := useLocalTimestampUDP.GetOptionalArg(argIdx)
 
+	streamFieldsStr := streamFieldsUDP.GetOptionalArg(argIdx)
+	streamFields, err := parseFieldsList(streamFieldsStr)
+	if err != nil {
+		logger.Fatalf("cannot parse -syslog.streamFields.udp=%q for -syslog.listenAddr.udp=%q: %s", streamFieldsStr, addr, err)
+	}
+
+	ignoreFieldsStr := ignoreFieldsUDP.GetOptionalArg(argIdx)
+	ignoreFields, err := parseFieldsList(ignoreFieldsStr)
+	if err != nil {
+		logger.Fatalf("cannot parse -syslog.ignoreFields.udp=%q for -syslog.listenAddr.udp=%q: %s", ignoreFieldsStr, addr, err)
+	}
+
+	extraFieldsStr := extraFieldsUDP.GetOptionalArg(argIdx)
+	extraFields, err := parseExtraFields(extraFieldsStr)
+	if err != nil {
+		logger.Fatalf("cannot parse -syslog.extraFields.udp=%q for -syslog.listenAddr.udp=%q: %s", extraFieldsStr, addr, err)
+	}
+
 	doneCh := make(chan struct{})
 	go func() {
-		serveUDP(ln, tenantID, compressMethod, useLocalTimestamp)
+		serveUDP(ln, tenantID, compressMethod, useLocalTimestamp, streamFields, ignoreFields, extraFields)
 		close(doneCh)
 	}()
 
@@ -193,7 +228,7 @@ func runTCPListener(addr string, argIdx int) {
 		logger.Fatalf("syslog: cannot start TCP listener at %s: %s", addr, err)
 	}
 
-	tenantIDStr := syslogTenantIDTCP.GetOptionalArg(argIdx)
+	tenantIDStr := tenantIDTCP.GetOptionalArg(argIdx)
 	tenantID, err := logstorage.ParseTenantID(tenantIDStr)
 	if err != nil {
 		logger.Fatalf("cannot parse -syslog.tenantID.tcp=%q for -syslog.listenAddr.tcp=%q: %s", tenantIDStr, addr, err)
@@ -204,9 +239,27 @@ func runTCPListener(addr string, argIdx int) {
 
 	useLocalTimestamp := useLocalTimestampTCP.GetOptionalArg(argIdx)
 
+	streamFieldsStr := streamFieldsTCP.GetOptionalArg(argIdx)
+	streamFields, err := parseFieldsList(streamFieldsStr)
+	if err != nil {
+		logger.Fatalf("cannot parse -syslog.streamFields.tcp=%q for -syslog.listenAddr.tcp=%q: %s", streamFieldsStr, addr, err)
+	}
+
+	ignoreFieldsStr := ignoreFieldsTCP.GetOptionalArg(argIdx)
+	ignoreFields, err := parseFieldsList(ignoreFieldsStr)
+	if err != nil {
+		logger.Fatalf("cannot parse -syslog.ignoreFields.tcp=%q for -syslog.listenAddr.tcp=%q: %s", ignoreFieldsStr, addr, err)
+	}
+
+	extraFieldsStr := extraFieldsTCP.GetOptionalArg(argIdx)
+	extraFields, err := parseExtraFields(extraFieldsStr)
+	if err != nil {
+		logger.Fatalf("cannot parse -syslog.extraFields.tcp=%q for -syslog.listenAddr.tcp=%q: %s", extraFieldsStr, addr, err)
+	}
+
 	doneCh := make(chan struct{})
 	go func() {
-		serveTCP(ln, tenantID, compressMethod, useLocalTimestamp)
+		serveTCP(ln, tenantID, compressMethod, useLocalTimestamp, streamFields, ignoreFields, extraFields)
 		close(doneCh)
 	}()
 
@@ -228,7 +281,7 @@ func checkCompressMethod(compressMethod, addr, protocol string) {
 	}
 }
 
-func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, compressMethod string, useLocalTimestamp bool) {
+func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, compressMethod string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
 	gomaxprocs := cgroup.AvailableCPUs()
 	var wg sync.WaitGroup
 	localAddr := ln.LocalAddr()
@@ -236,7 +289,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, compressMethod st
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			cp := insertutils.GetCommonParamsForSyslog(tenantID)
+			cp := insertutils.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
 			var bb bytesutil.ByteBuffer
 			bb.B = bytesutil.ResizeNoCopyNoOverallocate(bb.B, 64*1024)
 			for {
@@ -270,7 +323,7 @@ func serveUDP(ln net.PacketConn, tenantID logstorage.TenantID, compressMethod st
 	wg.Wait()
 }
 
-func serveTCP(ln net.Listener, tenantID logstorage.TenantID, compressMethod string, useLocalTimestamp bool) {
+func serveTCP(ln net.Listener, tenantID logstorage.TenantID, compressMethod string, useLocalTimestamp bool, streamFields, ignoreFields []string, extraFields []logstorage.Field) {
 	var cm ingestserver.ConnsMap
 	cm.Init("syslog")
 
@@ -300,7 +353,7 @@ func serveTCP(ln net.Listener, tenantID logstorage.TenantID, compressMethod stri
 
 		wg.Add(1)
 		go func() {
-			cp := insertutils.GetCommonParamsForSyslog(tenantID)
+			cp := insertutils.GetCommonParamsForSyslog(tenantID, streamFields, ignoreFields, extraFields)
 			if err := processStream(c, compressMethod, useLocalTimestamp, cp); err != nil {
 				logger.Errorf("syslog: cannot process TCP data at %q: %s", addr, err)
 			}
@@ -531,3 +584,35 @@ var (
 	udpRequestsTotal = metrics.NewCounter(`vl_udp_reqests_total{type="syslog"}`)
 	udpErrorsTotal   = metrics.NewCounter(`vl_udp_errors_total{type="syslog"}`)
 )
+
+func parseFieldsList(s string) ([]string, error) {
+	if s == "" {
+		return nil, nil
+	}
+
+	var a []string
+	err := json.Unmarshal([]byte(s), &a)
+	return a, err
+}
+
+func parseExtraFields(s string) ([]logstorage.Field, error) {
+	if s == "" {
+		return nil, nil
+	}
+
+	var m map[string]string
+	if err := json.Unmarshal([]byte(s), &m); err != nil {
+		return nil, err
+	}
+	fields := make([]logstorage.Field, 0, len(m))
+	for k, v := range m {
+		fields = append(fields, logstorage.Field{
+			Name:  k,
+			Value: v,
+		})
+	}
+	sort.Slice(fields, func(i, j int) bool {
+		return fields[i].Name < fields[j].Name
+	})
+	return fields, nil
+}
diff --git a/app/vlogscli/less_wrapper.go b/app/vlogscli/less_wrapper.go
index d7a0d5637..cecb25149 100644
--- a/app/vlogscli/less_wrapper.go
+++ b/app/vlogscli/less_wrapper.go
@@ -17,7 +17,7 @@ func isTerminal() bool {
 	return isatty.IsTerminal(os.Stdout.Fd()) && isatty.IsTerminal(os.Stderr.Fd())
 }
 
-func readWithLess(r io.Reader) error {
+func readWithLess(r io.Reader, wrapLongLines bool) error {
 	if !isTerminal() {
 		// Just write everything to stdout if no terminal is available.
 		_, err := io.Copy(os.Stdout, r)
@@ -48,7 +48,11 @@ func readWithLess(r io.Reader) error {
 	if err != nil {
 		return fmt.Errorf("cannot find 'less' command: %w", err)
 	}
-	p, err := os.StartProcess(path, []string{"less", "-F", "-X"}, &os.ProcAttr{
+	opts := []string{"less", "-F", "-X"}
+	if !wrapLongLines {
+		opts = append(opts, "-S")
+	}
+	p, err := os.StartProcess(path, opts, &os.ProcAttr{
 		Env:   append(os.Environ(), "LESSCHARSET=utf-8"),
 		Files: []*os.File{pr, os.Stdout, os.Stderr},
 	})
diff --git a/app/vlogscli/main.go b/app/vlogscli/main.go
index 8799ab199..febb1cb27 100644
--- a/app/vlogscli/main.go
+++ b/app/vlogscli/main.go
@@ -91,6 +91,7 @@ func runReadlineLoop(rl *readline.Instance, incompleteLine *string) {
 	}
 
 	outputMode := outputModeJSONMultiline
+	wrapLongLines := false
 	s := ""
 	for {
 		line, err := rl.ReadLine()
@@ -99,7 +100,7 @@ func runReadlineLoop(rl *readline.Instance, incompleteLine *string) {
 			case io.EOF:
 				if s != "" {
 					// This is non-interactive query execution.
-					executeQuery(context.Background(), rl, s, outputMode)
+					executeQuery(context.Background(), rl, s, outputMode, wrapLongLines)
 				}
 				return
 			case readline.ErrInterrupt:
@@ -163,6 +164,18 @@ func runReadlineLoop(rl *readline.Instance, incompleteLine *string) {
 			s = ""
 			continue
 		}
+		if s == `\wrap_long_lines` {
+			if wrapLongLines {
+				wrapLongLines = false
+				fmt.Fprintf(rl, "wrapping of long lines is disabled\n")
+			} else {
+				wrapLongLines = true
+				fmt.Fprintf(rl, "wrapping of long lines is enabled\n")
+			}
+			historyLines = pushToHistory(rl, historyLines, s)
+			s = ""
+			continue
+		}
 		if line != "" && !strings.HasSuffix(line, ";") {
 			// Assume the query is incomplete and allow the user finishing the query on the next line
 			s += "\n"
@@ -172,7 +185,7 @@ func runReadlineLoop(rl *readline.Instance, incompleteLine *string) {
 
 		// Execute the query
 		ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
-		executeQuery(ctx, rl, s, outputMode)
+		executeQuery(ctx, rl, s, outputMode, wrapLongLines)
 		cancel()
 
 		historyLines = pushToHistory(rl, historyLines, s)
@@ -259,13 +272,14 @@ func printCommandsHelp(w io.Writer) {
 \m - multiline json output mode
 \c - compact output
 \logfmt - logfmt output mode
+\wrap_long_lines - toggles wrapping long lines
 \tail <query> - live tail <query> results
 
 See https://docs.victoriametrics.com/victorialogs/querying/vlogscli/ for more details
 `)
 }
 
-func executeQuery(ctx context.Context, output io.Writer, qStr string, outputMode outputMode) {
+func executeQuery(ctx context.Context, output io.Writer, qStr string, outputMode outputMode, wrapLongLines bool) {
 	if strings.HasPrefix(qStr, `\tail `) {
 		tailQuery(ctx, output, qStr, outputMode)
 		return
@@ -279,7 +293,7 @@ func executeQuery(ctx context.Context, output io.Writer, qStr string, outputMode
 		_ = respBody.Close()
 	}()
 
-	if err := readWithLess(respBody); err != nil {
+	if err := readWithLess(respBody, wrapLongLines); err != nil {
 		fmt.Fprintf(output, "error when reading query response: %s\n", err)
 		return
 	}
diff --git a/app/vlselect/logsql/logsql.go b/app/vlselect/logsql/logsql.go
index 6113918bd..f4b8a86e9 100644
--- a/app/vlselect/logsql/logsql.go
+++ b/app/vlselect/logsql/logsql.go
@@ -73,7 +73,6 @@ func ProcessHitsRequest(ctx context.Context, w http.ResponseWriter, r *http.Requ
 	}
 
 	// Prepare the query for hits count.
-	q.Optimize()
 	q.DropAllPipes()
 	q.AddCountByTimePipe(int64(step), int64(offset), fields)
 
@@ -204,7 +203,6 @@ func ProcessFieldNamesRequest(ctx context.Context, w http.ResponseWriter, r *htt
 	}
 
 	// Obtain field names for the given query
-	q.Optimize()
 	fieldNames, err := vlstorage.GetFieldNames(ctx, tenantIDs, q)
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot obtain field names: %s", err)
@@ -244,7 +242,6 @@ func ProcessFieldValuesRequest(ctx context.Context, w http.ResponseWriter, r *ht
 	}
 
 	// Obtain unique values for the given field
-	q.Optimize()
 	values, err := vlstorage.GetFieldValues(ctx, tenantIDs, q, fieldName, uint64(limit))
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot obtain values for field %q: %s", fieldName, err)
@@ -267,7 +264,6 @@ func ProcessStreamFieldNamesRequest(ctx context.Context, w http.ResponseWriter,
 	}
 
 	// Obtain stream field names for the given query
-	q.Optimize()
 	names, err := vlstorage.GetStreamFieldNames(ctx, tenantIDs, q)
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot obtain stream field names: %s", err)
@@ -306,7 +302,6 @@ func ProcessStreamFieldValuesRequest(ctx context.Context, w http.ResponseWriter,
 	}
 
 	// Obtain stream field values for the given query and the given fieldName
-	q.Optimize()
 	values, err := vlstorage.GetStreamFieldValues(ctx, tenantIDs, q, fieldName, uint64(limit))
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot obtain stream field values: %s", err)
@@ -338,7 +333,6 @@ func ProcessStreamIDsRequest(ctx context.Context, w http.ResponseWriter, r *http
 	}
 
 	// Obtain streamIDs for the given query
-	q.Optimize()
 	streamIDs, err := vlstorage.GetStreamIDs(ctx, tenantIDs, q, uint64(limit))
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot obtain stream_ids: %s", err)
@@ -370,7 +364,6 @@ func ProcessStreamsRequest(ctx context.Context, w http.ResponseWriter, r *http.R
 	}
 
 	// Obtain streams for the given query
-	q.Optimize()
 	streams, err := vlstorage.GetStreams(ctx, tenantIDs, q, uint64(limit))
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot obtain streams: %s", err)
@@ -398,7 +391,6 @@ func ProcessLiveTailRequest(ctx context.Context, w http.ResponseWriter, r *http.
 			"see https://docs.victoriametrics.com/victorialogs/querying/#live-tailing for details", q)
 		return
 	}
-	q.Optimize()
 
 	refreshIntervalMsecs, err := httputils.GetDuration(r, "refresh_interval", 1000)
 	if err != nil {
@@ -414,13 +406,20 @@ func ProcessLiveTailRequest(ctx context.Context, w http.ResponseWriter, r *http.
 	}
 	startOffset := startOffsetMsecs * 1e6
 
+	offsetMsecs, err := httputils.GetDuration(r, "offset", 1000)
+	if err != nil {
+		httpserver.Errorf(w, r, "%s", err)
+		return
+	}
+	offset := offsetMsecs * 1e6
+
 	ctxWithCancel, cancel := context.WithCancel(ctx)
 	tp := newTailProcessor(cancel)
 
 	ticker := time.NewTicker(refreshInterval)
 	defer ticker.Stop()
 
-	end := time.Now().UnixNano()
+	end := time.Now().UnixNano() - offset
 	start := end - startOffset
 	doneCh := ctxWithCancel.Done()
 	flusher, ok := w.(http.Flusher)
@@ -429,12 +428,7 @@ func ProcessLiveTailRequest(ctx context.Context, w http.ResponseWriter, r *http.
 	}
 	qOrig := q
 	for {
-		q = qOrig.Clone(end)
-		q.AddTimeFilter(start, end)
-		// q.Optimize() call is needed for converting '*' into filterNoop.
-		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6785#issuecomment-2358547733
-		q.Optimize()
-
+		q = qOrig.CloneWithTimeFilter(end, start, end)
 		if err := vlstorage.RunQuery(ctxWithCancel, tenantIDs, q, tp.writeBlock); err != nil {
 			httpserver.Errorf(w, r, "cannot execute tail query [%s]: %s", q, err)
 			return
@@ -454,7 +448,7 @@ func ProcessLiveTailRequest(ctx context.Context, w http.ResponseWriter, r *http.
 			return
 		case <-ticker.C:
 			start = end - tailOffsetNsecs
-			end = time.Now().UnixNano()
+			end = time.Now().UnixNano() - offset
 		}
 	}
 }
@@ -613,8 +607,6 @@ func ProcessStatsQueryRangeRequest(ctx context.Context, w http.ResponseWriter, r
 		return
 	}
 
-	q.Optimize()
-
 	m := make(map[string]*statsSeries)
 	var mLock sync.Mutex
 
@@ -725,8 +717,6 @@ func ProcessStatsQueryRequest(ctx context.Context, w http.ResponseWriter, r *htt
 		return
 	}
 
-	q.Optimize()
-
 	var rows []statsRow
 	var rowsLock sync.Mutex
 
@@ -826,7 +816,6 @@ func ProcessQueryRequest(ctx context.Context, w http.ResponseWriter, r *http.Req
 
 		q.AddPipeLimit(uint64(limit))
 	}
-	q.Optimize()
 
 	writeBlock := func(_ uint, timestamps []int64, columns []logstorage.BlockColumn) {
 		if len(columns) == 0 || len(columns[0].Values) == 0 {
@@ -857,7 +846,6 @@ type row struct {
 func getLastNQueryResults(ctx context.Context, tenantIDs []logstorage.TenantID, q *logstorage.Query, limit int) ([]row, error) {
 	limitUpper := 2 * limit
 	q.AddPipeLimit(uint64(limitUpper))
-	q.Optimize()
 
 	rows, err := getQueryResultsWithLimit(ctx, tenantIDs, q, limitUpper)
 	if err != nil {
@@ -877,11 +865,7 @@ func getLastNQueryResults(ctx context.Context, tenantIDs []logstorage.TenantID,
 	qOrig := q
 	for {
 		timestamp := qOrig.GetTimestamp()
-		q = qOrig.Clone(timestamp)
-		q.AddTimeFilter(start, end)
-		// q.Optimize() call is needed for converting '*' into filterNoop.
-		// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6785#issuecomment-2358547733
-		q.Optimize()
+		q = qOrig.CloneWithTimeFilter(timestamp, start, end)
 		rows, err := getQueryResultsWithLimit(ctx, tenantIDs, q, limitUpper)
 		if err != nil {
 			return nil, err
@@ -985,14 +969,29 @@ func parseCommonArgs(r *http.Request) (*logstorage.Query, []logstorage.TenantID,
 	}
 	tenantIDs := []logstorage.TenantID{tenantID}
 
+	// Parse optional start and end args
+	start, okStart, err := getTimeNsec(r, "start")
+	if err != nil {
+		return nil, nil, err
+	}
+	end, okEnd, err := getTimeNsec(r, "end")
+	if err != nil {
+		return nil, nil, err
+	}
+
 	// Parse optional time arg
 	timestamp, okTime, err := getTimeNsec(r, "time")
 	if err != nil {
 		return nil, nil, err
 	}
 	if !okTime {
-		// If time arg is missing, then evaluate query at the current timestamp
-		timestamp = time.Now().UnixNano()
+		// If time arg is missing, then evaluate query either at the end timestamp (if it is set)
+		// or at the current timestamp (if end query arg isn't set)
+		if okEnd {
+			timestamp = end
+		} else {
+			timestamp = time.Now().UnixNano()
+		}
 	}
 
 	// decrease timestamp by one nanosecond in order to avoid capturing logs belonging
@@ -1006,16 +1005,8 @@ func parseCommonArgs(r *http.Request) (*logstorage.Query, []logstorage.TenantID,
 		return nil, nil, fmt.Errorf("cannot parse query [%s]: %s", qStr, err)
 	}
 
-	// Parse optional start and end args
-	start, okStart, err := getTimeNsec(r, "start")
-	if err != nil {
-		return nil, nil, err
-	}
-	end, okEnd, err := getTimeNsec(r, "end")
-	if err != nil {
-		return nil, nil, err
-	}
 	if okStart || okEnd {
+		// Add _time:[start, end] filter if start or end args were set.
 		if !okStart {
 			start = math.MinInt64
 		}
diff --git a/app/vmalert/config/types.go b/app/vmalert/config/types.go
index b71325ec3..1cf286fe0 100644
--- a/app/vmalert/config/types.go
+++ b/app/vmalert/config/types.go
@@ -71,7 +71,7 @@ func (t *Type) ValidateExpr(expr string) error {
 			return fmt.Errorf("bad prometheus expr: %q, err: %w", expr, err)
 		}
 	case "vlogs":
-		if _, err := logstorage.ParseStatsQuery(expr); err != nil {
+		if _, err := logstorage.ParseStatsQuery(expr, 0); err != nil {
 			return fmt.Errorf("bad LogsQL expr: %q, err: %w", expr, err)
 		}
 	default:
diff --git a/app/vmalert/rule/recording.go b/app/vmalert/rule/recording.go
index 3de0524b5..7b2a7635e 100644
--- a/app/vmalert/rule/recording.go
+++ b/app/vmalert/rule/recording.go
@@ -10,6 +10,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/datasource"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/utils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logstorage"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
 )
@@ -221,6 +222,9 @@ func setIntervalAsTimeFilter(dType, expr string) bool {
 	if dType != "vlogs" {
 		return false
 	}
-	q, _ := logstorage.ParseStatsQuery(expr)
-	return !q.ContainAnyTimeFilter()
+	q, err := logstorage.ParseStatsQuery(expr, 0)
+	if err != nil {
+		logger.Panicf("BUG: the LogsQL query must be valid here; got error: %s; query=[%s]", err, expr)
+	}
+	return !q.HasGlobalTimeFilter()
 }
diff --git a/app/vmauth/auth_config.go b/app/vmauth/auth_config.go
index 237bd18e6..018dce6ba 100644
--- a/app/vmauth/auth_config.go
+++ b/app/vmauth/auth_config.go
@@ -462,13 +462,17 @@ func getLeastLoadedBackendURL(bus []*backendURL, atomicCounter *atomic.Uint32) *
 
 	// Slow path - select other backend urls.
 	n := atomicCounter.Add(1) - 1
-
+	buMin := bus[n%uint32(len(bus))]
 	for i := uint32(0); i < uint32(len(bus)); i++ {
 		idx := (n + i) % uint32(len(bus))
 		bu := bus[idx]
 		if bu.isBroken() {
 			continue
 		}
+		if buMin.isBroken() {
+			// verify that buMin isn't set as broken
+			buMin = bu
+		}
 		if bu.concurrentRequests.Load() == 0 {
 			// Fast path - return the backend with zero concurrently executed requests.
 			// Do not use CompareAndSwap() instead of Load(), since it is much slower on systems with many CPU cores.
@@ -478,7 +482,6 @@ func getLeastLoadedBackendURL(bus []*backendURL, atomicCounter *atomic.Uint32) *
 	}
 
 	// Slow path - return the backend with the minimum number of concurrently executed requests.
-	buMin := bus[n%uint32(len(bus))]
 	minRequests := buMin.concurrentRequests.Load()
 	for _, bu := range bus {
 		if bu.isBroken() {
diff --git a/docs/CaseStudies.md b/docs/CaseStudies.md
index 92fe1c4ba..ea212361f 100644
--- a/docs/CaseStudies.md
+++ b/docs/CaseStudies.md
@@ -461,20 +461,20 @@ VictoriaMetrics has allowed us to extend data retention for our metrics effortle
 
 Across our production VictoriaMetrics clusters, in a 12 months period we go beyond the following figures.
 
-- Active time series: 10M
-- Ingestion rate: 300k samples per second
-- Total number of datapoints: 4400G
-- Data size on disk: 3600 GiB
-- Available memory: 320 GiB as seen by kubernetes (160 GiB physical memory for the hosts)
-- CPU: 20 cores (AMD EPYC 7763), about 70 % idle
+- Active time series: 20M
+- Ingestion rate: 1M samples per second (2M before stream aggregation)
+- Total number of datapoints: 15T+
+- Data size on disk: 12+ TiB
+- Available memory: 1.2 TiB as seen by kubernetes (640 GiB physical memory for the hosts)
+- CPU: 80 cores (AMD EPYC 7763), about 50 % idle
 - Retention period: ~1 year
-- Churn rate: 6M new time series per day (monthly average)
+- Churn rate: 16M new time series per day (monthly average)
 - Query rate:
-  - `/api/v1/query_range`: 10 queries per second
+  - `/api/v1/query_range`: 15 queries per second
   - `/api/v1/query`: 10 queries per second
 - Query duration for `/api/v1/query_range` (weekly mean):
-  - 99th percentile: 700 ms
-  - median: 10 ms
+  - 99th percentile: 1.5 s
+  - median: 15 ms
 
 ## Roblox
 
diff --git a/docs/Cluster-VictoriaMetrics.md b/docs/Cluster-VictoriaMetrics.md
index 2ff400a52..8e460a2e4 100644
--- a/docs/Cluster-VictoriaMetrics.md
+++ b/docs/Cluster-VictoriaMetrics.md
@@ -114,10 +114,9 @@ such as [Graphite](https://docs.victoriametrics.com/#how-to-send-data-from-graph
 
 **Reads**
 
-_Available from [v1.104.0](https://docs.victoriametrics.com/changelog/#v11040)._
 _For better performance prefer specifying [tenants in read URL](https://docs.victoriametrics.com/cluster-victoriametrics/#url-format)._
 
-`vmselect` can execute queries over multiple [tenants](#multitenancy) via special `multitenant` endpoints `http://vmselect:8481/select/multitenant/<suffix>`.
+`vmselect` can execute queries over multiple [tenants](#multitenancy) {{% available_from "v1.104.0" %}} via special `multitenant` endpoints `http://vmselect:8481/select/multitenant/<suffix>`.
 Currently supported endpoints for `<suffix>` are:
 - `/prometheus/api/v1/query`
 - `/prometheus/api/v1/query_range`
diff --git a/docs/Makefile b/docs/Makefile
index ece1d1b51..990b44cb1 100644
--- a/docs/Makefile
+++ b/docs/Makefile
@@ -21,6 +21,18 @@ docs-debug: docs docs-image
 		-p 1313:1313 \
 		-v ./docs:/opt/docs/content vmdocs-docker-package
 
+docs-update-version: docs-image
+	$(if $(filter v%,$(PKG_TAG)), \
+		docker run \
+			--rm \
+			--entrypoint /usr/bin/find \
+			--name vmdocs-docker-container \
+			-v ./docs:/opt/docs/content vmdocs-docker-package \
+				content \
+					-regex ".*\.md" \
+					-exec sed -i 's/{{% available_from "#" %}}/{{% available_from "$(PKG_TAG)" %}}/g' {} \;, \
+		$(info "Skipping docs version update, invalid $$PKG_TAG: $(PKG_TAG)"))
+
 docs-images-to-webp: docs-image
 	docker run \
 		--rm \
diff --git a/docs/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md
index 3e0cda7ca..115e28cae 100644
--- a/docs/VictoriaLogs/CHANGELOG.md
+++ b/docs/VictoriaLogs/CHANGELOG.md
@@ -15,6 +15,24 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
 
 ## tip
 
+## [v0.42.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.42.0-victorialogs)
+
+Released at 2024-11-08
+
+* FEATURE: [`join` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#join-pipe): add an ability to add prefix to all the log field names from the joined query, by using `| join by (<by_fields>) (<query>) prefix "some_prefix"` syntax.
+* FEATURE: [`_time` filter](https://docs.victoriametrics.com/victorialogs/logsql/#time-filter): allow specifying offset without time range. For example, `_time:offset 1d` matches all the logs until `now-1d` in the [`_time` field](https://docs.victoriametrics.com/victorialogs/keyconcepts/#time-field). This is useful when building graphs for time ranges with some offset in the past.
+* FEATURE: [`/select/logsql/tail` HTTP endpoint](): support for `offset` query arg, which can be used for delayed emission of matching logs during live tailing. Thanks to @Fusl for the initial idea and implementation in [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7428).
+* FEATURE: [vlogscli](https://docs.victoriametrics.com/victorialogs/querying/vlogscli/): allow enabling and disabling wrapping of long lines, which do not fit screen width, with `\wrap_long_lines` command.
+* FEATURE: [syslog data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/): allow overriding default [stream fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) with the given [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) during data ingestion. See [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#stream-fields). See [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7480).
+* FEATURE: [syslog data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/): allow adding arbitrary [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) via `[label1=value1 ... labelN=valueN]` syntax inside Syslog messages. For example, `<165>1 2024-06-03T17:42:00.000Z example.com appname 12345 ID47 [field1=value1 field2=value2] some message`.
+* FEATURE: [syslog data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/): allow dropping the specified [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) during data ingestion. See [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#dropping-fields).
+* FEATURE: [syslog data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/): allow adding the specified [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) during data ingestion. See [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#adding-extra-fields). See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7354).
+* FEATURE: [Loki data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/#loki-json-api): show the original request body on parse errors. This should simplify debugging. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7490).
+
+* BUGFIX: [`values` stats function](https://docs.victoriametrics.com/victorialogs/logsql/#values-stats): fix a bug, which could lead to corrupted results. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7458).
+* BUGFIX: [`uniq_values`](https://docs.victoriametrics.com/victorialogs/logsql/#uniq_values-stats), [`min`](https://docs.victoriametrics.com/victorialogs/logsql/#min-stats), [`max`](https://docs.victoriametrics.com/victorialogs/logsql/#max-stats), [`row_min`](https://docs.victoriametrics.com/victorialogs/logsql/#row_min-stats) and [`row_max`](https://docs.victoriametrics.com/victorialogs/logsql/#row_max-stats) stats functions: fix a bug, which could return non-matching field values for these functions. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7458).
+* BUGFIX: [HTTP querying APIs](https://docs.victoriametrics.com/victorialogs/querying/#http-api): properly take into account the `end` query arg when calculating time range for [`_time:duration` filter](https://docs.victoriametrics.com/victorialogs/logsql/#time-filter). Previously the `_time:duration` filter was treated as `_time:[now-duration, now)`, while it should be treated as `_time:[end-duration, end)`.
+
 ## [v0.41.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.41.0-victorialogs)
 
 Released at 2024-11-06
diff --git a/docs/VictoriaLogs/FAQ.md b/docs/VictoriaLogs/FAQ.md
index fc1024ed5..2bc81bb2c 100644
--- a/docs/VictoriaLogs/FAQ.md
+++ b/docs/VictoriaLogs/FAQ.md
@@ -131,3 +131,42 @@ Just send the query with the needed [filters](https://docs.victoriametrics.com/v
 to [`/select/logsql/query`](https://docs.victoriametrics.com/victorialogs/querying/#querying-logs) - VictoriaLogs will return
 the requested logs as a [stream of JSON lines](https://jsonlines.org/). It is recommended specifying [time filter](https://docs.victoriametrics.com/victorialogs/logsql/#time-filter)
 for limiting the amounts of exported logs.
+
+## I want to ingest logs without message field, is that possible?
+
+Starting from version `v0.30.0`, VictoriaLogs started blocking the ingestion of logs **without a message field**, as it is a requirement of the [VictoriaLogs data model](https://docs.victoriametrics.com/victorialogs/keyconcepts/#message-field). 
+
+However, some logs do not have a message field and only contain other fields, such as logs in [this comment](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7056#issuecomment-2434189718) and [this slack thread](https://victoriametrics.slack.com/archives/C05UNTPAEDN/p1730982146818249). Therefore, starting from version `v0.39.0`, logs without a message field are **allowed to be ingested**, 
+and their message field will be recorded as: 
+```json
+{"_msg": "missing _msg field; see https://docs.victoriametrics.com/victorialogs/keyconcepts/#message-field"}
+```
+
+The default message field value can be changed using the `-defaultMsgValue` flag, for example, `-defaultMsgValue=foo`.
+
+Please note that the message field is **crucial** for VictoriaLogs, so it is important to fill it with meaningful content.
+
+## What if my logs have multiple message fields candidates?
+
+When ingesting with VictoriaLogs, the message fields is specified through `_msg_field` param, which can accept **multiple fields**, and the **first non-empty field** will be used as the message field. 
+Here is an example URL when pushing logs to VictoriaLogs with Promtail:
+```yaml
+clients:
+  - url: http://localhost:9428/insert/loki/api/v1/push?_stream_fields=instance,job,host,app&_msg=message,body
+```
+
+For the following log, its `_msg` will be `foo bar in message`:
+```json
+{
+  "message": "foo bar in message",
+  "body": "foo bar in body"
+}
+```
+
+And for the following log, its `_msg` will be `foo bar in body`:
+```json
+{
+  "message": "",
+  "body": "foo bar in body"
+}
+```
\ No newline at end of file
diff --git a/docs/VictoriaLogs/LogsQL.md b/docs/VictoriaLogs/LogsQL.md
index 9c1f3023e..cd3bbe065 100644
--- a/docs/VictoriaLogs/LogsQL.md
+++ b/docs/VictoriaLogs/LogsQL.md
@@ -316,6 +316,7 @@ For example, `_time:2023-10-20` matches all the logs for `2023-10-20` day accord
 
 It is possible to specify generic offset for the selected time range by appending `offset` after the `_time` filter. Examples:
 
+- `_time:offset 1h` matches logs until `now-1h`.
 - `_time:5m offset 1h` matches logs on the time range `(now-1h5m, now-1h]`.
 - `_time:2023-07Z offset 5h30m` matches logs on July, 2023 by UTC with offset 5h30m.
 - `_time:[2023-02-01Z, 2023-03-01Z) offset 1w` matches logs the week before the time range `[2023-02-01Z, 2023-03-01Z)` by UTC.
@@ -1817,6 +1818,16 @@ _time:1d {app="app1"} | stats by (user) count() app1_hits
   | filter app2_hits:*
 ```
 
+It is possible adding a prefix to all the field names returned by the `<query>` by specifying the needed prefix after the `<query>`.
+For example, the following query adds `app2.` prefix to all `<query>` log fields:
+
+```logsql
+_time:1d {app="app1"} | stats by (user) count() app1_hits
+  | join by (user) (
+    _time:1d {app="app2"} | stats by (user) count() app2_hits
+  ) prefix "app2."
+```
+
 **Performance tips**:
 
 - Make sure that the `<query>` in the `join` pipe returns relatively small number of results, since they are kept in RAM during execution of `join` pipe.
diff --git a/docs/VictoriaLogs/README.md b/docs/VictoriaLogs/README.md
index a48cd2766..52ae67272 100644
--- a/docs/VictoriaLogs/README.md
+++ b/docs/VictoriaLogs/README.md
@@ -244,6 +244,13 @@ VictoriaLogs has very low overhead for per-tenant management, so it is OK to hav
 
 VictoriaLogs doesn't perform per-tenant authorization. Use [vmauth](https://docs.victoriametrics.com/vmauth/) or similar tools for per-tenant authorization.
 
+## Security
+
+It is expected that VictoriaLogs runs in a protected environment, which is unreachable from the Internet without proper authorization.
+It is recommended providing access to VictoriaLogs [data ingestion APIs](https://docs.victoriametrics.com/victorialogs/data-ingestion/)
+and [querying APIs](https://docs.victoriametrics.com/victorialogs/querying/#http-api) via [vmauth](https://docs.victoriametrics.com/vmauth/)
+or similar authorization proxies.
+
 ## Benchmarks
 
 Here is a [benchmark suite](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/logs-benchmark) for comparing data ingestion performance
@@ -423,6 +430,22 @@ Pass `-help` to VictoriaLogs in order to see the list of supported command-line
     	Compression method for syslog messages received at the corresponding -syslog.listenAddr.udp. Supported values: none, gzip, deflate. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#compression
     	Supports an array of values separated by comma or specified via multiple flags.
     	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -syslog.extraFields.tcp array
+    	Fields to add to logs ingested via the corresponding -syslog.listenAddr.tcp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#adding-extra-fields
+    	Supports an array of values separated by comma or specified via multiple flags.
+    	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -syslog.extraFields.udp array
+    	Fields to add to logs ingested via the corresponding -syslog.listenAddr.udp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#adding-extra-fields
+    	Supports an array of values separated by comma or specified via multiple flags.
+    	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -syslog.ignoreFields.tcp array
+    	Fields to ignore at logs ingested via the corresponding -syslog.listenAddr.tcp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#dropping-fields
+    	Supports an array of values separated by comma or specified via multiple flags.
+    	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -syslog.ignoreFields.udp array
+    	Fields to ignore at logs ingested via the corresponding -syslog.listenAddr.udp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#dropping-fields
+    	Supports an array of values separated by comma or specified via multiple flags.
+    	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
   -syslog.listenAddr.tcp array
     	Comma-separated list of TCP addresses to listen to for Syslog messages. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/
     	Supports an array of values separated by comma or specified via multiple flags.
@@ -431,12 +454,20 @@ Pass `-help` to VictoriaLogs in order to see the list of supported command-line
     	Comma-separated list of UDP address to listen to for Syslog messages. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/
     	Supports an array of values separated by comma or specified via multiple flags.
     	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -syslog.streamFields.tcp array
+    	Fields to use as log stream labels for logs ingested via the corresponding -syslog.listenAddr.tcp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#stream-fields
+    	Supports an array of values separated by comma or specified via multiple flags.
+    	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
+  -syslog.streamFields.udp array
+    	Fields to use as log stream labels for logs ingested via the corresponding -syslog.listenAddr.udp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#stream-fields
+    	Supports an array of values separated by comma or specified via multiple flags.
+    	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
   -syslog.tenantID.tcp array
-    	TenantID for logs ingested via the corresponding -syslog.listenAddr.tcp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/
+    	TenantID for logs ingested via the corresponding -syslog.listenAddr.tcp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#multitenancy
     	Supports an array of values separated by comma or specified via multiple flags.
     	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
   -syslog.tenantID.udp array
-    	TenantID for logs ingested via the corresponding -syslog.listenAddr.udp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/
+    	TenantID for logs ingested via the corresponding -syslog.listenAddr.udp. See https://docs.victoriametrics.com/victorialogs/data-ingestion/syslog/#multitenancy
     	Supports an array of values separated by comma or specified via multiple flags.
     	Value can contain comma inside single-quoted or double-quoted string, {}, [] and () braces.
   -syslog.timezone string
diff --git a/docs/VictoriaLogs/data-ingestion/DataDogAgent.md b/docs/VictoriaLogs/data-ingestion/DataDogAgent.md
index 6002ac55e..ccbe5322f 100644
--- a/docs/VictoriaLogs/data-ingestion/DataDogAgent.md
+++ b/docs/VictoriaLogs/data-ingestion/DataDogAgent.md
@@ -49,4 +49,4 @@ See also:
 
 - [Data ingestion troubleshooting](https://docs.victoriametrics.com/victorialogs/data-ingestion/#troubleshooting).
 - [How to query VictoriaLogs](https://docs.victoriametrics.com/victorialogs/querying/).
-- [Docker-compose demo for Datadog integration with VictoriaLogs](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker/victorialogs/datadog).
+- [Docker-compose demo for Datadog integration with VictoriaLogs](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker/victorialogs/datadog-agent).
diff --git a/docs/VictoriaLogs/data-ingestion/README.md b/docs/VictoriaLogs/data-ingestion/README.md
index 4de3f3463..f92c6812e 100644
--- a/docs/VictoriaLogs/data-ingestion/README.md
+++ b/docs/VictoriaLogs/data-ingestion/README.md
@@ -10,6 +10,7 @@
 - Telegraf - see [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/telegraf/).
 - OpenTelemetry Collector - see [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/opentelemetry/).
 - Journald - see [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/).
+- DataDog - see [these docs](https://docs.victoriametrics.com/victorialogs/data-ingestion/datadog-agent/).
 
 The ingested logs can be queried according to [these docs](https://docs.victoriametrics.com/victorialogs/querying/).
 
@@ -322,5 +323,5 @@ Here is the list of log collectors and their ingestion formats supported by Vict
 | [Telegraf](https://docs.victoriametrics.com/victorialogs/data-ingestion/telegraf/) | [Yes](https://github.com/influxdata/telegraf/tree/master/plugins/outputs/elasticsearch) | [Yes](https://github.com/influxdata/telegraf/tree/master/plugins/outputs/http) | [Yes](https://github.com/influxdata/telegraf/tree/master/plugins/outputs/loki) | [Yes](https://github.com/influxdata/telegraf/blob/master/plugins/outputs/syslog) | Yes | No | No |
 | [Fluentd](https://docs.victoriametrics.com/victorialogs/data-ingestion/fluentd/) | [Yes](https://github.com/uken/fluent-plugin-elasticsearch) | [Yes](https://docs.fluentd.org/output/http) | [Yes](https://grafana.com/docs/loki/latest/send-data/fluentd/) | [Yes](https://github.com/fluent-plugins-nursery/fluent-plugin-remote_syslog) | No | No | No |
 | [Journald](https://docs.victoriametrics.com/victorialogs/data-ingestion/journald/) | No | No | No | No | No | Yes | No |
-| [DataDog Agent](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/DataDogAgent.html) | No | No | No | No | No | No | Yes |
+| [DataDog Agent](https://docs.victoriametrics.com/VictoriaLogs/data-ingestion/datadog-agent) | No | No | No | No | No | No | Yes |
 
diff --git a/docs/VictoriaLogs/data-ingestion/syslog.md b/docs/VictoriaLogs/data-ingestion/syslog.md
index 873e98ce2..6c964570b 100644
--- a/docs/VictoriaLogs/data-ingestion/syslog.md
+++ b/docs/VictoriaLogs/data-ingestion/syslog.md
@@ -41,8 +41,8 @@ from the received Syslog lines:
 
 - [`_time`](https://docs.victoriametrics.com/victorialogs/keyconcepts/#time-field) - log timestamp. See also [log timestamps](#log-timestamps)
 - [`_msg`](https://docs.victoriametrics.com/victorialogs/keyconcepts/#message-field) - the `MESSAGE` field from the supported syslog formats above
-- `hostname`, `app_name` and `proc_id` - [stream fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) for unique identification
-  over every log stream
+- `hostname`, `app_name` and `proc_id` - for unique identification of [log streams](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields).
+  It is possible to change the list of fields for log streams - see [these docs](#stream-fields).
 - `priority`, `facility` and `severity` - these fields are extracted from `<PRI>` field
 - `format` - this field is set to either `rfc3164` or `rfc5424` depending on the format of the parsed syslog line
 - `msg_id` - `MSGID` field from log line in `RFC5424` format.
@@ -71,6 +71,9 @@ See also:
 - [Security](#security)
 - [Compression](#compression)
 - [Multitenancy](#multitenancy)
+- [Stream fields](#stream-fields)
+- [Dropping fields](#dropping-fields)
+- [Adding extra fields](#adding-extra-fields)
 - [Data ingestion troubleshooting](https://docs.victoriametrics.com/victorialogs/data-ingestion/#troubleshooting).
 - [How to query VictoriaLogs](https://docs.victoriametrics.com/victorialogs/querying/).
 
@@ -132,6 +135,40 @@ For example, the following command starts VictoriaLogs, which writes syslog mess
 ./victoria-logs -syslog.listenAddr.tcp=:514 -syslog.tenantID.tcp=12:34
 ```
 
+## Stream fields
+
+VictoriaLogs uses `(hostname, app_name, proc_id)` fields as labels for [log streams](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) by default.
+It is possible setting other set of labels via `-syslog.streamFields.tcp` and `-syslog.streamFields.udp` command-line flags
+for logs insted via the corresponding `-syslog.listenAddr.tcp` and `-syslog.listenAddr.dup` addresses.
+For example, the following command starts VictoriaLogs, which uses `(hostname, app_name)` fields as log stream labels
+for logs received at TCP port 514:
+
+```sh
+./victoria-logs -syslog.listenAddr.tcp=:514 -syslog.streamFields.tcp='["hostname","app_name"]'
+```
+
+## Dropping fields
+
+VictoriaLogs supports `-syslog.ignoreFields.tcp` and `-syslog.ignoreFields.udp` command-line flags for skipping
+the given [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) during inestion
+of Syslog logs into `-syslog.listenAddr.tcp` and `-syslog.listenAddr.udp` addresses.
+For example, the following command starts VictoriaLogs, which drops `proc_id` and `msg_id` fields from logs received at TCP port 514:
+
+```sh
+./victoria-logs -syslog.listenAddr.tcp=:514 -syslog.ignoreFields.tcp='["prod_id","msg_id"]'
+```
+
+## Adding extra fields
+
+VictoriaLogs supports -`syslog.extraFields.tcp` and `-syslog.extraFields.udp` command-line flags for adding
+the given [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) during data ingestion
+of Syslog logs into `-syslog.listenAddr.tcp` and `-syslog.listenAddr.udp` addresses.
+For example, the following command starts VictoriaLogs, which adds `source=foo` and `abc=def` fields to logs received at TCP port 514:
+
+```sh
+./victoria-logs -syslog.listenAddr.tcp=:514 -syslog.extraFields.tcp='{"source":"foo","abc":"def"}'
+```
+
 ## Multiple configs
 
 VictoriaLogs can accept syslog messages via multiple TCP and UDP ports with individual configurations for [log timestamps](#log-timestamps), [compression](#compression), [security](#security)
diff --git a/docs/VictoriaLogs/querying/README.md b/docs/VictoriaLogs/querying/README.md
index 743064c55..2a023c0a2 100644
--- a/docs/VictoriaLogs/querying/README.md
+++ b/docs/VictoriaLogs/querying/README.md
@@ -148,13 +148,29 @@ The `<query>` must conform the following rules:
   across multiple streams.
 
 Live tailing supports returning historical logs, which were ingested into VictoriaLogs before the start of live tailing. Pass `start_offset=<d>` query
-arg to `/select/logsql/tail` where `<d>` is the duration for returning historical logs. For example, the following request returns historical logs
+arg to `/select/logsql/tail` where `<d>` is the duration for returning historical logs. For example, the following command returns historical logs
 which were ingested into VictoriaLogs during the last hour, before starting live tailing:
 
 ```sh
 curl -N http://localhost:9428/select/logsql/tail -d 'query=*' -d 'start_offset=1h'
 ```
 
+Live tailing delays delivering new logs for one second, so they could be properly delivered from log collectors to VictoriaLogs.
+This delay can be changed via `offset` query arg. For example, the following command delays delivering new logs for 30 seconds:
+
+```sh
+curl -N http://localhost:9428/select/logsql/tail -d 'query=*' -d 'offset=30s'
+```
+
+Live tailing checks for new logs every second. The frequency for the check can be changed via `refresh_interval` query arg.
+For example, the following command instructs live tailing to check for new logs every 10 seconds:
+
+```sh
+curl -N http://localhost:9428/select/logsql/tail -d 'query=*' -d 'refresh_interval=10s'
+```
+
+It isn't recommended setting too low value for `refresh_interval` query arg, since this may increase load on VictoriaLogs without measurable benefits.
+
 **Performance tip**: live tailing works the best if it matches newly ingested logs at relatively slow rate (e.g. up to 1K matching logs per second),
 e.g. it is optimized for the case when real humans inspect the output of live tailing in the real time. If live tailing returns logs at too high rate,
 then it is recommended adding more specific [filters](https://docs.victoriametrics.com/victorialogs/logsql/#filters) to the `<query>`, so it matches less logs.
diff --git a/docs/changelog/CHANGELOG.md b/docs/changelog/CHANGELOG.md
index 218b6c1ed..76f19c5fe 100644
--- a/docs/changelog/CHANGELOG.md
+++ b/docs/changelog/CHANGELOG.md
@@ -21,6 +21,7 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 * BUGFIX: [vmctl](https://docs.victoriametrics.com/vmctl/): drop rows that do not belong to the current series during import. The dropped rows should belong to another series whose tags are a superset of the current series. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7301) and [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7330). Thanks to @dpedu for reporting and cooperating with the test.
 * BUGFIX: [vmsingle](https://docs.victoriametrics.com/single-server-victoriametrics/), `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): keep the order of resulting time series when `limit_offset` is applied. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7068).
 * BUGFIX: [graphite](https://docs.victoriametrics.com/#graphite-render-api-usage): properly handle xFilesFactor=0 for `transformRemoveEmptySeries` function. See [this PR](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/7337) for details.
+* BUGFIX: [vmauth](https://docs.victoriametrics.com/vmauth): properly check availability of all the backends before giving up when proxying requests. Previously, vmauth could return an error even if there were healthy backends available. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3061) for details.
 
 ## [v1.106.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.106.0)
 
diff --git a/docs/url-examples.md b/docs/url-examples.md
index 6c85ef78d..fa20036a4 100644
--- a/docs/url-examples.md
+++ b/docs/url-examples.md
@@ -661,7 +661,7 @@ curl -H 'Content-Type: application/json' -d '[{"metric":"foo","value":45.34},{"m
 Cluster version of VictoriaMetrics:
 
 ```sh
-curl -H 'Content-Type: application/json' -d '[{"metric":"foo","value":45.34},{"metric":"bar","value":43}]' http://<vminsert>:8480/insert/42/opentsdb/api/put
+curl -H 'Content-Type: application/json' -d '[{"metric":"foo","value":45.34},{"metric":"bar","value":43}]' http://<vminsert>:4242/insert/42/opentsdb/api/put
 ```
 
 
diff --git a/lib/logstorage/block_result.go b/lib/logstorage/block_result.go
index 76459953a..b62f4a985 100644
--- a/lib/logstorage/block_result.go
+++ b/lib/logstorage/block_result.go
@@ -1726,7 +1726,9 @@ func (c *blockResultColumn) getValuesEncoded(br *blockResult) []string {
 	return c.valuesEncoded
 }
 
-// forEachDictValue calls f for every value in the column dictionary.
+// forEachDictValue calls f for every matching value in the column dictionary.
+//
+// It properly skips non-matching dict values.
 func (c *blockResultColumn) forEachDictValue(br *blockResult, f func(v string)) {
 	if c.valueType != valueTypeDict {
 		logger.Panicf("BUG: unexpected column valueType=%d; want %d", c.valueType, valueTypeDict)
@@ -1756,7 +1758,9 @@ func (c *blockResultColumn) forEachDictValue(br *blockResult, f func(v string))
 	encoding.PutUint64s(a)
 }
 
-// forEachDictValueWithHits calls f for every value in the column dictionary.
+// forEachDictValueWithHits calls f for every matching value in the column dictionary.
+//
+// It properly skips non-matching dict values.
 //
 // hits is the number of rows with the given value v in the column.
 func (c *blockResultColumn) forEachDictValueWithHits(br *blockResult, f func(v string, hits uint64)) {
diff --git a/lib/logstorage/filter.go b/lib/logstorage/filter.go
index 4b8024bff..19917a3bf 100644
--- a/lib/logstorage/filter.go
+++ b/lib/logstorage/filter.go
@@ -49,6 +49,17 @@ func visitFilters(filters []filter, visitFunc func(f filter) bool) bool {
 //
 // It doesn't copy other filters by returning them as is.
 func copyFilter(f filter, visitFunc func(f filter) bool, copyFunc func(f filter) (filter, error)) (filter, error) {
+	f, err := copyFilterInternal(f, visitFunc, copyFunc)
+	if err != nil {
+		return nil, err
+	}
+	if !visitFunc(f) {
+		return f, nil
+	}
+	return copyFunc(f)
+}
+
+func copyFilterInternal(f filter, visitFunc func(f filter) bool, copyFunc func(f filter) (filter, error)) (filter, error) {
 	switch t := f.(type) {
 	case *filterAnd:
 		filters, err := copyFilters(t.filters, visitFunc, copyFunc)
@@ -78,11 +89,7 @@ func copyFilter(f filter, visitFunc func(f filter) bool, copyFunc func(f filter)
 		}
 		return fn, nil
 	default:
-		if !visitFunc(t) {
-			// Nothing to copy
-			return t, nil
-		}
-		return copyFunc(t)
+		return f, nil
 	}
 }
 
diff --git a/lib/logstorage/if_filter.go b/lib/logstorage/if_filter.go
index f6852e604..9cfb6e4fd 100644
--- a/lib/logstorage/if_filter.go
+++ b/lib/logstorage/if_filter.go
@@ -50,26 +50,3 @@ func parseIfFilter(lex *lexer) (*ifFilter, error) {
 
 	return iff, nil
 }
-
-func (iff *ifFilter) optimizeFilterIn() {
-	if iff == nil {
-		return
-	}
-
-	optimizeFilterIn(iff.f)
-}
-
-func optimizeFilterIn(f filter) {
-	if f == nil {
-		return
-	}
-
-	visitFunc := func(f filter) bool {
-		fi, ok := f.(*filterIn)
-		if ok && fi.q != nil {
-			fi.q.Optimize()
-		}
-		return false
-	}
-	_ = visitFilter(f, visitFunc)
-}
diff --git a/lib/logstorage/parser.go b/lib/logstorage/parser.go
index d770581a1..2f047c167 100644
--- a/lib/logstorage/parser.go
+++ b/lib/logstorage/parser.go
@@ -335,7 +335,7 @@ func (q *Query) AddCountByTimePipe(step, off int64, fields []string) {
 	}
 }
 
-// Clone returns a copy of q.
+// Clone returns a copy of q at the given timestamp.
 func (q *Query) Clone(timestamp int64) *Query {
 	qStr := q.String()
 	qCopy, err := ParseQueryAtTimestamp(qStr, timestamp)
@@ -345,6 +345,13 @@ func (q *Query) Clone(timestamp int64) *Query {
 	return qCopy
 }
 
+// CloneWithTimeFilter clones q at the given timestamp and adds _time:[start, end] filter to the cloned q.
+func (q *Query) CloneWithTimeFilter(timestamp, start, end int64) *Query {
+	q = q.Clone(timestamp)
+	q.AddTimeFilter(start, end)
+	return q
+}
+
 // CanReturnLastNResults returns true if time range filter at q can be adjusted for returning the last N results.
 func (q *Query) CanReturnLastNResults() bool {
 	for _, p := range q.pipes {
@@ -524,8 +531,8 @@ func (q *Query) AddPipeLimit(n uint64) {
 	})
 }
 
-// Optimize tries optimizing the query.
-func (q *Query) Optimize() {
+// optimize applies various optimations to q.
+func (q *Query) optimize() {
 	q.pipes = optimizeSortOffsetPipes(q.pipes)
 	q.pipes = optimizeSortLimitPipes(q.pipes)
 	q.pipes = optimizeUniqLimitPipes(q.pipes)
@@ -548,16 +555,14 @@ func (q *Query) Optimize() {
 		}
 	}
 
-	// Substitute '*' prefixFilter with filterNoop in order to avoid reading _msg data.
-	q.f = removeStarFilters(q.f)
+	// flatten nested AND filters
+	q.f = flattenFiltersAnd(q.f)
 
-	// Call Optimize for queries from 'in(query)' filters.
-	optimizeFilterIn(q.f)
+	// flatten nested OR filters
+	q.f = flattenFiltersOr(q.f)
 
-	// Optimize individual pipes.
-	for _, p := range q.pipes {
-		p.optimize()
-	}
+	// Substitute '*' prefixFilter with filterNoop in order to avoid reading _msg data.
+	q.f = removeStarFilters(q.f)
 }
 
 // GetStatsByFields returns `by (...)` fields from the last `stats` pipe at q.
@@ -735,7 +740,80 @@ func addByTimeField(byFields []*byStatsField, step int64) []*byStatsField {
 	return dstFields
 }
 
+func flattenFiltersAnd(f filter) filter {
+	visitFunc := func(f filter) bool {
+		fa, ok := f.(*filterAnd)
+		if !ok {
+			return false
+		}
+		for _, f := range fa.filters {
+			if _, ok := f.(*filterAnd); ok {
+				return true
+			}
+		}
+		return false
+	}
+	copyFunc := func(f filter) (filter, error) {
+		fa := f.(*filterAnd)
+
+		var resultFilters []filter
+		for _, f := range fa.filters {
+			child, ok := f.(*filterAnd)
+			if !ok {
+				resultFilters = append(resultFilters, f)
+				continue
+			}
+			resultFilters = append(resultFilters, child.filters...)
+		}
+		return &filterAnd{
+			filters: resultFilters,
+		}, nil
+	}
+	f, err := copyFilter(f, visitFunc, copyFunc)
+	if err != nil {
+		logger.Fatalf("BUG: unexpected error: %s", err)
+	}
+	return f
+}
+
+func flattenFiltersOr(f filter) filter {
+	visitFunc := func(f filter) bool {
+		fo, ok := f.(*filterOr)
+		if !ok {
+			return false
+		}
+		for _, f := range fo.filters {
+			if _, ok := f.(*filterOr); ok {
+				return true
+			}
+		}
+		return false
+	}
+	copyFunc := func(f filter) (filter, error) {
+		fo := f.(*filterOr)
+
+		var resultFilters []filter
+		for _, f := range fo.filters {
+			child, ok := f.(*filterOr)
+			if !ok {
+				resultFilters = append(resultFilters, f)
+				continue
+			}
+			resultFilters = append(resultFilters, child.filters...)
+		}
+		return &filterOr{
+			filters: resultFilters,
+		}, nil
+	}
+	f, err := copyFilter(f, visitFunc, copyFunc)
+	if err != nil {
+		logger.Fatalf("BUG: unexpected error: %s", err)
+	}
+	return f
+}
+
 func removeStarFilters(f filter) filter {
+	// Substitute `*` filterPrefix with filterNoop
 	visitFunc := func(f filter) bool {
 		fp, ok := f.(*filterPrefix)
 		return ok && isMsgFieldName(fp.fieldName) && fp.prefix == ""
@@ -748,6 +826,43 @@ func removeStarFilters(f filter) filter {
 	if err != nil {
 		logger.Fatalf("BUG: unexpected error: %s", err)
 	}
+
+	// Drop filterNoop inside filterAnd
+	visitFunc = func(f filter) bool {
+		fa, ok := f.(*filterAnd)
+		if !ok {
+			return false
+		}
+		for _, f := range fa.filters {
+			if _, ok := f.(*filterNoop); ok {
+				return true
+			}
+		}
+		return false
+	}
+	copyFunc = func(f filter) (filter, error) {
+		fa := f.(*filterAnd)
+		var resultFilters []filter
+		for _, f := range fa.filters {
+			if _, ok := f.(*filterNoop); !ok {
+				resultFilters = append(resultFilters, f)
+			}
+		}
+		if len(resultFilters) == 0 {
+			return &filterNoop{}, nil
+		}
+		if len(resultFilters) == 1 {
+			return resultFilters[0], nil
+		}
+		return &filterAnd{
+			filters: resultFilters,
+		}, nil
+	}
+	f, err = copyFilter(f, visitFunc, copyFunc)
+	if err != nil {
+		logger.Fatalf("BUG: unexpected error: %s", err)
+	}
+
 	return f
 }
 
@@ -878,9 +993,9 @@ func ParseQuery(s string) (*Query, error) {
 	return ParseQueryAtTimestamp(s, timestamp)
 }
 
-// ParseStatsQuery parses s with needed stats query checks.
-func ParseStatsQuery(s string) (*Query, error) {
-	q, err := ParseQuery(s)
+// ParseStatsQuery parses LogsQL query s at the given timestamp with the needed stats query checks.
+func ParseStatsQuery(s string, timestamp int64) (*Query, error) {
+	q, err := ParseQueryAtTimestamp(s, timestamp)
 	if err != nil {
 		return nil, err
 	}
@@ -890,29 +1005,12 @@ func ParseStatsQuery(s string) (*Query, error) {
 	return q, nil
 }
 
-// ContainAnyTimeFilter returns true when query contains a global time filter.
-func (q *Query) ContainAnyTimeFilter() bool {
-	if hasTimeFilter(q.f) {
-		return true
-	}
-	for _, p := range q.pipes {
-		if pf, ok := p.(*pipeFilter); ok {
-			if hasTimeFilter(pf.f) {
-				return true
-			}
-		}
-	}
-	return false
-}
-
-func hasTimeFilter(f filter) bool {
-	if f == nil {
-		return false
-	}
-	switch t := f.(type) {
+// HasGlobalTimeFilter returns true when query contains a global time filter.
+func (q *Query) HasGlobalTimeFilter() bool {
+	switch t := q.f.(type) {
 	case *filterAnd:
 		for _, subF := range t.filters {
-			if hasTimeFilter(subF) {
+			if _, ok := subF.(*filterTime); ok {
 				return true
 			}
 		}
@@ -936,6 +1034,7 @@ func ParseQueryAtTimestamp(s string, timestamp int64) (*Query, error) {
 		return nil, fmt.Errorf("unexpected unparsed tail after [%s]; context: [%s]; tail: [%s]", q, lex.context(), lex.s)
 	}
 	q.timestamp = timestamp
+	q.optimize()
 	return q, nil
 }
 
@@ -2036,6 +2135,20 @@ func getWeekRangeArg(lex *lexer) (time.Weekday, string, error) {
 }
 
 func parseFilterTimeRange(lex *lexer) (*filterTime, error) {
+	if lex.isKeyword("offset") {
+		ft := &filterTime{
+			minTimestamp: math.MinInt64,
+			maxTimestamp: lex.currentTimestamp,
+		}
+		offset, offsetStr, err := parseTimeOffset(lex)
+		if err != nil {
+			return nil, fmt.Errorf("cannot parse offset for _time filter []: %w", err)
+		}
+		ft.maxTimestamp -= offset
+		ft.stringRepr = offsetStr
+		return ft, nil
+	}
+
 	ft, err := parseFilterTime(lex)
 	if err != nil {
 		return nil, err
@@ -2043,20 +2156,33 @@ func parseFilterTimeRange(lex *lexer) (*filterTime, error) {
 	if !lex.isKeyword("offset") {
 		return ft, nil
 	}
+
+	offset, offsetStr, err := parseTimeOffset(lex)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse offset for _time filter [%s]: %w", ft, err)
+	}
+	ft.minTimestamp -= offset
+	ft.maxTimestamp -= offset
+	ft.stringRepr += " " + offsetStr
+	return ft, nil
+}
+
+func parseTimeOffset(lex *lexer) (int64, string, error) {
+	if !lex.isKeyword("offset") {
+		return 0, "", fmt.Errorf("unexpected token %q; want 'offset'", lex.token)
+	}
 	lex.nextToken()
+
 	s, err := getCompoundToken(lex)
 	if err != nil {
-		return nil, fmt.Errorf("cannot parse offset in _time filter: %w", err)
+		return 0, "", err
 	}
 	d, ok := tryParseDuration(s)
 	if !ok {
-		return nil, fmt.Errorf("cannot parse offset %q for _time filter %s", s, ft)
+		return 0, "", fmt.Errorf("cannot parse duration [%s]", s)
 	}
 	offset := int64(d)
-	ft.minTimestamp -= offset
-	ft.maxTimestamp -= offset
-	ft.stringRepr += " offset " + s
-	return ft, nil
+	return offset, "offset " + s, nil
 }
 
 func parseFilterTime(lex *lexer) (*filterTime, error) {
diff --git a/lib/logstorage/pipe.go b/lib/logstorage/pipe.go
index 00ac68e42..f4b49e910 100644
--- a/lib/logstorage/pipe.go
+++ b/lib/logstorage/pipe.go
@@ -26,9 +26,6 @@ type pipe interface {
 	// The returned pipeProcessor may call cancel() at any time in order to notify the caller to stop sending new data to it.
 	newPipeProcessor(workersCount int, stopCh <-chan struct{}, cancel func(), ppNext pipeProcessor) pipeProcessor
 
-	// optimize must optimize the pipe
-	optimize()
-
 	// hasFilterInWithQuery must return true of pipe contains 'in(subquery)' filter (recursively).
 	hasFilterInWithQuery() bool
 
diff --git a/lib/logstorage/pipe_block_stats.go b/lib/logstorage/pipe_block_stats.go
index 8e6fc0b47..40e74201c 100644
--- a/lib/logstorage/pipe_block_stats.go
+++ b/lib/logstorage/pipe_block_stats.go
@@ -22,10 +22,6 @@ func (ps *pipeBlockStats) canLiveTail() bool {
 	return false
 }
 
-func (ps *pipeBlockStats) optimize() {
-	// nothing to do
-}
-
 func (ps *pipeBlockStats) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_blocks_count.go b/lib/logstorage/pipe_blocks_count.go
index f1381b241..6c109147c 100644
--- a/lib/logstorage/pipe_blocks_count.go
+++ b/lib/logstorage/pipe_blocks_count.go
@@ -31,10 +31,6 @@ func (pc *pipeBlocksCount) updateNeededFields(neededFields, unneededFields field
 	unneededFields.reset()
 }
 
-func (pc *pipeBlocksCount) optimize() {
-	// nothing to do
-}
-
 func (pc *pipeBlocksCount) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_copy.go b/lib/logstorage/pipe_copy.go
index 997023167..2c43920fb 100644
--- a/lib/logstorage/pipe_copy.go
+++ b/lib/logstorage/pipe_copy.go
@@ -54,10 +54,6 @@ func (pc *pipeCopy) updateNeededFields(neededFields, unneededFields fieldsSet) {
 	}
 }
 
-func (pc *pipeCopy) optimize() {
-	// Nothing to do
-}
-
 func (pc *pipeCopy) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_delete.go b/lib/logstorage/pipe_delete.go
index e61fcfcc6..15429d15d 100644
--- a/lib/logstorage/pipe_delete.go
+++ b/lib/logstorage/pipe_delete.go
@@ -34,10 +34,6 @@ func (pd *pipeDelete) updateNeededFields(neededFields, unneededFields fieldsSet)
 	}
 }
 
-func (pd *pipeDelete) optimize() {
-	// nothing to do
-}
-
 func (pd *pipeDelete) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_drop_empty_fields.go b/lib/logstorage/pipe_drop_empty_fields.go
index ba18ad65b..292842b66 100644
--- a/lib/logstorage/pipe_drop_empty_fields.go
+++ b/lib/logstorage/pipe_drop_empty_fields.go
@@ -21,10 +21,6 @@ func (pd *pipeDropEmptyFields) canLiveTail() bool {
 	return true
 }
 
-func (pd *pipeDropEmptyFields) optimize() {
-	// nothing to do
-}
-
 func (pd *pipeDropEmptyFields) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_extract.go b/lib/logstorage/pipe_extract.go
index ec89b15f7..4cf90897c 100644
--- a/lib/logstorage/pipe_extract.go
+++ b/lib/logstorage/pipe_extract.go
@@ -45,10 +45,6 @@ func (pe *pipeExtract) canLiveTail() bool {
 	return true
 }
 
-func (pe *pipeExtract) optimize() {
-	pe.iff.optimizeFilterIn()
-}
-
 func (pe *pipeExtract) hasFilterInWithQuery() bool {
 	return pe.iff.hasFilterInWithQuery()
 }
diff --git a/lib/logstorage/pipe_extract_regexp.go b/lib/logstorage/pipe_extract_regexp.go
index 2f40d36ad..c5abe5dc3 100644
--- a/lib/logstorage/pipe_extract_regexp.go
+++ b/lib/logstorage/pipe_extract_regexp.go
@@ -47,10 +47,6 @@ func (pe *pipeExtractRegexp) canLiveTail() bool {
 	return true
 }
 
-func (pe *pipeExtractRegexp) optimize() {
-	pe.iff.optimizeFilterIn()
-}
-
 func (pe *pipeExtractRegexp) hasFilterInWithQuery() bool {
 	return pe.iff.hasFilterInWithQuery()
 }
diff --git a/lib/logstorage/pipe_field_names.go b/lib/logstorage/pipe_field_names.go
index 283cb7675..c5257e108 100644
--- a/lib/logstorage/pipe_field_names.go
+++ b/lib/logstorage/pipe_field_names.go
@@ -39,10 +39,6 @@ func (pf *pipeFieldNames) updateNeededFields(neededFields, unneededFields fields
 	unneededFields.reset()
 }
 
-func (pf *pipeFieldNames) optimize() {
-	// nothing to do
-}
-
 func (pf *pipeFieldNames) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_field_values.go b/lib/logstorage/pipe_field_values.go
index 4270f9e31..80c8408ca 100644
--- a/lib/logstorage/pipe_field_values.go
+++ b/lib/logstorage/pipe_field_values.go
@@ -46,10 +46,6 @@ func (pf *pipeFieldValues) updateNeededFields(neededFields, unneededFields field
 	}
 }
 
-func (pf *pipeFieldValues) optimize() {
-	// nothing to do
-}
-
 func (pf *pipeFieldValues) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_fields.go b/lib/logstorage/pipe_fields.go
index f08e5c608..f2dd03ba5 100644
--- a/lib/logstorage/pipe_fields.go
+++ b/lib/logstorage/pipe_fields.go
@@ -54,10 +54,6 @@ func (pf *pipeFields) updateNeededFields(neededFields, unneededFields fieldsSet)
 	unneededFields.reset()
 }
 
-func (pf *pipeFields) optimize() {
-	// nothing to do
-}
-
 func (pf *pipeFields) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_filter.go b/lib/logstorage/pipe_filter.go
index 7b7308342..2c153d26f 100644
--- a/lib/logstorage/pipe_filter.go
+++ b/lib/logstorage/pipe_filter.go
@@ -33,10 +33,6 @@ func (pf *pipeFilter) updateNeededFields(neededFields, unneededFields fieldsSet)
 	}
 }
 
-func (pf *pipeFilter) optimize() {
-	optimizeFilterIn(pf.f)
-}
-
 func (pf *pipeFilter) hasFilterInWithQuery() bool {
 	return hasFilterInWithQueryForFilter(pf.f)
 }
diff --git a/lib/logstorage/pipe_format.go b/lib/logstorage/pipe_format.go
index 765810186..6d83312b5 100644
--- a/lib/logstorage/pipe_format.go
+++ b/lib/logstorage/pipe_format.go
@@ -87,10 +87,6 @@ func (pf *pipeFormat) updateNeededFields(neededFields, unneededFields fieldsSet)
 	}
 }
 
-func (pf *pipeFormat) optimize() {
-	pf.iff.optimizeFilterIn()
-}
-
 func (pf *pipeFormat) hasFilterInWithQuery() bool {
 	return pf.iff.hasFilterInWithQuery()
 }
diff --git a/lib/logstorage/pipe_join.go b/lib/logstorage/pipe_join.go
index 82a584053..89c7556e0 100644
--- a/lib/logstorage/pipe_join.go
+++ b/lib/logstorage/pipe_join.go
@@ -18,22 +18,25 @@ type pipeJoin struct {
 	// q is a query for obtaining results for joining
 	q *Query
 
+	// prefix is the prefix to add to log fields from q query
+	prefix string
+
 	// m contains results for joining. They are automatically initialized during query execution
 	m map[string][][]Field
 }
 
 func (pj *pipeJoin) String() string {
-	return fmt.Sprintf("join by (%s) (%s)", fieldNamesString(pj.byFields), pj.q.String())
+	s := fmt.Sprintf("join by (%s) (%s)", fieldNamesString(pj.byFields), pj.q.String())
+	if pj.prefix != "" {
+		s += " prefix " + quoteTokenIfNeeded(pj.prefix)
+	}
+	return s
 }
 
 func (pj *pipeJoin) canLiveTail() bool {
 	return true
 }
 
-func (pj *pipeJoin) optimize() {
-	pj.q.Optimize()
-}
-
 func (pj *pipeJoin) hasFilterInWithQuery() bool {
 	return false
 }
@@ -43,7 +46,7 @@ func (pj *pipeJoin) initFilterInValues(_ map[string][]string, _ getFieldValuesFu
 }
 
 func (pj *pipeJoin) initJoinMap(getJoinMapFunc getJoinMapFunc) (pipe, error) {
-	m, err := getJoinMapFunc(pj.q, pj.byFields)
+	m, err := getJoinMapFunc(pj.q, pj.byFields, pj.prefix)
 	if err != nil {
 		return nil, fmt.Errorf("cannot execute query at pipe [%s]: %w", pj, err)
 	}
@@ -88,8 +91,9 @@ type pipeJoinProcessorShard struct {
 type pipeJoinProcessorShardNopad struct {
 	wctx pipeUnpackWriteContext
 
-	byValues []string
-	tmpBuf   []byte
+	byValues     []string
+	byValuesIdxs []int
+	tmpBuf       []byte
 }
 
 func (pjp *pipeJoinProcessor) writeBlock(workerID uint, br *blockResult) {
@@ -105,12 +109,19 @@ func (pjp *pipeJoinProcessor) writeBlock(workerID uint, br *blockResult) {
 	byValues := shard.byValues
 
 	cs := br.getColumns()
+	shard.byValuesIdxs = slicesutil.SetLength(shard.byValuesIdxs, len(cs))
+	byValuesIdxs := shard.byValuesIdxs
+	for i := range cs {
+		name := cs[i].name
+		byValuesIdxs[i] = slices.Index(pj.byFields, name)
+
+	}
+
 	for rowIdx := 0; rowIdx < br.rowsLen; rowIdx++ {
 		clear(byValues)
-		for i := range cs {
-			name := cs[i].name
-			if cIdx := slices.Index(pj.byFields, name); cIdx >= 0 {
-				byValues[cIdx] = cs[i].getValueAtRow(br, rowIdx)
+		for j := range cs {
+			if cIdx := byValuesIdxs[j]; cIdx >= 0 {
+				byValues[cIdx] = cs[j].getValueAtRow(br, rowIdx)
 			}
 		}
 
@@ -180,5 +191,14 @@ func parsePipeJoin(lex *lexer) (*pipeJoin, error) {
 		q:        q,
 	}
 
+	if lex.isKeyword("prefix") {
+		lex.nextToken()
+		prefix, err := getCompoundToken(lex)
+		if err != nil {
+			return nil, fmt.Errorf("cannot read prefix for [%s]: %w", pj, err)
+		}
+		pj.prefix = prefix
+	}
+
 	return pj, nil
 }
diff --git a/lib/logstorage/pipe_len.go b/lib/logstorage/pipe_len.go
index 907257e17..4d902a7e4 100644
--- a/lib/logstorage/pipe_len.go
+++ b/lib/logstorage/pipe_len.go
@@ -41,10 +41,6 @@ func (pl *pipeLen) updateNeededFields(neededFields, unneededFields fieldsSet) {
 	}
 }
 
-func (pl *pipeLen) optimize() {
-	// Nothing to do
-}
-
 func (pl *pipeLen) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_limit.go b/lib/logstorage/pipe_limit.go
index 6dec31bae..99a2e65ca 100644
--- a/lib/logstorage/pipe_limit.go
+++ b/lib/logstorage/pipe_limit.go
@@ -24,10 +24,6 @@ func (pl *pipeLimit) updateNeededFields(_, _ fieldsSet) {
 	// nothing to do
 }
 
-func (pl *pipeLimit) optimize() {
-	// nothing to do
-}
-
 func (pl *pipeLimit) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_math.go b/lib/logstorage/pipe_math.go
index c3999351f..1bc93f656 100644
--- a/lib/logstorage/pipe_math.go
+++ b/lib/logstorage/pipe_math.go
@@ -221,10 +221,6 @@ func (me *mathExpr) updateNeededFields(neededFields fieldsSet) {
 	}
 }
 
-func (pm *pipeMath) optimize() {
-	// nothing to do
-}
-
 func (pm *pipeMath) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_offset.go b/lib/logstorage/pipe_offset.go
index 5f3f92314..458858669 100644
--- a/lib/logstorage/pipe_offset.go
+++ b/lib/logstorage/pipe_offset.go
@@ -24,10 +24,6 @@ func (po *pipeOffset) updateNeededFields(_, _ fieldsSet) {
 	// nothing to do
 }
 
-func (po *pipeOffset) optimize() {
-	// nothing to do
-}
-
 func (po *pipeOffset) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_pack_json.go b/lib/logstorage/pipe_pack_json.go
index a7c2a3fec..01c7f9c7b 100644
--- a/lib/logstorage/pipe_pack_json.go
+++ b/lib/logstorage/pipe_pack_json.go
@@ -33,10 +33,6 @@ func (pp *pipePackJSON) updateNeededFields(neededFields, unneededFields fieldsSe
 	updateNeededFieldsForPipePack(neededFields, unneededFields, pp.resultField, pp.fields)
 }
 
-func (pp *pipePackJSON) optimize() {
-	// nothing to do
-}
-
 func (pp *pipePackJSON) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_pack_logfmt.go b/lib/logstorage/pipe_pack_logfmt.go
index c01c888a6..3edc4ce22 100644
--- a/lib/logstorage/pipe_pack_logfmt.go
+++ b/lib/logstorage/pipe_pack_logfmt.go
@@ -33,10 +33,6 @@ func (pp *pipePackLogfmt) updateNeededFields(neededFields, unneededFields fields
 	updateNeededFieldsForPipePack(neededFields, unneededFields, pp.resultField, pp.fields)
 }
 
-func (pp *pipePackLogfmt) optimize() {
-	// nothing to do
-}
-
 func (pp *pipePackLogfmt) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_rename.go b/lib/logstorage/pipe_rename.go
index c35c409ee..e4dbb6ac3 100644
--- a/lib/logstorage/pipe_rename.go
+++ b/lib/logstorage/pipe_rename.go
@@ -58,10 +58,6 @@ func (pr *pipeRename) updateNeededFields(neededFields, unneededFields fieldsSet)
 	}
 }
 
-func (pr *pipeRename) optimize() {
-	// nothing to do
-}
-
 func (pr *pipeRename) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_replace.go b/lib/logstorage/pipe_replace.go
index 31aaff9da..fa0ab7c1f 100644
--- a/lib/logstorage/pipe_replace.go
+++ b/lib/logstorage/pipe_replace.go
@@ -45,10 +45,6 @@ func (pr *pipeReplace) updateNeededFields(neededFields, unneededFields fieldsSet
 	updateNeededFieldsForUpdatePipe(neededFields, unneededFields, pr.field, pr.iff)
 }
 
-func (pr *pipeReplace) optimize() {
-	pr.iff.optimizeFilterIn()
-}
-
 func (pr *pipeReplace) hasFilterInWithQuery() bool {
 	return pr.iff.hasFilterInWithQuery()
 }
diff --git a/lib/logstorage/pipe_replace_regexp.go b/lib/logstorage/pipe_replace_regexp.go
index 1265da996..4f11f3cfd 100644
--- a/lib/logstorage/pipe_replace_regexp.go
+++ b/lib/logstorage/pipe_replace_regexp.go
@@ -45,10 +45,6 @@ func (pr *pipeReplaceRegexp) updateNeededFields(neededFields, unneededFields fie
 	updateNeededFieldsForUpdatePipe(neededFields, unneededFields, pr.field, pr.iff)
 }
 
-func (pr *pipeReplaceRegexp) optimize() {
-	pr.iff.optimizeFilterIn()
-}
-
 func (pr *pipeReplaceRegexp) hasFilterInWithQuery() bool {
 	return pr.iff.hasFilterInWithQuery()
 }
diff --git a/lib/logstorage/pipe_sort.go b/lib/logstorage/pipe_sort.go
index 7f9a7e779..6901cbd21 100644
--- a/lib/logstorage/pipe_sort.go
+++ b/lib/logstorage/pipe_sort.go
@@ -89,10 +89,6 @@ func (ps *pipeSort) updateNeededFields(neededFields, unneededFields fieldsSet) {
 	}
 }
 
-func (ps *pipeSort) optimize() {
-	// nothing to do
-}
-
 func (ps *pipeSort) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_stats.go b/lib/logstorage/pipe_stats.go
index 7cf723a1b..b2d85e28c 100644
--- a/lib/logstorage/pipe_stats.go
+++ b/lib/logstorage/pipe_stats.go
@@ -123,12 +123,6 @@ func (ps *pipeStats) updateNeededFields(neededFields, unneededFields fieldsSet)
 	unneededFields.reset()
 }
 
-func (ps *pipeStats) optimize() {
-	for _, f := range ps.funcs {
-		f.iff.optimizeFilterIn()
-	}
-}
-
 func (ps *pipeStats) hasFilterInWithQuery() bool {
 	for _, f := range ps.funcs {
 		if f.iff.hasFilterInWithQuery() {
diff --git a/lib/logstorage/pipe_stream_context.go b/lib/logstorage/pipe_stream_context.go
index 09400bdf5..0fe0ba95d 100644
--- a/lib/logstorage/pipe_stream_context.go
+++ b/lib/logstorage/pipe_stream_context.go
@@ -55,10 +55,6 @@ func (pc *pipeStreamContext) updateNeededFields(neededFields, unneededFields fie
 	unneededFields.removeFields(neededFieldsForStreamContext)
 }
 
-func (pc *pipeStreamContext) optimize() {
-	// nothing to do
-}
-
 func (pc *pipeStreamContext) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_top.go b/lib/logstorage/pipe_top.go
index e33a57840..ad7ff4465 100644
--- a/lib/logstorage/pipe_top.go
+++ b/lib/logstorage/pipe_top.go
@@ -71,10 +71,6 @@ func (pt *pipeTop) updateNeededFields(neededFields, unneededFields fieldsSet) {
 	}
 }
 
-func (pt *pipeTop) optimize() {
-	// nothing to do
-}
-
 func (pt *pipeTop) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_uniq.go b/lib/logstorage/pipe_uniq.go
index ddf043c58..087dffe7a 100644
--- a/lib/logstorage/pipe_uniq.go
+++ b/lib/logstorage/pipe_uniq.go
@@ -58,10 +58,6 @@ func (pu *pipeUniq) updateNeededFields(neededFields, unneededFields fieldsSet) {
 	}
 }
 
-func (pu *pipeUniq) optimize() {
-	// nothing to do
-}
-
 func (pu *pipeUniq) hasFilterInWithQuery() bool {
 	return false
 }
diff --git a/lib/logstorage/pipe_unpack_json.go b/lib/logstorage/pipe_unpack_json.go
index 635c92511..f2855ff46 100644
--- a/lib/logstorage/pipe_unpack_json.go
+++ b/lib/logstorage/pipe_unpack_json.go
@@ -60,10 +60,6 @@ func (pu *pipeUnpackJSON) updateNeededFields(neededFields, unneededFields fields
 	updateNeededFieldsForUnpackPipe(pu.fromField, pu.fields, pu.keepOriginalFields, pu.skipEmptyResults, pu.iff, neededFields, unneededFields)
 }
 
-func (pu *pipeUnpackJSON) optimize() {
-	pu.iff.optimizeFilterIn()
-}
-
 func (pu *pipeUnpackJSON) hasFilterInWithQuery() bool {
 	return pu.iff.hasFilterInWithQuery()
 }
diff --git a/lib/logstorage/pipe_unpack_logfmt.go b/lib/logstorage/pipe_unpack_logfmt.go
index ab1d4d61d..d45343c62 100644
--- a/lib/logstorage/pipe_unpack_logfmt.go
+++ b/lib/logstorage/pipe_unpack_logfmt.go
@@ -58,10 +58,6 @@ func (pu *pipeUnpackLogfmt) updateNeededFields(neededFields, unneededFields fiel
 	updateNeededFieldsForUnpackPipe(pu.fromField, pu.fields, pu.keepOriginalFields, pu.skipEmptyResults, pu.iff, neededFields, unneededFields)
 }
 
-func (pu *pipeUnpackLogfmt) optimize() {
-	pu.iff.optimizeFilterIn()
-}
-
 func (pu *pipeUnpackLogfmt) hasFilterInWithQuery() bool {
 	return pu.iff.hasFilterInWithQuery()
 }
diff --git a/lib/logstorage/pipe_unpack_syslog.go b/lib/logstorage/pipe_unpack_syslog.go
index 6717c2dcc..6eea365db 100644
--- a/lib/logstorage/pipe_unpack_syslog.go
+++ b/lib/logstorage/pipe_unpack_syslog.go
@@ -54,10 +54,6 @@ func (pu *pipeUnpackSyslog) updateNeededFields(neededFields, unneededFields fiel
 	updateNeededFieldsForUnpackPipe(pu.fromField, nil, pu.keepOriginalFields, false, pu.iff, neededFields, unneededFields)
 }
 
-func (pu *pipeUnpackSyslog) optimize() {
-	pu.iff.optimizeFilterIn()
-}
-
 func (pu *pipeUnpackSyslog) hasFilterInWithQuery() bool {
 	return pu.iff.hasFilterInWithQuery()
 }
diff --git a/lib/logstorage/pipe_unroll.go b/lib/logstorage/pipe_unroll.go
index fb9202caa..0a406d64a 100644
--- a/lib/logstorage/pipe_unroll.go
+++ b/lib/logstorage/pipe_unroll.go
@@ -36,10 +36,6 @@ func (pu *pipeUnroll) canLiveTail() bool {
 	return true
 }
 
-func (pu *pipeUnroll) optimize() {
-	pu.iff.optimizeFilterIn()
-}
-
 func (pu *pipeUnroll) hasFilterInWithQuery() bool {
 	return pu.iff.hasFilterInWithQuery()
 }
diff --git a/lib/logstorage/stats_max.go b/lib/logstorage/stats_max.go
index 7075215fc..aee16ce4d 100644
--- a/lib/logstorage/stats_max.go
+++ b/lib/logstorage/stats_max.go
@@ -114,9 +114,9 @@ func (smp *statsMaxProcessor) updateStateForColumn(br *blockResult, c *blockResu
 			smp.updateStateString(v)
 		}
 	case valueTypeDict:
-		for _, v := range c.dictValues {
+		c.forEachDictValue(br, func(v string) {
 			smp.updateStateString(v)
-		}
+		})
 	case valueTypeUint8, valueTypeUint16, valueTypeUint32, valueTypeUint64:
 		bb := bbPool.Get()
 		bb.B = marshalUint64String(bb.B[:0], c.maxValue)
diff --git a/lib/logstorage/stats_min.go b/lib/logstorage/stats_min.go
index 458fb1ee9..62f67c684 100644
--- a/lib/logstorage/stats_min.go
+++ b/lib/logstorage/stats_min.go
@@ -116,9 +116,9 @@ func (smp *statsMinProcessor) updateStateForColumn(br *blockResult, c *blockResu
 			smp.updateStateString(v)
 		}
 	case valueTypeDict:
-		for _, v := range c.dictValues {
+		c.forEachDictValue(br, func(v string) {
 			smp.updateStateString(v)
-		}
+		})
 	case valueTypeUint8, valueTypeUint16, valueTypeUint32, valueTypeUint64:
 		bb := bbPool.Get()
 		bb.B = marshalUint64String(bb.B[:0], c.minValue)
diff --git a/lib/logstorage/stats_row_max.go b/lib/logstorage/stats_row_max.go
index 7af6bd9f3..07336d472 100644
--- a/lib/logstorage/stats_row_max.go
+++ b/lib/logstorage/stats_row_max.go
@@ -82,12 +82,11 @@ func (smp *statsRowMaxProcessor) updateStatsForAllRows(br *blockResult) int {
 	case valueTypeString:
 		needUpdateState = true
 	case valueTypeDict:
-		for _, v := range c.dictValues {
-			if smp.needUpdateStateString(v) {
+		c.forEachDictValue(br, func(v string) {
+			if !needUpdateState && smp.needUpdateStateString(v) {
 				needUpdateState = true
-				break
 			}
-		}
+		})
 	case valueTypeUint8, valueTypeUint16, valueTypeUint32, valueTypeUint64:
 		bb := bbPool.Get()
 		bb.B = marshalUint64String(bb.B[:0], c.maxValue)
diff --git a/lib/logstorage/stats_row_min.go b/lib/logstorage/stats_row_min.go
index 051ddb57f..8c4516580 100644
--- a/lib/logstorage/stats_row_min.go
+++ b/lib/logstorage/stats_row_min.go
@@ -82,12 +82,11 @@ func (smp *statsRowMinProcessor) updateStatsForAllRows(br *blockResult) int {
 	case valueTypeString:
 		needUpdateState = true
 	case valueTypeDict:
-		for _, v := range c.dictValues {
-			if smp.needUpdateStateString(v) {
+		c.forEachDictValue(br, func(v string) {
+			if !needUpdateState && smp.needUpdateStateString(v) {
 				needUpdateState = true
-				break
 			}
-		}
+		})
 	case valueTypeUint8, valueTypeUint16, valueTypeUint32, valueTypeUint64:
 		bb := bbPool.Get()
 		bb.B = marshalUint64String(bb.B[:0], c.minValue)
diff --git a/lib/logstorage/stats_uniq_values.go b/lib/logstorage/stats_uniq_values.go
index 25417efc9..fec11551a 100644
--- a/lib/logstorage/stats_uniq_values.go
+++ b/lib/logstorage/stats_uniq_values.go
@@ -65,36 +65,24 @@ func (sup *statsUniqValuesProcessor) updateStatsForAllRows(br *blockResult) int
 }
 
 func (sup *statsUniqValuesProcessor) updateStatsForAllRowsColumn(c *blockResultColumn, br *blockResult) int {
-	stateSizeIncrease := 0
 	if c.isConst {
 		// collect unique const values
 		v := c.valuesEncoded[0]
-		if v == "" {
-			// skip empty values
-			return stateSizeIncrease
-		}
-		stateSizeIncrease += sup.updateState(v)
-		return stateSizeIncrease
+		return sup.updateState(v)
 	}
+
+	stateSizeIncrease := 0
 	if c.valueType == valueTypeDict {
 		// collect unique non-zero c.dictValues
-		for _, v := range c.dictValues {
-			if v == "" {
-				// skip empty values
-				continue
-			}
+		c.forEachDictValue(br, func(v string) {
 			stateSizeIncrease += sup.updateState(v)
-		}
+		})
 		return stateSizeIncrease
 	}
 
 	// slow path - collect unique values across all rows
 	values := c.getValues(br)
 	for i, v := range values {
-		if v == "" {
-			// skip empty values
-			continue
-		}
 		if i > 0 && values[i-1] == v {
 			// This value has been already counted.
 			continue
@@ -126,38 +114,23 @@ func (sup *statsUniqValuesProcessor) updateStatsForRow(br *blockResult, rowIdx i
 }
 
 func (sup *statsUniqValuesProcessor) updateStatsForRowColumn(c *blockResultColumn, br *blockResult, rowIdx int) int {
-	stateSizeIncrease := 0
 	if c.isConst {
 		// collect unique const values
 		v := c.valuesEncoded[0]
-		if v == "" {
-			// skip empty values
-			return stateSizeIncrease
-		}
-		stateSizeIncrease += sup.updateState(v)
-		return stateSizeIncrease
+		return sup.updateState(v)
 	}
+
 	if c.valueType == valueTypeDict {
 		// collect unique non-zero c.dictValues
 		valuesEncoded := c.getValuesEncoded(br)
 		dictIdx := valuesEncoded[rowIdx][0]
 		v := c.dictValues[dictIdx]
-		if v == "" {
-			// skip empty values
-			return stateSizeIncrease
-		}
-		stateSizeIncrease += sup.updateState(v)
-		return stateSizeIncrease
+		return sup.updateState(v)
 	}
 
 	// collect unique values for the given rowIdx.
 	v := c.getValueAtRow(br, rowIdx)
-	if v == "" {
-		// skip empty values
-		return stateSizeIncrease
-	}
-	stateSizeIncrease += sup.updateState(v)
-	return stateSizeIncrease
+	return sup.updateState(v)
 }
 
 func (sup *statsUniqValuesProcessor) mergeState(sfp statsProcessor) {
@@ -204,13 +177,16 @@ func sortStrings(a []string) {
 }
 
 func (sup *statsUniqValuesProcessor) updateState(v string) int {
-	stateSizeIncrease := 0
-	if _, ok := sup.m[v]; !ok {
-		vCopy := strings.Clone(v)
-		sup.m[vCopy] = struct{}{}
-		stateSizeIncrease += len(vCopy) + int(unsafe.Sizeof(vCopy))
+	if v == "" {
+		// Skip empty values
+		return 0
 	}
-	return stateSizeIncrease
+	if _, ok := sup.m[v]; ok {
+		return 0
+	}
+	vCopy := strings.Clone(v)
+	sup.m[vCopy] = struct{}{}
+	return len(vCopy) + int(unsafe.Sizeof(vCopy))
 }
 
 func (sup *statsUniqValuesProcessor) limitReached() bool {
diff --git a/lib/logstorage/stats_values.go b/lib/logstorage/stats_values.go
index eb9c100de..5d79bec5e 100644
--- a/lib/logstorage/stats_values.go
+++ b/lib/logstorage/stats_values.go
@@ -91,12 +91,13 @@ func (svp *statsValuesProcessor) updateStatsForAllRowsColumn(c *blockResultColum
 	}
 
 	values := svp.values
+	vPrev := ""
 	for _, v := range c.getValues(br) {
-		if len(values) == 0 || values[len(values)-1] != v {
-			v = strings.Clone(v)
-			stateSizeIncrease += len(v)
+		if len(values) == 0 || v != vPrev {
+			vPrev = strings.Clone(v)
+			stateSizeIncrease += len(vPrev)
 		}
-		values = append(values, v)
+		values = append(values, vPrev)
 	}
 	svp.values = values
 
diff --git a/lib/logstorage/storage_search.go b/lib/logstorage/storage_search.go
index d6c2b92ea..221e0450c 100644
--- a/lib/logstorage/storage_search.go
+++ b/lib/logstorage/storage_search.go
@@ -217,7 +217,7 @@ func (s *Storage) GetFieldNames(ctx context.Context, tenantIDs []TenantID, q *Qu
 	return s.runValuesWithHitsQuery(ctx, tenantIDs, q)
 }
 
-func (s *Storage) getJoinMap(ctx context.Context, tenantIDs []TenantID, q *Query, byFields []string) (map[string][][]Field, error) {
+func (s *Storage) getJoinMap(ctx context.Context, tenantIDs []TenantID, q *Query, byFields []string, prefix string) (map[string][][]Field, error) {
 	// TODO: track memory usage
 
 	m := make(map[string][][]Field)
@@ -229,8 +229,15 @@ func (s *Storage) getJoinMap(ctx context.Context, tenantIDs []TenantID, q *Query
 
 		cs := br.getColumns()
 		columnNames := make([]string, len(cs))
+		byValuesIdxs := make([]int, len(cs))
 		for i := range cs {
-			columnNames[i] = strings.Clone(cs[i].name)
+			name := strings.Clone(cs[i].name)
+			idx := slices.Index(byFields, name)
+			if prefix != "" && idx < 0 {
+				name = prefix + name
+			}
+			columnNames[i] = name
+			byValuesIdxs[i] = idx
 		}
 
 		byValues := make([]string, len(byFields))
@@ -242,16 +249,17 @@ func (s *Storage) getJoinMap(ctx context.Context, tenantIDs []TenantID, q *Query
 			for j := range cs {
 				name := columnNames[j]
 				v := cs[j].getValueAtRow(br, rowIdx)
-				if cIdx := slices.Index(byFields, name); cIdx >= 0 {
+				if cIdx := byValuesIdxs[j]; cIdx >= 0 {
 					byValues[cIdx] = v
 					continue
 				}
 				if v == "" {
 					continue
 				}
+				value := strings.Clone(v)
 				fields = append(fields, Field{
 					Name:  name,
-					Value: strings.Clone(v),
+					Value: value,
 				})
 			}
 
@@ -526,15 +534,15 @@ func (s *Storage) initFilterInValues(ctx context.Context, tenantIDs []TenantID,
 	return qNew, nil
 }
 
-type getJoinMapFunc func(q *Query, byFields []string) (map[string][][]Field, error)
+type getJoinMapFunc func(q *Query, byFields []string, prefix string) (map[string][][]Field, error)
 
 func (s *Storage) initJoinMaps(ctx context.Context, tenantIDs []TenantID, q *Query) (*Query, error) {
 	if !hasJoinPipes(q.pipes) {
 		return q, nil
 	}
 
-	getJoinMap := func(q *Query, byFields []string) (map[string][][]Field, error) {
-		return s.getJoinMap(ctx, tenantIDs, q, byFields)
+	getJoinMap := func(q *Query, byFields []string, prefix string) (map[string][][]Field, error) {
+		return s.getJoinMap(ctx, tenantIDs, q, byFields, prefix)
 	}
 
 	pipesNew := make([]pipe, len(q.pipes))
diff --git a/lib/logstorage/syslog_parser.go b/lib/logstorage/syslog_parser.go
index 25c5b4961..e43287092 100644
--- a/lib/logstorage/syslog_parser.go
+++ b/lib/logstorage/syslog_parser.go
@@ -240,6 +240,12 @@ func (p *SyslogParser) parseRFC5424SDLine(s string) (string, bool) {
 	sdID := s[:n]
 	s = s[n:]
 
+	if n := strings.IndexByte(sdID, '='); n >= 0 {
+		// Special case when sdID contains `key=value`
+		p.addField(sdID[:n], sdID[n+1:])
+		sdID = ""
+	}
+
 	// Parse structured data
 	i := 0
 	for i < len(s) && s[i] != ']' {
@@ -257,11 +263,19 @@ func (p *SyslogParser) parseRFC5424SDLine(s string) (string, bool) {
 		i += n + 1
 
 		// Parse value
-		qp, err := strconv.QuotedPrefix(s[i:])
-		if err != nil {
-			return s, false
+		if strings.HasPrefix(s[i:], `"`) {
+			qp, err := strconv.QuotedPrefix(s[i:])
+			if err != nil {
+				return s, false
+			}
+			i += len(qp)
+		} else {
+			n := strings.IndexAny(s[i:], " ]")
+			if n < 0 {
+				return s, false
+			}
+			i += n
 		}
-		i += len(qp)
 	}
 	if i == len(s) {
 		return s, false
@@ -272,9 +286,16 @@ func (p *SyslogParser) parseRFC5424SDLine(s string) (string, bool) {
 	p.sdParser.parse(sdValue)
 	if len(p.sdParser.fields) == 0 {
 		// Special case when structured data doesn't contain any fields
-		p.addField(sdID, "")
+		if sdID != "" {
+			p.addField(sdID, "")
+		}
 	} else {
 		for _, f := range p.sdParser.fields {
+			if sdID == "" {
+				p.addField(f.Name, f.Value)
+				continue
+			}
+
 			bufLen := len(p.buf)
 			p.buf = append(p.buf, sdID...)
 			p.buf = append(p.buf, '.')


Test patch:
diff --git a/app/vmauth/auth_config_test.go b/app/vmauth/auth_config_test.go
index 8ee7c0d18..5e686224f 100644
--- a/app/vmauth/auth_config_test.go
+++ b/app/vmauth/auth_config_test.go
@@ -777,6 +777,28 @@ func TestGetLeastLoadedBackendURL(t *testing.T) {
 	fn(7, 7, 7)
 }
 
+func TestBrokenBackend(t *testing.T) {
+	up := mustParseURLs([]string{
+		"http://node1:343",
+		"http://node2:343",
+		"http://node3:343",
+	})
+	up.loadBalancingPolicy = "least_loaded"
+	pbus := up.bus.Load()
+	bus := *pbus
+
+	// explicitly mark one of the backends as broken
+	bus[1].setBroken()
+
+	// broken backend should never return while there are healthy backends
+	for i := 0; i < 1e3; i++ {
+		b := up.getBackendURL()
+		if b.isBroken() {
+			t.Fatalf("unexpected broken backend %q", b.url)
+		}
+	}
+}
+
 func getRegexs(paths []string) []*Regex {
 	var sps []*Regex
 	for _, path := range paths {
diff --git a/apptest/testcase.go b/apptest/testcase.go
index 6ec8455fe..5cda68df8 100644
--- a/apptest/testcase.go
+++ b/apptest/testcase.go
@@ -11,11 +11,18 @@ import (
 type TestCase struct {
 	t   *testing.T
 	cli *Client
+
+	startedApps []Stopper
+}
+
+// Stopper is an interface of objects that needs to be stopped via Stop() call
+type Stopper interface {
+	Stop()
 }
 
 // NewTestCase creates a new test case.
 func NewTestCase(t *testing.T) *TestCase {
-	return &TestCase{t, NewClient()}
+	return &TestCase{t, NewClient(), nil}
 }
 
 // Dir returns the directory name that should be used by as the -storageDataDir.
@@ -29,14 +36,73 @@ func (tc *TestCase) Client() *Client {
 	return tc.cli
 }
 
-// Close performs the test case clean up, such as closing all client connections
+// Stop performs the test case clean up, such as closing all client connections
 // and removing the -storageDataDir directory.
 //
 // Note that the -storageDataDir is not removed in case of test case failure to
-// allow for furher manual debugging.
-func (tc *TestCase) Close() {
+// allow for further manual debugging.
+func (tc *TestCase) Stop() {
 	tc.cli.CloseConnections()
+	for _, app := range tc.startedApps {
+		app.Stop()
+	}
 	if !tc.t.Failed() {
 		fs.MustRemoveAll(tc.Dir())
 	}
 }
+
+// MustStartVmsingle is a test helper function that starts an instance of
+// vmsingle and fails the test if the app fails to start.
+func (tc *TestCase) MustStartVmsingle(instance string, flags []string) *Vmsingle {
+	tc.t.Helper()
+
+	app, err := StartVmsingle(instance, flags, tc.cli)
+	if err != nil {
+		tc.t.Fatalf("Could not start %s: %v", instance, err)
+	}
+	tc.addApp(app)
+	return app
+}
+
+// MustStartVmstorage is a test helper function that starts an instance of
+// vmstorage and fails the test if the app fails to start.
+func (tc *TestCase) MustStartVmstorage(instance string, flags []string) *Vmstorage {
+	tc.t.Helper()
+
+	app, err := StartVmstorage(instance, flags, tc.cli)
+	if err != nil {
+		tc.t.Fatalf("Could not start %s: %v", instance, err)
+	}
+	tc.addApp(app)
+	return app
+}
+
+// MustStartVmselect is a test helper function that starts an instance of
+// vmselect and fails the test if the app fails to start.
+func (tc *TestCase) MustStartVmselect(instance string, flags []string) *Vmselect {
+	tc.t.Helper()
+
+	app, err := StartVmselect(instance, flags, tc.cli)
+	if err != nil {
+		tc.t.Fatalf("Could not start %s: %v", instance, err)
+	}
+	tc.addApp(app)
+	return app
+}
+
+// MustStartVminsert is a test helper function that starts an instance of
+// vminsert and fails the test if the app fails to start.
+func (tc *TestCase) MustStartVminsert(instance string, flags []string) *Vminsert {
+	tc.t.Helper()
+
+	app, err := StartVminsert(instance, flags, tc.cli)
+	if err != nil {
+		tc.t.Fatalf("Could not start %s: %v", instance, err)
+	}
+	tc.addApp(app)
+	return app
+}
+
+func (tc *TestCase) addApp(app Stopper) {
+	tc.startedApps = append(tc.startedApps, app)
+}
diff --git a/apptest/tests/key_concepts_test.go b/apptest/tests/key_concepts_test.go
index b2f388ecf..3675a7c3b 100644
--- a/apptest/tests/key_concepts_test.go
+++ b/apptest/tests/key_concepts_test.go
@@ -1,11 +1,11 @@
 package tests
 
 import (
-	"testing"
-
 	"github.com/VictoriaMetrics/VictoriaMetrics/apptest"
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
+	"testing"
+	"time"
 )
 
 // Data used in examples in
@@ -30,15 +30,12 @@ var docData = []string{
 // TestSingleKeyConceptsQuery verifies cases from https://docs.victoriametrics.com/keyconcepts/#query-data
 func TestSingleKeyConceptsQuery(t *testing.T) {
 	tc := apptest.NewTestCase(t)
-	defer tc.Close()
-
-	cli := tc.Client()
+	defer tc.Stop()
 
-	vmsingle := apptest.MustStartVmsingle(t, "vmsingle", []string{
+	vmsingle := tc.MustStartVmsingle("vmsingle", []string{
 		"-storageDataPath=" + tc.Dir() + "/vmstorage",
 		"-retentionPeriod=100y",
-	}, cli)
-	defer vmsingle.Stop()
+	})
 
 	opts := apptest.QueryOpts{Timeout: "5s"}
 
@@ -53,7 +50,7 @@ func TestSingleKeyConceptsQuery(t *testing.T) {
 // TestClusterKeyConceptsQuery verifies cases from https://docs.victoriametrics.com/keyconcepts/#query-data
 func TestClusterKeyConceptsQuery(t *testing.T) {
 	tc := apptest.NewTestCase(t)
-	defer tc.Close()
+	defer tc.Stop()
 
 	// Set up the following cluster configuration:
 	//
@@ -64,29 +61,27 @@ func TestClusterKeyConceptsQuery(t *testing.T) {
 	// - vmselect points to the two vmstorages and is expected to query both
 	//   vmstorages and build the full result out of the two partial results.
 
-	cli := tc.Client()
-
-	vmstorage1 := apptest.MustStartVmstorage(t, "vmstorage-1", []string{
+	vmstorage1 := tc.MustStartVmstorage("vmstorage-1", []string{
 		"-storageDataPath=" + tc.Dir() + "/vmstorage-1",
-	}, cli)
-	defer vmstorage1.Stop()
-	vmstorage2 := apptest.MustStartVmstorage(t, "vmstorage-2", []string{
+		"-retentionPeriod=100y",
+	})
+	vmstorage2 := tc.MustStartVmstorage("vmstorage-2", []string{
 		"-storageDataPath=" + tc.Dir() + "/vmstorage-2",
-	}, cli)
-	defer vmstorage2.Stop()
-	vminsert := apptest.MustStartVminsert(t, "vminsert", []string{
+		"-retentionPeriod=100y",
+	})
+	vminsert := tc.MustStartVminsert("vminsert", []string{
 		"-storageNode=" + vmstorage1.VminsertAddr() + "," + vmstorage2.VminsertAddr(),
-	}, cli)
-	defer vminsert.Stop()
-	vmselect := apptest.MustStartVmselect(t, "vmselect", []string{
+	})
+	vmselect := tc.MustStartVmselect("vmselect", []string{
 		"-storageNode=" + vmstorage1.VmselectAddr() + "," + vmstorage2.VmselectAddr(),
-	}, cli)
-	defer vmselect.Stop()
+	})
 
 	opts := apptest.QueryOpts{Timeout: "5s", Tenant: "0"}
 
 	// Insert example data from documentation.
 	vminsert.PrometheusAPIV1ImportPrometheus(t, docData, opts)
+	time.Sleep(2 * time.Second)
+
 	vmstorage1.ForceFlush(t)
 	vmstorage2.ForceFlush(t)
 
diff --git a/apptest/tests/multilevel_test.go b/apptest/tests/multilevel_test.go
index 0cbd7c64d..f0239ee5d 100644
--- a/apptest/tests/multilevel_test.go
+++ b/apptest/tests/multilevel_test.go
@@ -11,7 +11,7 @@ import (
 
 func TestClusterMultilevelSelect(t *testing.T) {
 	tc := apptest.NewTestCase(t)
-	defer tc.Close()
+	defer tc.Stop()
 
 	// Set up the following multi-level cluster configuration:
 	//
@@ -20,24 +20,18 @@ func TestClusterMultilevelSelect(t *testing.T) {
 	// vmisert writes data into vmstorage.
 	// vmselect (L2) reads that data via vmselect (L1).
 
-	cli := tc.Client()
-
-	vmstorage := apptest.MustStartVmstorage(t, "vmstorage", []string{
+	vmstorage := tc.MustStartVmstorage("vmstorage", []string{
 		"-storageDataPath=" + tc.Dir() + "/vmstorage",
-	}, cli)
-	defer vmstorage.Stop()
-	vminsert := apptest.MustStartVminsert(t, "vminsert", []string{
+	})
+	vminsert := tc.MustStartVminsert("vminsert", []string{
 		"-storageNode=" + vmstorage.VminsertAddr(),
-	}, cli)
-	defer vminsert.Stop()
-	vmselectL1 := apptest.MustStartVmselect(t, "vmselect-level1", []string{
+	})
+	vmselectL1 := tc.MustStartVmselect("vmselect-level1", []string{
 		"-storageNode=" + vmstorage.VmselectAddr(),
-	}, cli)
-	defer vmselectL1.Stop()
-	vmselectL2 := apptest.MustStartVmselect(t, "vmselect-level2", []string{
+	})
+	vmselectL2 := tc.MustStartVmselect("vmselect-level2", []string{
 		"-storageNode=" + vmselectL1.ClusternativeListenAddr(),
-	}, cli)
-	defer vmselectL2.Stop()
+	})
 
 	// Insert 1000 unique time series.Wait for 2 seconds to let vmstorage
 	// flush pending items so they become searchable.
diff --git a/apptest/tests/sharding_test.go b/apptest/tests/sharding_test.go
index 8c0e3057f..2c21316e7 100644
--- a/apptest/tests/sharding_test.go
+++ b/apptest/tests/sharding_test.go
@@ -11,7 +11,7 @@ import (
 
 func TestClusterVminsertShardsDataVmselectBuildsFullResultFromShards(t *testing.T) {
 	tc := apptest.NewTestCase(t)
-	defer tc.Close()
+	defer tc.Stop()
 
 	// Set up the following cluster configuration:
 	//
@@ -22,24 +22,18 @@ func TestClusterVminsertShardsDataVmselectBuildsFullResultFromShards(t *testing.
 	// - vmselect points to the two vmstorages and is expected to query both
 	//   vmstorages and build the full result out of the two partial results.
 
-	cli := tc.Client()
-
-	vmstorage1 := apptest.MustStartVmstorage(t, "vmstorage-1", []string{
+	vmstorage1 := tc.MustStartVmstorage("vmstorage-1", []string{
 		"-storageDataPath=" + tc.Dir() + "/vmstorage-1",
-	}, cli)
-	defer vmstorage1.Stop()
-	vmstorage2 := apptest.MustStartVmstorage(t, "vmstorage-2", []string{
+	})
+	vmstorage2 := tc.MustStartVmstorage("vmstorage-2", []string{
 		"-storageDataPath=" + tc.Dir() + "/vmstorage-2",
-	}, cli)
-	defer vmstorage2.Stop()
-	vminsert := apptest.MustStartVminsert(t, "vminsert", []string{
+	})
+	vminsert := tc.MustStartVminsert("vminsert", []string{
 		"-storageNode=" + vmstorage1.VminsertAddr() + "," + vmstorage2.VminsertAddr(),
-	}, cli)
-	defer vminsert.Stop()
-	vmselect := apptest.MustStartVmselect(t, "vmselect", []string{
+	})
+	vmselect := tc.MustStartVmselect("vmselect", []string{
 		"-storageNode=" + vmstorage1.VmselectAddr() + "," + vmstorage2.VmselectAddr(),
-	}, cli)
-	defer vmselect.Stop()
+	})
 
 	// Insert 1000 unique time series and verify the that inserted data has been
 	// indeed sharded by checking various metrics exposed by vminsert and
diff --git a/apptest/vminsert.go b/apptest/vminsert.go
index 36e892872..47604dda3 100644
--- a/apptest/vminsert.go
+++ b/apptest/vminsert.go
@@ -18,19 +18,6 @@ type Vminsert struct {
 	cli            *Client
 }
 
-// MustStartVminsert is a test helper function that starts an instance of
-// vminsert and fails the test if the app fails to start.
-func MustStartVminsert(t *testing.T, instance string, flags []string, cli *Client) *Vminsert {
-	t.Helper()
-
-	app, err := StartVminsert(instance, flags, cli)
-	if err != nil {
-		t.Fatalf("Could not start %s: %v", instance, err)
-	}
-
-	return app
-}
-
 // StartVminsert starts an instance of vminsert with the given flags. It also
 // sets the default flags and populates the app instance state with runtime
 // values extracted from the application log (such as httpListenAddr)
diff --git a/apptest/vmselect.go b/apptest/vmselect.go
index ec98b1fae..a93fc5fbe 100644
--- a/apptest/vmselect.go
+++ b/apptest/vmselect.go
@@ -19,19 +19,6 @@ type Vmselect struct {
 	cli                     *Client
 }
 
-// MustStartVmselect is a test helper function that starts an instance of
-// vmselect and fails the test if the app fails to start.
-func MustStartVmselect(t *testing.T, instance string, flags []string, cli *Client) *Vmselect {
-	t.Helper()
-
-	app, err := StartVmselect(instance, flags, cli)
-	if err != nil {
-		t.Fatalf("Could not start %s: %v", instance, err)
-	}
-
-	return app
-}
-
 // StartVmselect starts an instance of vmselect with the given flags. It also
 // sets the default flags and populates the app instance state with runtime
 // values extracted from the application log (such as httpListenAddr)
diff --git a/apptest/vmsingle.go b/apptest/vmsingle.go
index 1de461d0f..03c5a5aa3 100644
--- a/apptest/vmsingle.go
+++ b/apptest/vmsingle.go
@@ -27,19 +27,6 @@ type Vmsingle struct {
 	prometheusAPIV1SeriesURL           string
 }
 
-// MustStartVmsingle is a test helper function that starts an instance of
-// vmsingle and fails the test if the app fails to start.
-func MustStartVmsingle(t *testing.T, instance string, flags []string, cli *Client) *Vmsingle {
-	t.Helper()
-
-	app, err := StartVmsingle(instance, flags, cli)
-	if err != nil {
-		t.Fatalf("Could not start %s: %v", instance, err)
-	}
-
-	return app
-}
-
 // StartVmsingle starts an instance of vmsingle with the given flags. It also
 // sets the default flags and populates the app instance state with runtime
 // values extracted from the application log (such as httpListenAddr).
@@ -75,8 +62,8 @@ func StartVmsingle(instance string, flags []string, cli *Client) (*Vmsingle, err
 	}, nil
 }
 
-// ForceFlush is a test helper function that forces the flushing of insterted
-// data so it becomes available for searching immediately.
+// ForceFlush is a test helper function that forces the flushing of inserted
+// data, so it becomes available for searching immediately.
 func (app *Vmsingle) ForceFlush(t *testing.T) {
 	t.Helper()
 
diff --git a/apptest/vmstorage.go b/apptest/vmstorage.go
index 92b78f9aa..361bd842e 100644
--- a/apptest/vmstorage.go
+++ b/apptest/vmstorage.go
@@ -23,19 +23,6 @@ type Vmstorage struct {
 	forceFlushURL string
 }
 
-// MustStartVmstorage is a test helper function that starts an instance of
-// vmstorage and fails the test if the app fails to start.
-func MustStartVmstorage(t *testing.T, instance string, flags []string, cli *Client) *Vmstorage {
-	t.Helper()
-
-	app, err := StartVmstorage(instance, flags, cli)
-	if err != nil {
-		t.Fatalf("Could not start %s: %v", instance, err)
-	}
-
-	return app
-}
-
 // StartVmstorage starts an instance of vmstorage with the given flags. It also
 // sets the default flags and populates the app instance state with runtime
 // values extracted from the application log (such as httpListenAddr)
@@ -85,8 +72,8 @@ func (app *Vmstorage) VmselectAddr() string {
 	return app.vmselectAddr
 }
 
-// ForceFlush is a test helper function that forces the flushing of insterted
-// data so it becomes available for searching immediately.
+// ForceFlush is a test helper function that forces the flushing of inserted
+// data, so it becomes available for searching immediately.
 func (app *Vmstorage) ForceFlush(t *testing.T) {
 	t.Helper()
 
diff --git a/lib/logstorage/filter_and_test.go b/lib/logstorage/filter_and_test.go
index 223fffb7a..8e25dd588 100644
--- a/lib/logstorage/filter_and_test.go
+++ b/lib/logstorage/filter_and_test.go
@@ -85,6 +85,13 @@ func TestGetCommonTokensForAndFilters(t *testing.T) {
 		if err != nil {
 			t.Fatalf("unexpected error in ParseQuery: %s", err)
 		}
+		if _, ok := q.f.(*filterNoop); ok {
+			if len(tokensExpected) != 0 {
+				t.Fatalf("expecting non-empty tokens %q", tokensExpected)
+			}
+			return
+		}
+
 		fa, ok := q.f.(*filterAnd)
 		if !ok {
 			t.Fatalf("unexpected filter type: %T; want *filterAnd", q.f)
@@ -134,6 +141,7 @@ func TestGetCommonTokensForAndFilters(t *testing.T) {
 			tokens: []string{"foo", "bar"},
 		},
 	})
+	f(`*`, nil)
 	f(`* *`, nil)
 
 	// empty filter must be skipped
diff --git a/lib/logstorage/parser_test.go b/lib/logstorage/parser_test.go
index 615a80a40..37b67e777 100644
--- a/lib/logstorage/parser_test.go
+++ b/lib/logstorage/parser_test.go
@@ -534,19 +534,28 @@ func TestParseFilterPrefix(t *testing.T) {
 		if err != nil {
 			t.Fatalf("unexpected error: %s", err)
 		}
-		fp, ok := q.f.(*filterPrefix)
-		if !ok {
-			t.Fatalf("unexpected filter type; got %T; want *filterPrefix; filter: %s", q.f, q.f)
-		}
-		if fp.fieldName != fieldNameExpected {
-			t.Fatalf("unexpected fieldName; got %q; want %q", fp.fieldName, fieldNameExpected)
-		}
-		if fp.prefix != prefixExpected {
-			t.Fatalf("unexpected prefix; got %q; want %q", fp.prefix, prefixExpected)
+		switch f := q.f.(type) {
+		case *filterPrefix:
+			if f.fieldName != fieldNameExpected {
+				t.Fatalf("unexpected fieldName; got %q; want %q", f.fieldName, fieldNameExpected)
+			}
+			if f.prefix != prefixExpected {
+				t.Fatalf("unexpected prefix; got %q; want %q", f.prefix, prefixExpected)
+			}
+		case *filterNoop:
+			if fieldNameExpected != "" {
+				t.Fatalf("expecting non-empty fieldName %q", fieldNameExpected)
+			}
+			if prefixExpected != "" {
+				t.Fatalf("expecting non-empty prefix %q", prefixExpected)
+			}
+		default:
+			t.Fatalf("unexpected filter type; got %T; want *filterPrefix or *filterNoop; filter: %s", q.f, q.f)
 		}
 	}
 
 	f(`*`, ``, ``)
+	f(`f:*`, `f`, ``)
 	f(`""*`, ``, ``)
 	f(`foo*`, ``, `foo`)
 	f(`abc-de.fg:foo-bar+baz*`, `abc-de.fg`, `foo-bar+baz`)
@@ -691,12 +700,20 @@ func TestParseQuerySuccess(t *testing.T) {
 	f("level: ( ((error or warn*) and re(foo))) (not (bar))", `(level:error or level:warn*) level:~foo !bar`)
 	f("!(foo bar or baz and not aa*)", `!(foo bar or baz !aa*)`)
 
+	// nested AND filters
+	f(`(foo AND bar) AND (baz AND x:y)`, `foo bar baz x:y`)
+	f(`(foo AND bar) OR (baz AND x:y)`, `foo bar or baz x:y`)
+
+	// nested OR filters
+	f(`(foo OR bar) OR (baz OR x:y)`, `foo or bar or baz or x:y`)
+	f(`(foo OR bar) AND (baz OR x:y)`, `(foo or bar) (baz or x:y)`)
+
 	// prefix search
 	f(`'foo'* and (a:x* and x:* or y:i(""*)) and i("abc def"*)`, `foo* (a:x* x:* or y:i(*)) i("abc def"*)`)
 
 	// This isn't a prefix search - it equals to `foo AND *`
-	f(`foo *`, `foo *`)
-	f(`"foo" *`, `foo *`)
+	f(`foo *`, `foo`)
+	f(`"foo" *`, `foo`)
 
 	// empty filter
 	f(`"" or foo:"" and not bar:""`, `"" or foo:"" !bar:""`)
@@ -752,7 +769,9 @@ func TestParseQuerySuccess(t *testing.T) {
 	f(`_time:[2023-01-05, 2023-01-06) OFFset 5m`, `_time:[2023-01-05,2023-01-06) offset 5m`)
 	f(`_time:(2023-01-05, 2023-01-06] OFFset 5m`, `_time:(2023-01-05,2023-01-06] offset 5m`)
 	f(`_time:(2023-01-05, 2023-01-06) OFFset 5m`, `_time:(2023-01-05,2023-01-06) offset 5m`)
-	f(`_time:1h offset 5m`, `_time:1h offset 5m`)
+	f(`_time:1h offset 5.3m`, `_time:1h offset 5.3m`)
+	f(`_time:offset 1d`, `_time:offset 1d`)
+	f(`_time:offset -1.5d`, `_time:offset -1.5d`)
 	f(`_time:1h "offSet"`, `_time:1h "offSet"`) // "offset" is a search word, since it is quoted
 	f(`_time:1h (Offset)`, `_time:1h "Offset"`) // "offset" is a search word, since it is in parens
 	f(`_time:1h "and"`, `_time:1h "and"`)       // "and" is a search word, since it is quoted
@@ -1197,13 +1216,13 @@ func TestParseQuerySuccess(t *testing.T) {
 	f(`* | uniq limit 10`, `* | uniq limit 10`)
 
 	// filter pipe
-	f(`* | filter error ip:12.3.4.5 or warn`, `* | filter error ip:12.3.4.5 or warn`)
-	f(`foo | stats by (host) count() logs | filter logs:>50 | sort by (logs desc) | limit 10`, `foo | stats by (host) count(*) as logs | filter logs:>50 | sort by (logs desc) | limit 10`)
-	f(`* | error`, `* | filter error`)
-	f(`* | "by"`, `* | filter "by"`)
-	f(`* | "stats"`, `* | filter "stats"`)
-	f(`* | "count"`, `* | filter "count"`)
-	f(`* | foo:bar AND baz:<10`, `* | filter foo:bar baz:<10`)
+	f(`* | filter error ip:12.3.4.5 or warn`, `error ip:12.3.4.5 or warn`)
+	f(`foo | stats by (host) count() logs | filter logs:>50 | sort by (logs desc) | limit 10`, `foo | stats by (host) count(*) as logs | filter logs:>50 | sort by (logs desc) limit 10`)
+	f(`* | error`, `error`)
+	f(`* | "by"`, `"by"`)
+	f(`* | "stats" *`, `"stats"`)
+	f(`* | * "count"`, `"count"`)
+	f(`* | foo:bar AND baz:<10`, `foo:bar baz:<10`)
 
 	// extract pipe
 	f(`* | extract "foo<bar>baz"`, `* | extract "foo<bar>baz"`)
@@ -1347,6 +1366,8 @@ func TestParseQueryFailure(t *testing.T) {
 	f("_time:234foo")
 	f("_time:5m offset")
 	f("_time:10m offset foobar")
+	f("_time:offset")
+	f("_time:offset foobar")
 
 	// invalid day_range filters
 	f("_time:day_range")
@@ -1364,7 +1385,7 @@ func TestParseQueryFailure(t *testing.T) {
 	f("_time:week_range[Mon,")
 	f("_time:week_range[Mon,bar")
 	f("_time:week_range[Mon,Fri")
-	f("_time:week_range[Mon,Fri] offset")
+	f("_time:week_range[Mon,Fri] offset foobar")
 
 	// long query with error
 	f(`very long query with error aaa ffdfd fdfdfd fdfd:( ffdfdfdfdfd`)
@@ -1734,7 +1755,6 @@ func TestQueryGetNeededColumns(t *testing.T) {
 		if err != nil {
 			t.Fatalf("cannot parse query [%s]: %s", s, err)
 		}
-		q.Optimize()
 
 		needed, unneeded := q.getNeededColumns()
 		neededColumns := strings.Join(needed, ",")
@@ -2200,7 +2220,6 @@ func TestQueryDropAllPipes(t *testing.T) {
 		if err != nil {
 			t.Fatalf("cannot parse [%s]: %s", qStr, err)
 		}
-		q.Optimize()
 		q.DropAllPipes()
 		result := q.String()
 		if result != resultExpected {
@@ -2410,16 +2429,17 @@ func TestQueryGetStatsByFields_Failure(t *testing.T) {
 	f(`* | by (x) count() y | format 'foo' as y`)
 }
 
-func TestHasTimeFilter(t *testing.T) {
-	f := func(qStr string, expected bool) {
+func TestQueryHasGlobalTimeFilter(t *testing.T) {
+	f := func(qStr string, resultExpected bool) {
 		t.Helper()
 
-		q, err := ParseStatsQuery(qStr)
+		q, err := ParseStatsQuery(qStr, 0)
 		if err != nil {
 			t.Fatalf("cannot parse [%s]: %s", qStr, err)
 		}
-		if q.ContainAnyTimeFilter() != expected {
-			t.Fatalf("unexpected result for hasTimeFilter(%q); want %v", qStr, expected)
+		result := q.HasGlobalTimeFilter()
+		if result != resultExpected {
+			t.Fatalf("unexpected result for hasTimeFilter(%q); got %v; want %v", qStr, result, resultExpected)
 		}
 	}
 
diff --git a/lib/logstorage/pipe_join_test.go b/lib/logstorage/pipe_join_test.go
index 644a80e46..f1fced32a 100644
--- a/lib/logstorage/pipe_join_test.go
+++ b/lib/logstorage/pipe_join_test.go
@@ -12,6 +12,7 @@ func TestParsePipeJoinSuccess(t *testing.T) {
 
 	f(`join by (foo) (error)`)
 	f(`join by (foo, bar) (a:b | fields x, y)`)
+	f(`join by (foo) (a:b) prefix c`)
 }
 
 func TestParsePipeJoinFailure(t *testing.T) {
@@ -31,6 +32,8 @@ func TestParsePipeJoinFailure(t *testing.T) {
 	f(`join by (x) ()`)
 	f(`join by (x) (`)
 	f(`join by (x) (abc`)
+	f(`join (x) (y) prefix`)
+	f(`join (x) (y) prefix |`)
 }
 
 func TestPipeJoinUpdateNeededFields(t *testing.T) {
diff --git a/lib/logstorage/storage_search_test.go b/lib/logstorage/storage_search_test.go
index 06beb6407..f08468429 100644
--- a/lib/logstorage/storage_search_test.go
+++ b/lib/logstorage/storage_search_test.go
@@ -729,7 +729,7 @@ func TestStorageRunQuery(t *testing.T) {
 			},
 		})
 	})
-	t.Run("pipe-join-single", func(t *testing.T) {
+	t.Run("pipe-join", func(t *testing.T) {
 		f(t, `'message 5' | stats by (instance) count() x
 			| join on (instance) (
 				'block 0' instance:host-1 | stats by (instance)
@@ -753,6 +753,31 @@ func TestStorageRunQuery(t *testing.T) {
 			},
 		})
 	})
+	t.Run("pipe-join-prefix", func(t *testing.T) {
+		f(t, `'message 5' | stats by (instance) count() x
+			| join on (instance) (
+				'block 0' instance:host-1 | stats by (instance)
+					count() total,
+					count_uniq(stream-id) streams,
+					count_uniq(stream-id) x
+			) prefix "abc."`, [][]Field{
+			{
+				{"instance", "host-0:234"},
+				{"x", "55"},
+			},
+			{
+				{"instance", "host-2:234"},
+				{"x", "55"},
+			},
+			{
+				{"instance", "host-1:234"},
+				{"x", "55"},
+				{"abc.total", "77"},
+				{"abc.streams", "1"},
+				{"abc.x", "1"},
+			},
+		})
+	})
 
 	// Close the storage and delete its data
 	s.MustClose()
diff --git a/lib/logstorage/syslog_parser_test.go b/lib/logstorage/syslog_parser_test.go
index ed14a607a..a5fd6a303 100644
--- a/lib/logstorage/syslog_parser_test.go
+++ b/lib/logstorage/syslog_parser_test.go
@@ -39,8 +39,8 @@ func TestSyslogParser(t *testing.T) {
 		`format=rfc5424 timestamp=2023-06-03T17:42:32.123456789Z hostname=mymachine.example.com app_name=appname proc_id=12345 msg_id=ID47 message="This is a test message with structured data."`)
 	f(`<165>1 2023-06-03T17:42:00.000Z mymachine.example.com appname 12345 ID47 [exampleSDID@32473 iut="3" eventSource="Application 123 = ] 56" eventID="11211"] This is a test message with structured data.`, time.UTC,
 		`priority=165 facility=20 severity=5 format=rfc5424 timestamp=2023-06-03T17:42:00.000Z hostname=mymachine.example.com app_name=appname proc_id=12345 msg_id=ID47 exampleSDID@32473.iut=3 exampleSDID@32473.eventSource="Application 123 = ] 56" exampleSDID@32473.eventID=11211 message="This is a test message with structured data."`)
-	f(`<165>1 2023-06-03T17:42:00.000Z mymachine.example.com appname 12345 ID47 [foo@123 iut="3"][bar@456 eventID="11211"] This is a test message with structured data.`, time.UTC,
-		`priority=165 facility=20 severity=5 format=rfc5424 timestamp=2023-06-03T17:42:00.000Z hostname=mymachine.example.com app_name=appname proc_id=12345 msg_id=ID47 foo@123.iut=3 bar@456.eventID=11211 message="This is a test message with structured data."`)
+	f(`<165>1 2023-06-03T17:42:00.000Z mymachine.example.com appname 12345 ID47 [foo@123 iut="3"][bar@456 eventID="11211"][abc=def][x=y z=a q="]= "] This is a test message with structured data.`, time.UTC,
+		`priority=165 facility=20 severity=5 format=rfc5424 timestamp=2023-06-03T17:42:00.000Z hostname=mymachine.example.com app_name=appname proc_id=12345 msg_id=ID47 foo@123.iut=3 bar@456.eventID=11211 abc=def x=y z=a q="]= " message="This is a test message with structured data."`)
 
 	// Incomplete RFC 3164
 	f("", time.UTC, ``)