From 2e635a42d886fbd60d53ecfe69e43a233515bc5b Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@victoriametrics.com>
Date: Wed, 30 Oct 2024 22:14:18 +0100
Subject: [PATCH] lib/logstorage: properly cache replace() and replace_regexp()
 results for identical adjacent field values

Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7162
---
 docs/VictoriaLogs/CHANGELOG.md             |  2 ++
 lib/logstorage/pipe_replace_regexp_test.go | 12 ++++++++++++
 lib/logstorage/pipe_replace_test.go        | 12 ++++++++++++
 lib/logstorage/pipe_update.go              |  7 +++++--
 4 files changed, 31 insertions(+), 2 deletions(-)

diff --git a/docs/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md
index 106cea8c0..7aed28ee1 100644
--- a/docs/VictoriaLogs/CHANGELOG.md
+++ b/docs/VictoriaLogs/CHANGELOG.md
@@ -17,6 +17,8 @@ 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).
+
 ## [v0.39.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.39.0-victorialogs)
 
 Released at 2024-10-30
diff --git a/lib/logstorage/pipe_replace_regexp_test.go b/lib/logstorage/pipe_replace_regexp_test.go
index 81e0230d3..f40fd43d5 100644
--- a/lib/logstorage/pipe_replace_regexp_test.go
+++ b/lib/logstorage/pipe_replace_regexp_test.go
@@ -70,6 +70,12 @@ func TestPipeReplaceRegexp(t *testing.T) {
 			{"_msg", `a_bc_d/ef`},
 			{"bar", `cde`},
 		},
+		{
+			{"_msg", `a_bc_d/ef`},
+		},
+		{
+			{"_msg", `1234`},
+		},
 		{
 			{"_msg", `1234`},
 		},
@@ -78,6 +84,12 @@ func TestPipeReplaceRegexp(t *testing.T) {
 			{"_msg", `a-bc-d-ef`},
 			{"bar", `cde`},
 		},
+		{
+			{"_msg", `a-bc-d-ef`},
+		},
+		{
+			{"_msg", `1234`},
+		},
 		{
 			{"_msg", `1234`},
 		},
diff --git a/lib/logstorage/pipe_replace_test.go b/lib/logstorage/pipe_replace_test.go
index 2663790e1..cbec9e9ca 100644
--- a/lib/logstorage/pipe_replace_test.go
+++ b/lib/logstorage/pipe_replace_test.go
@@ -48,6 +48,12 @@ func TestPipeReplace(t *testing.T) {
 			{"_msg", `a_bc_def`},
 			{"bar", `cde`},
 		},
+		{
+			{"_msg", `a_bc_def`},
+		},
+		{
+			{"_msg", `1234`},
+		},
 		{
 			{"_msg", `1234`},
 		},
@@ -56,6 +62,12 @@ func TestPipeReplace(t *testing.T) {
 			{"_msg", `a-bc-def`},
 			{"bar", `cde`},
 		},
+		{
+			{"_msg", `a-bc-def`},
+		},
+		{
+			{"_msg", `1234`},
+		},
 		{
 			{"_msg", `1234`},
 		},
diff --git a/lib/logstorage/pipe_update.go b/lib/logstorage/pipe_update.go
index 15fc02415..3ff41e14f 100644
--- a/lib/logstorage/pipe_update.go
+++ b/lib/logstorage/pipe_update.go
@@ -86,16 +86,19 @@ func (pup *pipeUpdateProcessor) writeBlock(workerID uint, br *blockResult) {
 
 	hadUpdates := false
 	vPrev := ""
+	vNew := ""
 	for rowIdx, v := range values {
 		if bm.isSetBit(rowIdx) {
 			if !hadUpdates || vPrev != v {
 				vPrev = v
 				hadUpdates = true
 
-				v = pup.updateFunc(&shard.a, v)
+				vNew = pup.updateFunc(&shard.a, v)
 			}
+			shard.rc.addValue(vNew)
+		} else {
+			shard.rc.addValue(v)
 		}
-		shard.rc.addValue(v)
 	}
 
 	br.addResultColumn(&shard.rc)