github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-12-31 15:06:26 +00:00
Code Issues Projects Releases Packages Wiki Activity

wip

Browse source
This commit is contained in:
Aliaksandr Valialkin 2024-05-30 12:29:24 +02:00
parent 277f13f3fa
commit fa20f150a6
No known key found for this signature in database
GPG key ID: 52C003EE2BCDB9EB
11 changed files with 56 additions and 43 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
lib/logstorage/parser_test.go
View file

@ -1552,7 +1552,7 @@ func TestQueryGetNeededColumns(t *testing.T) {
f(`* | fields f1, f2 | sort by(f3)`, `f1,f2`, ``)
f(`* | fields f1, f2 | sort by(f1,f3)`, `f1,f2`, ``)
f(`* | fields f1, f2 | sort by(f3) | stats count() r1`, ``, ``)
f(`* | fields f1, f2 | sort by(f1) | stats count() r1`, `f1`, ``)
f(`* | fields f1, f2 | sort by(f1) | stats count() r1`, ``, ``)
f(`* | fields f1, f2 | sort by(f1) | stats count(f2,f3) r1`, `f1,f2`, ``)
f(`* | fields f1, f2 | sort by(f3) | fields f2`, `f2`, ``)
f(`* | fields f1, f2 | sort by(f1,f3) | fields f2`, `f1,f2`, ``)
@ -1783,17 +1783,41 @@ func TestQueryGetNeededColumns(t *testing.T) {
f(`* | rm f1, f2 | stats by(f3) count(f4) r1`, `f3,f4`, ``)
// Verify that fields are correctly tracked before count(*)
f(`* | copy a b, c d | count() r1`, `a`, ``)
f(`* | delete a, b | count() r1`, `*`, `a`)
f(`* | extract "<f1>bar" from x | count() r1`, `x`, ``)
f(`* | extract if (q:w p:a) "<f1>bar" from x | count() r1`, `p,q,x`, ``)
f(`* | extract_regexp "(?P<f1>.*)bar" from x | count() r1`, `x`, ``)
f(`* | extract_regexp if (q:w p:a) "(?P<f1>.*)bar" from x | count() r1`, `p,q,x`, ``)
f(`* | copy a b, c d | count() r1`, ``, ``)
f(`* | delete a, b | count() r1`, ``, ``)
f(`* | extract "<f1>bar" from x | count() r1`, ``, ``)
f(`* | extract if (q:w p:a) "<f1>bar" from x | count() r1`, `p,q`, ``)
f(`* | extract_regexp "(?P<f1>.*)bar" from x | count() r1`, ``, ``)
f(`* | extract_regexp if (q:w p:a) "(?P<f1>.*)bar" from x | count() r1`, `p,q`, ``)
f(`* | field_names | count() r1`, `*`, `_time`)
f(`* | limit 10 | field_names as abc | count() r1`, `*`, ``)
f(`* | fields a, b | count() r1`, `a`, ``)
f(`* | fields a, b | count() r1`, ``, ``)
f(`* | field_values a | count() r1`, `a`, ``)
f(`* | limit 10 | filter a:b c:d | count() r1`, `a,c`, ``)
f(`* | format "<a><b>" as c | count() r1`, `a,b`, ``)
f(`* | format if (q:w p:a) "<a><b>" as c | count() r1`, `a,b,p,q`, ``)
f(`* | limit 10 | count() r1`, ``, ``)
f(`* | format "<a><b>" as c | count() r1`, ``, ``)
f(`* | format if (q:w p:a) "<a><b>" as c | count() r1`, `p,q`, ``)
f(`* | math (a + b) as c, d * 2 as x | count() r1`, ``, ``)
f(`* | offset 10 | count() r1`, ``, ``)
f(`* | pack_json | count() r1`, ``, ``)
f(`* | pack_json fields(a,b) | count() r1`, ``, ``)
f(`* | rename a b, c d | count() r1`, ``, ``)
f(`* | replace ("a", "b") at x | count() r1`, ``, ``)
f(`* | replace if (q:w p:a) ("a", "b") at x | count() r1`, `p,q`, ``)
f(`* | replace_regexp ("a", "b") at x | count() r1`, ``, ``)
f(`* | replace_regexp if (q:w p:a) ("a", "b") at x | count() r1`, `p,q`, ``)
f(`* | sort by (a,b) | count() r1`, ``, ``)
f(`* | stats count_uniq(a, b) as c | count() r1`, ``, ``)
f(`* | stats count_uniq(a, b) if (q:w p:a) as c | count() r1`, ``, ``)
f(`* | stats by (a1,a2) count_uniq(a, b) as c | count() r1`, `a1,a2`, ``)
f(`* | stats by (a1,a2) count_uniq(a, b) if (q:w p:a) as c | count() r1`, `a1,a2`, ``)
f(`* | uniq by (a, b) | count() r1`, `a,b`, ``)
f(`* | unpack_json from x | count() r1`, ``, ``)
f(`* | unpack_json from x fields (a,b) | count() r1`, ``, ``)
f(`* | unpack_json if (q:w p:a) from x | count() r1`, `p,q`, ``)
f(`* | unpack_json if (q:w p:a) from x fields(a,b) | count() r1`, `p,q`, ``)
f(`* | unpack_logfmt from x | count() r1`, ``, ``)
f(`* | unpack_logfmt from x fields (a,b) | count() r1`, ``, ``)
f(`* | unpack_logfmt if (q:w p:a) from x | count() r1`, `p,q`, ``)
f(`* | unpack_logfmt if (q:w p:a) from x fields(a,b) | count() r1`, `p,q`, ``)
}

5
lib/logstorage/pipe_copy.go
View file

@ -32,11 +32,6 @@ func (pc *pipeCopy) String() string {
}
func (pc *pipeCopy) updateNeededFields(neededFields, unneededFields fieldsSet) {
if neededFields.isEmpty() {
neededFields.add(pc.srcFields[0])
return
}
for i := len(pc.srcFields) - 1; i >= 0; i-- {
srcField := pc.srcFields[i]
dstField := pc.dstFields[i]

6
lib/logstorage/pipe_delete.go
View file

@ -23,12 +23,6 @@ func (pd *pipeDelete) String() string {
}
func (pd *pipeDelete) updateNeededFields(neededFields, unneededFields fieldsSet) {
if neededFields.isEmpty() {
neededFields.add("*")
unneededFields.add(pd.fields[0])
return
}
if neededFields.contains("*") {
unneededFields.addFields(pd.fields)
} else {

1
lib/logstorage/pipe_extract.go
View file

@ -61,7 +61,6 @@ func (pe *pipeExtract) initFilterInValues(cache map[string][]string, getFieldVal
func (pe *pipeExtract) updateNeededFields(neededFields, unneededFields fieldsSet) {
if neededFields.isEmpty() {
neededFields.add(pe.fromField)
if pe.iff != nil {
neededFields.addFields(pe.iff.neededFields)
}

1
lib/logstorage/pipe_extract_regexp.go
View file

@ -63,7 +63,6 @@ func (pe *pipeExtractRegexp) initFilterInValues(cache map[string][]string, getFi
func (pe *pipeExtractRegexp) updateNeededFields(neededFields, unneededFields fieldsSet) {
if neededFields.isEmpty() {
neededFields.add(pe.fromField)
if pe.iff != nil {
neededFields.addFields(pe.iff.neededFields)
}

4
lib/logstorage/pipe_fields.go
View file

@ -29,10 +29,6 @@ func (pf *pipeFields) updateNeededFields(neededFields, unneededFields fieldsSet)
if pf.containsStar {
return
}
if neededFields.isEmpty() {
neededFields.add(pf.fields[0])
return
}
if neededFields.contains("*") {
// subtract unneeded fields from pf.fields

5
lib/logstorage/pipe_format.go
View file

@ -44,11 +44,6 @@ func (pf *pipeFormat) String() string {
func (pf *pipeFormat) updateNeededFields(neededFields, unneededFields fieldsSet) {
if neededFields.isEmpty() {
for _, step := range pf.steps {
if step.field != "" {
neededFields.add(step.field)
}
}
if pf.iff != nil {
neededFields.addFields(pf.iff.neededFields)
}

15
lib/logstorage/pipe_math.go
View file

@ -175,26 +175,19 @@ func (pm *pipeMath) updateNeededFields(neededFields, unneededFields fieldsSet) {
if !unneededFields.contains(e.resultField) {
unneededFields.add(e.resultField)
entryFields := e.getNeededFields()
unneededFields.removeFields(entryFields)
fs := newFieldsSet()
e.expr.updateNeededFields(fs)
unneededFields.removeFields(fs.getAll())
}
} else {
if neededFields.contains(e.resultField) {
neededFields.remove(e.resultField)
entryFields := e.getNeededFields()
neededFields.addFields(entryFields)
e.expr.updateNeededFields(neededFields)
}
}
}
}
func (me *mathEntry) getNeededFields() []string {
neededFields := newFieldsSet()
me.expr.updateNeededFields(neededFields)
return neededFields.getAll()
}
func (me *mathExpr) updateNeededFields(neededFields fieldsSet) {
if me.isConst {
return

4
lib/logstorage/pipe_sort.go
View file

@ -56,6 +56,10 @@ func (ps *pipeSort) String() string {
}
func (ps *pipeSort) updateNeededFields(neededFields, unneededFields fieldsSet) {
if neededFields.isEmpty() {
return
}
if len(ps.byFields) == 0 {
neededFields.add("*")
unneededFields.reset()

7
lib/logstorage/pipe_unpack.go
View file

@ -7,6 +7,13 @@ import (
)
func updateNeededFieldsForUnpackPipe(fromField string, outFields []string, keepOriginalFields, skipEmptyResults bool, iff *ifFilter, neededFields, unneededFields fieldsSet) {
if neededFields.isEmpty() {
if iff != nil {
neededFields.addFields(iff.neededFields)
}
return
}
if neededFields.contains("*") {
unneededFieldsOrig := unneededFields.clone()
unneededFieldsCount := 0

7
lib/logstorage/pipe_update.go
View file

@ -5,6 +5,13 @@ import (
)
func updateNeededFieldsForUpdatePipe(neededFields, unneededFields fieldsSet, field string, iff *ifFilter) {
if neededFields.isEmpty() {
if iff != nil {
neededFields.addFields(iff.neededFields)
}
return
}
if neededFields.contains("*") {
if !unneededFields.contains(field) && iff != nil {
unneededFields.removeFields(iff.neededFields)