github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-01-10 15:14:09 +00:00
Code Issues Projects Releases Packages Wiki Activity

wip

Browse source
This commit is contained in:
Aliaksandr Valialkin 2024-04-29 03:27:46 +02:00
parent dd55ed98a8
commit 5d791d617b
No known key found for this signature in database
GPG key ID: 52C003EE2BCDB9EB
2 changed files with 21 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/logstorage/parser.go
View file

@ -204,7 +204,7 @@ func (q *Query) String() string {
func (q *Query) getResultColumnNames() []string { func (q *Query) getResultColumnNames() []string {
for _, p := range q.pipes { for _, p := range q.pipes {
switch t := p.(type) { switch t := p.(type) {
case *fieldsPipe: case *pipeFields:
return t.fields return t.fields
case *statsPipe: case *statsPipe:
return t.neededFields() return t.neededFields()

40
lib/logstorage/pipes.go
View file

@ -77,11 +77,11 @@ func parsePipes(lex *lexer) ([]pipe, error) {
} }
switch { switch {
case lex.isKeyword("fields"): case lex.isKeyword("fields"):
fp, err := parseFieldsPipe(lex) pf, err := parsePipeFields(lex)
if err != nil { if err != nil {
return nil, fmt.Errorf("cannot parse 'fields' pipe: %w", err) return nil, fmt.Errorf("cannot parse 'fields' pipe: %w", err)
} }
pipes = append(pipes, fp) pipes = append(pipes, pf)
case lex.isKeyword("stats"): case lex.isKeyword("stats"):
sp, err := parseStatsPipe(lex) sp, err := parseStatsPipe(lex)
if err != nil { if err != nil {
@ -107,7 +107,7 @@ func parsePipes(lex *lexer) ([]pipe, error) {
return pipes, nil return pipes, nil
} }
type fieldsPipe struct { type pipeFields struct {
// fields contains list of fields to fetch // fields contains list of fields to fetch
fields []string fields []string
@ -115,36 +115,36 @@ type fieldsPipe struct {
containsStar bool containsStar bool
} }
func (fp *fieldsPipe) String() string { func (pf *pipeFields) String() string {
if len(fp.fields) == 0 { if len(pf.fields) == 0 {
logger.Panicf("BUG: fieldsPipe must contain at least a single field") logger.Panicf("BUG: pipeFields must contain at least a single field")
} }
return "fields " + fieldNamesString(fp.fields) return "fields " + fieldNamesString(pf.fields)
} }
func (fp *fieldsPipe) newPipeProcessor(_ int, _ <-chan struct{}, _ func(), ppBase pipeProcessor) pipeProcessor { func (pf *pipeFields) newPipeProcessor(_ int, _ <-chan struct{}, _ func(), ppBase pipeProcessor) pipeProcessor {
return &fieldsPipeProcessor{ return &pipeFieldsProcessor{
fp: fp, pf: pf,
ppBase: ppBase, ppBase: ppBase,
} }
} }
type fieldsPipeProcessor struct { type pipeFieldsProcessor struct {
fp *fieldsPipe pf *pipeFields
ppBase pipeProcessor ppBase pipeProcessor
} }
func (fpp *fieldsPipeProcessor) writeBlock(workerID uint, timestamps []int64, columns []BlockColumn) { func (fpp *pipeFieldsProcessor) writeBlock(workerID uint, timestamps []int64, columns []BlockColumn) {
if fpp.fp.containsStar || areSameBlockColumns(columns, fpp.fp.fields) { if fpp.pf.containsStar || areSameBlockColumns(columns, fpp.pf.fields) {
// Fast path - there is no need in additional transformations before writing the block to ppBase. // Fast path - there is no need in additional transformations before writing the block to ppBase.
fpp.ppBase.writeBlock(workerID, timestamps, columns) fpp.ppBase.writeBlock(workerID, timestamps, columns)
return return
} }
// Slow path - construct columns for fpp.fp.fields before writing them to ppBase. // Slow path - construct columns for fpp.pf.fields before writing them to ppBase.
brs := getBlockRows() brs := getBlockRows()
cs := brs.cs cs := brs.cs
for _, f := range fpp.fp.fields { for _, f := range fpp.pf.fields {
values := getBlockColumnValues(columns, f, len(timestamps)) values := getBlockColumnValues(columns, f, len(timestamps))
cs = append(cs, BlockColumn{ cs = append(cs, BlockColumn{
Name: f, Name: f,
@ -156,11 +156,11 @@ func (fpp *fieldsPipeProcessor) writeBlock(workerID uint, timestamps []int64, co
putBlockRows(brs) putBlockRows(brs)
} }
func (fpp *fieldsPipeProcessor) flush() error { func (fpp *pipeFieldsProcessor) flush() error {
return nil return nil
} }
func parseFieldsPipe(lex *lexer) (*fieldsPipe, error) { func parsePipeFields(lex *lexer) (*pipeFields, error) {
var fields []string var fields []string
for { for {
if !lex.mustNextToken() { if !lex.mustNextToken() {
@ -176,11 +176,11 @@ func parseFieldsPipe(lex *lexer) (*fieldsPipe, error) {
fields = append(fields, field) fields = append(fields, field)
switch { switch {
case lex.isKeyword("|", ")", ""): case lex.isKeyword("|", ")", ""):
fp := &fieldsPipe{ pf := &pipeFields{
fields: fields, fields: fields,
containsStar: slices.Contains(fields, "*"), containsStar: slices.Contains(fields, "*"),
} }
return fp, nil return pf, nil
case lex.isKeyword(","): case lex.isKeyword(","):
default: default:
return nil, fmt.Errorf("unexpected token: %q; expecting ',', '|' or ')'", lex.token) return nil, fmt.Errorf("unexpected token: %q; expecting ',', '|' or ')'", lex.token)