mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2025-02-19 15:30:17 +00:00
fea934936b
The purpose of extra filters ( https://docs.victoriametrics.com/victorialogs/querying/#extra-filters ) is to limit the subset of logs, which can be queried. For example, it is expected that all the queries with `extra_filters={tenant=123}` can access only logs, which contain `123` value for the `tenant` field. Previously this wasn't the case, since the provided extra filters weren't applied to subqueries. For example, the following query could be used to select all the logs outside `tenant=123`, for any `extra_filters` arg: * | union({tenant!=123}) This commit fixes this by propagating extra filters to all the subqueries. While at it, this commit also properly propagates [start, end] time range filter from HTTP querying APIs into all the subqueries, since this is what most users expect. This behaviour can be overriden on per-subquery basis with the `options(ignore_global_time_filter=true)` option - see https://docs.victoriametrics.com/victorialogs/logsql/#query-options Also properly apply apply optimizations across all the subqueries. Previously the optimizations at Query.optimize() function were applied only to the top-level query.
134 lines
3 KiB
Go
134 lines
3 KiB
Go
package logstorage
|
|
|
|
import (
|
|
"fmt"
|
|
"strings"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
|
|
)
|
|
|
|
// pipeRename implements '| rename ...' pipe.
|
|
//
|
|
// See https://docs.victoriametrics.com/victorialogs/logsql/#rename-pipe
|
|
type pipeRename struct {
|
|
// srcFields contains a list of source fields to rename
|
|
srcFields []string
|
|
|
|
// dstFields contains a list of destination fields
|
|
dstFields []string
|
|
}
|
|
|
|
func (pr *pipeRename) String() string {
|
|
if len(pr.srcFields) == 0 {
|
|
logger.Panicf("BUG: pipeRename must contain at least a single srcField")
|
|
}
|
|
|
|
a := make([]string, len(pr.srcFields))
|
|
for i, srcField := range pr.srcFields {
|
|
dstField := pr.dstFields[i]
|
|
a[i] = quoteTokenIfNeeded(srcField) + " as " + quoteTokenIfNeeded(dstField)
|
|
}
|
|
return "rename " + strings.Join(a, ", ")
|
|
}
|
|
|
|
func (pr *pipeRename) canLiveTail() bool {
|
|
return true
|
|
}
|
|
|
|
func (pr *pipeRename) updateNeededFields(neededFields, unneededFields fieldsSet) {
|
|
for i := len(pr.srcFields) - 1; i >= 0; i-- {
|
|
srcField := pr.srcFields[i]
|
|
dstField := pr.dstFields[i]
|
|
|
|
if neededFields.contains("*") {
|
|
if unneededFields.contains(dstField) {
|
|
unneededFields.add(srcField)
|
|
} else {
|
|
unneededFields.add(dstField)
|
|
unneededFields.remove(srcField)
|
|
}
|
|
} else {
|
|
if neededFields.contains(dstField) {
|
|
neededFields.remove(dstField)
|
|
neededFields.add(srcField)
|
|
} else {
|
|
neededFields.remove(srcField)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (pr *pipeRename) hasFilterInWithQuery() bool {
|
|
return false
|
|
}
|
|
|
|
func (pr *pipeRename) initFilterInValues(_ *inValuesCache, _ getFieldValuesFunc) (pipe, error) {
|
|
return pr, nil
|
|
}
|
|
|
|
func (pr *pipeRename) visitSubqueries(_ func(q *Query)) {
|
|
// nothing to do
|
|
}
|
|
|
|
func (pr *pipeRename) newPipeProcessor(_ int, _ <-chan struct{}, _ func(), ppNext pipeProcessor) pipeProcessor {
|
|
return &pipeRenameProcessor{
|
|
pr: pr,
|
|
ppNext: ppNext,
|
|
}
|
|
}
|
|
|
|
type pipeRenameProcessor struct {
|
|
pr *pipeRename
|
|
ppNext pipeProcessor
|
|
}
|
|
|
|
func (prp *pipeRenameProcessor) writeBlock(workerID uint, br *blockResult) {
|
|
if br.rowsLen == 0 {
|
|
return
|
|
}
|
|
|
|
br.renameColumns(prp.pr.srcFields, prp.pr.dstFields)
|
|
prp.ppNext.writeBlock(workerID, br)
|
|
}
|
|
|
|
func (prp *pipeRenameProcessor) flush() error {
|
|
return nil
|
|
}
|
|
|
|
func parsePipeRename(lex *lexer) (pipe, error) {
|
|
if !lex.isKeyword("rename", "mv") {
|
|
return nil, fmt.Errorf("expecting 'rename' or 'mv'; got %q", lex.token)
|
|
}
|
|
|
|
var srcFields []string
|
|
var dstFields []string
|
|
for {
|
|
lex.nextToken()
|
|
srcField, err := parseFieldName(lex)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("cannot parse src field name: %w", err)
|
|
}
|
|
if lex.isKeyword("as") {
|
|
lex.nextToken()
|
|
}
|
|
dstField, err := parseFieldName(lex)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("cannot parse dst field name: %w", err)
|
|
}
|
|
|
|
srcFields = append(srcFields, srcField)
|
|
dstFields = append(dstFields, dstField)
|
|
|
|
switch {
|
|
case lex.isKeyword("|", ")", ""):
|
|
pr := &pipeRename{
|
|
srcFields: srcFields,
|
|
dstFields: dstFields,
|
|
}
|
|
return pr, nil
|
|
case lex.isKeyword(","):
|
|
default:
|
|
return nil, fmt.Errorf("unexpected token: %q; expecting ',', '|' or ')'", lex.token)
|
|
}
|
|
}
|
|
}
|