From 2dbb12563bb30500bec3c6d2b862b60c086980f2 Mon Sep 17 00:00:00 2001
From: Aliaksandr Valialkin <valyala@gmail.com>
Date: Mon, 8 Feb 2021 12:00:44 +0200
Subject: [PATCH] lib/storage: optimize data ingestion in the beginning of
 every hour

Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1046
---
 lib/storage/storage.go | 20 ++++++++++++++------
 1 file changed, 14 insertions(+), 6 deletions(-)

diff --git a/lib/storage/storage.go b/lib/storage/storage.go
index b0c5d42650..c74f15af56 100644
--- a/lib/storage/storage.go
+++ b/lib/storage/storage.go
@@ -1670,6 +1670,8 @@ func (s *Storage) updatePerDateData(rows []rawRow) error {
 		projectID uint32
 	}
 	var pendingDateMetricIDs []pendingDateMetricID
+	var pendingNextDayMetricIDs []uint64
+	var pendingHourEntries []pendingHourMetricIDEntry
 	for i := range rows {
 		r := &rows[i]
 		if r.Timestamp != prevTimestamp {
@@ -1696,20 +1698,16 @@ func (s *Storage) updatePerDateData(rows []rawRow) error {
 						accountID: r.TSID.AccountID,
 						projectID: r.TSID.ProjectID,
 					})
-					s.pendingNextDayMetricIDsLock.Lock()
-					s.pendingNextDayMetricIDs.Add(metricID)
-					s.pendingNextDayMetricIDsLock.Unlock()
+					pendingNextDayMetricIDs = append(pendingNextDayMetricIDs, metricID)
 				}
 				continue
 			}
-			s.pendingHourEntriesLock.Lock()
 			e := pendingHourMetricIDEntry{
 				AccountID: r.TSID.AccountID,
 				ProjectID: r.TSID.ProjectID,
 				MetricID:  metricID,
 			}
-			s.pendingHourEntries = append(s.pendingHourEntries, e)
-			s.pendingHourEntriesLock.Unlock()
+			pendingHourEntries = append(pendingHourEntries, e)
 			if date == hmPrevDate && hmPrev.m.Has(metricID) {
 				// The metricID is already registered for the current day on the previous hour.
 				continue
@@ -1736,6 +1734,16 @@ func (s *Storage) updatePerDateData(rows []rawRow) error {
 			})
 		}
 	}
+	if len(pendingNextDayMetricIDs) > 0 {
+		s.pendingNextDayMetricIDsLock.Lock()
+		s.pendingNextDayMetricIDs.AddMulti(pendingNextDayMetricIDs)
+		s.pendingNextDayMetricIDsLock.Unlock()
+	}
+	if len(pendingHourEntries) > 0 {
+		s.pendingHourEntriesLock.Lock()
+		s.pendingHourEntries = append(s.pendingHourEntries, pendingHourEntries...)
+		s.pendingHourEntriesLock.Unlock()
+	}
 	if len(pendingDateMetricIDs) == 0 {
 		// Fast path - there are no new (date, metricID) entires in rows.
 		return nil