From cb42a1a6fc8923d89da0c8348857864e7a962d7a Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@victoriametrics.com>
Date: Mon, 29 Apr 2024 02:18:48 +0200
Subject: [PATCH] wip

---
 lib/logstorage/pipes.go | 28 +++++++++++++++++++++++++++-
 1 file changed, 27 insertions(+), 1 deletion(-)

diff --git a/lib/logstorage/pipes.go b/lib/logstorage/pipes.go
index 266207e40..c8e97af43 100644
--- a/lib/logstorage/pipes.go
+++ b/lib/logstorage/pipes.go
@@ -737,6 +737,18 @@ func (sfup *statsFuncUniqProcessor) updateStatsForAllRows(timestamps []int64, co
 		// Count unique rows
 		keyBuf := sfup.keyBuf
 		for i := range timestamps {
+			seenKey := true
+			for _, c := range columns {
+				values := c.Values
+				if i == 0 || values[i-1] != values[i] {
+					seenKey = false
+					break
+				}
+			}
+			if seenKey {
+				continue
+			}
+
 			allEmptyValues := true
 			keyBuf = keyBuf[:0]
 			for _, c := range columns {
@@ -763,11 +775,15 @@ func (sfup *statsFuncUniqProcessor) updateStatsForAllRows(timestamps []int64, co
 	if len(fields) == 1 {
 		// Fast path for a single column
 		if idx := getBlockColumnIndex(columns, fields[0]); idx >= 0 {
-			for _, v := range columns[idx].Values {
+			values := columns[idx].Values
+			for i, v := range values {
 				if v == "" {
 					// Do not count empty values
 					continue
 				}
+				if i > 0 && values[i-1] == v {
+					continue
+				}
 				if _, ok := m[v]; !ok {
 					vCopy := strings.Clone(v)
 					m[vCopy] = struct{}{}
@@ -786,6 +802,16 @@ func (sfup *statsFuncUniqProcessor) updateStatsForAllRows(timestamps []int64, co
 
 	keyBuf := sfup.keyBuf
 	for i := range timestamps {
+		seenKey := true
+		for _, values := range columnValues {
+			if i == 0 || values[i-1] != values[i] {
+				seenKey = false
+			}
+		}
+		if seenKey {
+			continue
+		}
+
 		allEmptyValues := true
 		keyBuf = keyBuf[:0]
 		for _, values := range columnValues {