github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2024-11-21 14:44:00 +00:00
Code Issues Projects Releases Packages Wiki Activity

lib/logstorage: properly reset cached output fields for extract and extract_regexp pipes after the log entry matches if(...) condition

Browse source 
Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7162

(cherry picked from commit c5d08d317c)
This commit is contained in:
Aliaksandr Valialkin 2024-10-30 22:29:28 +01:00 committed by hagen1778
parent b74bcb7886
commit 1ea65d662f
No known key found for this signature in database
GPG key ID: E92986095E0DD614
5 changed files with 20 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
docs/VictoriaLogs/CHANGELOG.md
View file

@ -18,6 +18,7 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
* FEATURE: add support for extra filters across all the [HTTP querying APIs](https://docs.victoriametrics.com/victorialogs/querying/#http-api). See [these docs](https://docs.victoriametrics.com/victorialogs/querying/#extra-filters) for details. This is needed for implementing quick filtering on field values at [this feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7365).
* BUGFIX: properly apply [`replace`](https://docs.victoriametrics.com/victorialogs/logsql/#replace-pipe) and [`replace_regexp`](https://docs.victoriametrics.com/victorialogs/logsql/#replace_regexp-pipe) pipes to identical values in adjacent log entries. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7162).
* BUGFIX: properly apply [`extract`](https://docs.victoriametrics.com/victorialogs/logsql/#extract-pipe) and [`extract_regexp`](https://docs.victoriametrics.com/victorialogs/logsql/#extract_regexp-pipe) pipe with additional `if (...)` filter (aka [conditional extract](https://docs.victoriametrics.com/victorialogs/logsql/#conditional-extract) and [conditional extract_regexp](https://docs.victoriametrics.com/victorialogs/logsql/#conditional-extract_regexp)).
## [v0.39.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.39.0-victorialogs)

7
lib/logstorage/pipe_extract.go
View file

@ -192,13 +192,13 @@ func (pep *pipeExtractProcessor) writeBlock(workerID uint, br *blockResult) {
shard.resultValues = slicesutil.SetLength(shard.resultValues, len(rcs))
resultValues := shard.resultValues
hadUpdates := false
needUpdates := true
vPrev := ""
for rowIdx, v := range values {
if bm.isSetBit(rowIdx) {
if !hadUpdates || vPrev != v {
if needUpdates || vPrev != v {
vPrev = v
hadUpdates = true
needUpdates = false
ptn.apply(v)
@ -219,6 +219,7 @@ func (pep *pipeExtractProcessor) writeBlock(workerID uint, br *blockResult) {
for i, c := range resultColumns {
resultValues[i] = c.getValueAtRow(br, rowIdx)
}
needUpdates = true
}
for i, v := range resultValues {

7
lib/logstorage/pipe_extract_regexp.go
View file

@ -215,13 +215,13 @@ func (pep *pipeExtractRegexpProcessor) writeBlock(workerID uint, br *blockResult
shard.resultValues = slicesutil.SetLength(shard.resultValues, len(rcs))
resultValues := shard.resultValues
hadUpdates := false
needUpdates := true
vPrev := ""
for rowIdx, v := range values {
if bm.isSetBit(rowIdx) {
if !hadUpdates || vPrev != v {
if needUpdates || vPrev != v {
vPrev = v
hadUpdates = true
needUpdates = false
shard.apply(pe.re, v)
@ -246,6 +246,7 @@ func (pep *pipeExtractRegexpProcessor) writeBlock(workerID uint, br *blockResult
resultValues[i] = c.getValueAtRow(br, rowIdx)
}
}
needUpdates = true
}
for i, v := range resultValues {

8
lib/logstorage/pipe_extract_test.go
View file

@ -211,12 +211,20 @@ func TestPipeExtract(t *testing.T) {
{"x", `a foo=cc baz=aa b`},
{"bar", "abc"},
},
{
{"x", `a foo=cc baz=aa b`},
},
}, [][]Field{
{
{"x", `a foo=cc baz=aa b`},
{"bar", `cc`},
{"xx", `aa b`},
},
{
{"x", `a foo=cc baz=aa b`},
{"bar", `cc`},
{"xx", `aa b`},
},
})
// single row, if mismatch

6
lib/logstorage/pipe_update.go
View file

@ -84,14 +84,14 @@ func (pup *pipeUpdateProcessor) writeBlock(workerID uint, br *blockResult) {
c := br.getColumnByName(pup.field)
values := c.getValues(br)
hadUpdates := false
needUpdates := true
vPrev := ""
vNew := ""
for rowIdx, v := range values {
if bm.isSetBit(rowIdx) {
if !hadUpdates || vPrev != v {
if needUpdates || vPrev != v {
vPrev = v
hadUpdates = true
needUpdates = false
vNew = pup.updateFunc(&shard.a, v)
}