From 20fd87f86f62c6dbbd122c9623a9f302d9a29c8a Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@victoriametrics.com>
Date: Mon, 29 Apr 2024 00:05:31 +0200
Subject: [PATCH] wip

---
 lib/logstorage/block_search.go |  5 +++++
 lib/logstorage/filters.go      | 16 +++++++++++++
 lib/logstorage/filters_test.go | 41 ++++++++++++++++++++++++++++++++++
 lib/logstorage/pipes.go        |  2 ++
 4 files changed, 64 insertions(+)

diff --git a/lib/logstorage/block_search.go b/lib/logstorage/block_search.go
index c3778f109..bb3af749c 100644
--- a/lib/logstorage/block_search.go
+++ b/lib/logstorage/block_search.go
@@ -405,6 +405,11 @@ func (br *blockResult) mustInit(bs *blockSearch, bm *filterBitmap) {
 	}
 	// Initialize timestamps, since they are used for determining the number of rows in br.RowsCount()
 	srcTimestamps := bs.getTimestamps()
+	if bm.areAllBitsSet() {
+		br.timestamps = append(br.timestamps[:0], srcTimestamps...)
+		return
+	}
+
 	dstTimestamps := br.timestamps[:0]
 	bm.forEachSetBit(func(idx int) bool {
 		ts := srcTimestamps[idx]
diff --git a/lib/logstorage/filters.go b/lib/logstorage/filters.go
index 6c17d9cfe..efec45a66 100644
--- a/lib/logstorage/filters.go
+++ b/lib/logstorage/filters.go
@@ -91,6 +91,22 @@ func (bm *filterBitmap) isZero() bool {
 	return true
 }
 
+func (bm *filterBitmap) areAllBitsSet() bool {
+	a := bm.a
+	for i, word := range a {
+		if word != (1<<64)-1 {
+			if i+1 < len(a) {
+				return false
+			}
+			tailBits := bm.bitsLen % 64
+			if tailBits == 0 || word != (uint64(1)<<tailBits)-1 {
+				return false
+			}
+		}
+	}
+	return true
+}
+
 func (bm *filterBitmap) andNot(x *filterBitmap) {
 	if bm.bitsLen != x.bitsLen {
 		logger.Panicf("BUG: cannot merge bitmaps with distinct lengths; %d vs %d", bm.bitsLen, x.bitsLen)
diff --git a/lib/logstorage/filters_test.go b/lib/logstorage/filters_test.go
index e06093683..b6623bbfc 100644
--- a/lib/logstorage/filters_test.go
+++ b/lib/logstorage/filters_test.go
@@ -252,6 +252,18 @@ func TestFilterBitmap(t *testing.T) {
 			t.Fatalf("unexpected bits length: %d; want %d", bm.bitsLen, i)
 		}
 
+		if !bm.isZero() {
+			t.Fatalf("all the bits must be zero for bitmap with %d bits", i)
+		}
+		if i == 0 && !bm.areAllBitsSet() {
+			t.Fatalf("areAllBitsSet() must return true for bitmap with 0 bits")
+		}
+		if i > 0 && bm.areAllBitsSet() {
+			t.Fatalf("areAllBitsSet() must return false on new bitmap with %d bits; %#v", i, bm)
+		}
+
+		bm.setBits()
+
 		// Make sure that all the bits are set.
 		nextIdx := 0
 		bm.forEachSetBit(func(idx int) bool {
@@ -265,10 +277,28 @@ func TestFilterBitmap(t *testing.T) {
 			return true
 		})
 
+		if !bm.areAllBitsSet() {
+			t.Fatalf("all bits must be set for bitmap with %d bits", i)
+		}
+
 		// Clear a part of bits
 		bm.forEachSetBit(func(idx int) bool {
 			return idx%2 != 0
 		})
+
+		if i <= 1 && !bm.isZero() {
+			t.Fatalf("bm.isZero() must return true for bitmap with %d bits", i)
+		}
+		if i > 1 && bm.isZero() {
+			t.Fatalf("bm.isZero() must return false, since some bits are set for bitmap with %d bits", i)
+		}
+		if i == 0 && !bm.areAllBitsSet() {
+			t.Fatalf("areAllBitsSet() must return true for bitmap with 0 bits")
+		}
+		if i > 0 && bm.areAllBitsSet() {
+			t.Fatalf("some bits mustn't be set for bitmap with %d bits", i)
+		}
+
 		nextIdx = 1
 		bm.forEachSetBit(func(idx int) bool {
 			if idx != nextIdx {
@@ -282,6 +312,17 @@ func TestFilterBitmap(t *testing.T) {
 		bm.forEachSetBit(func(_ int) bool {
 			return false
 		})
+
+		if !bm.isZero() {
+			t.Fatalf("all the bits must be reset for bitmap with %d bits", i)
+		}
+		if i == 0 && !bm.areAllBitsSet() {
+			t.Fatalf("allAllBitsSet() must return true for bitmap with 0 bits")
+		}
+		if i > 0 && bm.areAllBitsSet() {
+			t.Fatalf("areAllBitsSet() must return false for bitmap with %d bits", i)
+		}
+
 		bitsCount := 0
 		bm.forEachSetBit(func(_ int) bool {
 			bitsCount++
diff --git a/lib/logstorage/pipes.go b/lib/logstorage/pipes.go
index f5fe0125a..72adddb30 100644
--- a/lib/logstorage/pipes.go
+++ b/lib/logstorage/pipes.go
@@ -616,6 +616,8 @@ func (sfcp *statsFuncCountProcessor) updateStatsForAllRows(timestamps []int64, c
 
 	// Slow path - count rows containing at least a single non-empty value for the fields enumerated inside count().
 	bm := getFilterBitmap(len(timestamps))
+	defer putFilterBitmap(bm)
+
 	bm.setBits()
 	for _, f := range fields {
 		if idx := getBlockColumnIndex(columns, f); idx >= 0 {