Repository: VictoriaMetrics/VictoriaMetrics
Pull request #6976 (2024-09-10 10:07:03)

Allow disabling per-day index
### Describe Your Changes

Allow disabling the per-day index using the `-disablePerDayIndex` flag. This should significantly improve the ingestion rate and decrease the disk space usage for the use cases that assume small or no churn rate. See the docs added to `docs/README.md` for details.

Both improvements are due to no data written to the per-day index. Benchmark results:

```shell
rm -Rf ./lib/storage/Benchmark*; go test ./lib/storage -run=NONE -bench=BenchmarkStorageInsertWithAndWithoutPerDayIndex --loggerLevel=ERROR
goos: linux
goarch: amd64
pkg: github.com/VictoriaMetrics/VictoriaMetrics/lib/storage
cpu: 13th Gen Intel(R) Core(TM) i7-1355U
BenchmarkStorageInsertWithAndWithoutPerDayIndex/HighChurnRate/perDayIndexes-12                 1        3850268120 ns/op                39.56 data-MiB          28.20 indexdb-MiB           259722 rows/s
BenchmarkStorageInsertWithAndWithoutPerDayIndex/HighChurnRate/noPerDayIndexes-12               1        2916865725 ns/op                39.57 data-MiB          25.73 indexdb-MiB           342834 rows/s
BenchmarkStorageInsertWithAndWithoutPerDayIndex/NoChurnRate/perDayIndexes-12                   1        2218073474 ns/op                 9.772 data-MiB         13.73 indexdb-MiB           450842 rows/s
BenchmarkStorageInsertWithAndWithoutPerDayIndex/NoChurnRate/noPerDayIndexes-12                 1        1295140898 ns/op                 9.771 data-MiB          0.3566 indexdb-MiB         772119 rows/s
PASS
ok      github.com/VictoriaMetrics/VictoriaMetrics/lib/storage  11.421s
```

Suggested review order:

-   `docs/README.md` adds the user documentation of this change
-   `app/vmstorage/main.go` introduces the CLI flag for disabling the index
-   `lib/storage/storage.go` accepts new param and, based on its value, either adjusts date or time range to global index value or leaves them as is.
-   `lib/storage/time.go` introduces special values for the date and time range that indicate that the global index search is required.
-   `lib/storage/index_db.go` and `lib/storage/search.go` accept the date or time range from storage and, based on its value, either perform search using per-day or global index.
-   `app/vmselect/netstorage/netstorage.go` renames functions
-   `lib/storage/time_test.go` tests the time and time range changes
-   `lib/storage/storage_test.go` adds tests for storing and retrieving the data when the per-day index is disabled.
-   `lib/storage/storage_timing_test.go` adds a benchmark that compares the data and index size for high and no churn data with the per-day index enabled and disabled.
-   `indexdb_test.go` and `indexdb_timing_test.go` rename functions and add the new parameter to storage which is always false (disabling the per-day index is tested in storage_test.go and storage_timing_test.go)
-   `lib/storage/search_test.go` adds the new param to the storage, also extracts some code into helper functions which are then used in `storage_test.go` for testing the search with per-day index disabled.

### Checklist

The following checks are **mandatory**:

- [x] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).


rtm0: Just in case it might be helpful, here is the diagram with affected functions:

![search](https://github.com/user-attachments/assets/99312abf-6d97-4cbb-9898-663f5708e232)

changshun-shi: releated to https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6584. A cool issue and commit.
hagen1778: Hey @f41gh7 @rtm0! I think we should merge this PR, it sounds very useful for IoT use case.
@rtm0 would you mind rebase it on master? @f41gh7 I'd appreciate your review.
hagen1778: @rtm0, I would appreciate you adding some integration tests as well :)
rtm0: > @rtm0, I would appreciate you adding some integration tests as well :)

Done.
hagen1778: @rtm0 would you mind taking a look at comments above and squash your commits with rebase on master?

Patch:
diff --git a/app/vmselect/netstorage/netstorage.go b/app/vmselect/netstorage/netstorage.go
index 723d8a2cb..da21f4133 100644
--- a/app/vmselect/netstorage/netstorage.go
+++ b/app/vmselect/netstorage/netstorage.go
@@ -791,7 +791,7 @@ func LabelNames(qt *querytracer.Tracer, sq *storage.SearchQuery, maxLabelNames i
 	if err != nil {
 		return nil, err
 	}
-	labels, err := vmstorage.SearchLabelNamesWithFiltersOnTimeRange(qt, tfss, tr, maxLabelNames, sq.MaxMetrics, deadline.Deadline())
+	labels, err := vmstorage.SearchLabelNames(qt, tfss, tr, maxLabelNames, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during labels search on time range: %w", err)
 	}
@@ -864,7 +864,7 @@ func LabelValues(qt *querytracer.Tracer, labelName string, sq *storage.SearchQue
 	if err != nil {
 		return nil, err
 	}
-	labelValues, err := vmstorage.SearchLabelValuesWithFiltersOnTimeRange(qt, labelName, tfss, tr, maxLabelValues, sq.MaxMetrics, deadline.Deadline())
+	labelValues, err := vmstorage.SearchLabelValues(qt, labelName, tfss, tr, maxLabelValues, sq.MaxMetrics, deadline.Deadline())
 	if err != nil {
 		return nil, fmt.Errorf("error during label values search on time range for labelName=%q: %w", labelName, err)
 	}
diff --git a/app/vmstorage/main.go b/app/vmstorage/main.go
index 8dd841646..a5f07b5bf 100644
--- a/app/vmstorage/main.go
+++ b/app/vmstorage/main.go
@@ -71,6 +71,11 @@ var (
 		"See https://docs.victoriametrics.com/single-server-victoriametrics/#cache-tuning")
 	cacheSizeIndexDBTagFilters = flagutil.NewBytes("storage.cacheSizeIndexDBTagFilters", 0, "Overrides max size for indexdb/tagFiltersToMetricIDs cache. "+
 		"See https://docs.victoriametrics.com/single-server-victoriametrics/#cache-tuning")
+
+	disablePerDayIndex = flag.Bool("disablePerDayIndex", false, "Disable per-day index and use global index for all searches. "+
+		"This may improve performance and decrease disk space usage for the use cases with fixed set of timeseries scattered across a "+
+		"big time range (for example, when loading years of historical data). "+
+		"See https://docs.victoriametrics.com/single-server-victoriametrics/#index-tuning")
 )
 
 // CheckTimeRange returns true if the given tr is denied for querying.
@@ -112,9 +117,10 @@ func Init(resetCacheIfNeeded func(mrs []storage.MetricRow)) {
 	WG = syncwg.WaitGroup{}
 
 	opts := storage.OpenOptions{
-		Retention:       retentionPeriod.Duration(),
-		MaxHourlySeries: *maxHourlySeries,
-		MaxDailySeries:  *maxDailySeries,
+		Retention:          retentionPeriod.Duration(),
+		MaxHourlySeries:    *maxHourlySeries,
+		MaxDailySeries:     *maxDailySeries,
+		DisablePerDayIndex: *disablePerDayIndex,
 	}
 	strg := storage.MustOpenStorage(*DataPath, opts)
 	Storage = strg
@@ -195,20 +201,19 @@ func SearchMetricNames(qt *querytracer.Tracer, tfss []*storage.TagFilters, tr st
 	return metricNames, err
 }
 
-// SearchLabelNamesWithFiltersOnTimeRange searches for tag keys matching the given tfss on tr.
-func SearchLabelNamesWithFiltersOnTimeRange(qt *querytracer.Tracer, tfss []*storage.TagFilters, tr storage.TimeRange, maxTagKeys, maxMetrics int, deadline uint64) ([]string, error) {
+// SearchLabelNames searches for tag keys matching the given tfss on tr.
+func SearchLabelNames(qt *querytracer.Tracer, tfss []*storage.TagFilters, tr storage.TimeRange, maxTagKeys, maxMetrics int, deadline uint64) ([]string, error) {
 	WG.Add(1)
-	labelNames, err := Storage.SearchLabelNamesWithFiltersOnTimeRange(qt, tfss, tr, maxTagKeys, maxMetrics, deadline)
+	labelNames, err := Storage.SearchLabelNames(qt, tfss, tr, maxTagKeys, maxMetrics, deadline)
 	WG.Done()
 	return labelNames, err
 }
 
-// SearchLabelValuesWithFiltersOnTimeRange searches for label values for the given labelName, tfss and tr.
-func SearchLabelValuesWithFiltersOnTimeRange(qt *querytracer.Tracer, labelName string, tfss []*storage.TagFilters,
-	tr storage.TimeRange, maxLabelValues, maxMetrics int, deadline uint64,
-) ([]string, error) {
+// SearchLabelValues searches for label values for the given labelName, tfss and
+// tr.
+func SearchLabelValues(qt *querytracer.Tracer, labelName string, tfss []*storage.TagFilters, tr storage.TimeRange, maxLabelValues, maxMetrics int, deadline uint64) ([]string, error) {
 	WG.Add(1)
-	labelValues, err := Storage.SearchLabelValuesWithFiltersOnTimeRange(qt, labelName, tfss, tr, maxLabelValues, maxMetrics, deadline)
+	labelValues, err := Storage.SearchLabelValues(qt, labelName, tfss, tr, maxLabelValues, maxMetrics, deadline)
 	WG.Done()
 	return labelValues, err
 }
diff --git a/docs/README.md b/docs/README.md
index 62cf0b8c2..759d8adfd 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -2005,6 +2005,8 @@ IndexDB respects [retention period](#retention) and once it is over, the indexes
 are dropped. For the new retention period, the indexes are gradually populated
 again as the new samples arrive.
 
+Also see how IndexDB can be [tuned](#index-tuning).
+
 ## Retention
 
 Retention is configured with the `-retentionPeriod` command-line flag, which takes a number followed by a time unit 
@@ -2592,6 +2594,53 @@ and vmstorage has enough free memory to accommodate new cache sizes.
 To override the default values see command-line flags with `-storage.cacheSize` prefix.
 See the full description of flags [here](#list-of-command-line-flags).
 
+## Index tuning for low churn rate
+
+By default, VictoriaMetrics uses the following indexes for data retrieval:
+`global` and `per-day`. Both store the same data and on query time
+VictoriaMetrics can choose between indexes for optimal performance. See
+[IndexDB](#indexdb) for details.
+
+If your use case involves
+[high cardinality](https://docs.victoriametrics.com/faq/#what-is-high-cardinality)
+with
+[high churn rate](https://docs.victoriametrics.com/faq/#what-is-high-churn-rate)
+then this default setting should be ideal for you.
+
+A prominent example is Kubernetes. Services in k8s expose big number of series
+with short life time, significantly increasing churn rate. The per-day index
+speeds up data retrieval in this case.
+
+But if your use case assumes low or no churn rate, then you might benefit
+from disabling the per-day index by setting the flag `-disablePerDayIndex`. This
+will improve the time series ingestion speed and decrease disk space usage,
+since no time or disk space is spent maintaining the per-day index.
+
+Example use cases:
+
+-   Historical weather data, such as
+    [ERA5](https://cds.climate.copernicus.eu/cdsapp#!/dataset/reanalysis-era5-single-levels).
+    It consists of millions time series whose hourly values span tens of years.
+    The time series set never changes. If the per-day index is disabled, once
+    the first hour of data is ingested the entire time series set will be
+    written into the global index and subsequent portions of data will not
+    result in index update. But if the per-day index is enabled, the same set of
+    time-series will be written to the per-day index for every day of data.
+
+-   IoT: a huge set of sensors exports time series with the sensor ID used as a
+    metric label value. Since sensor additions or removals happen infrequently,
+    the time series churn rate will be low. With the per-day index disabled, the
+    entire time series set will be registered in global index during the initial
+    data ingestion and the global index will receive small updates when a sensor
+    is added or removed.
+
+
+What to expect:
+
+-   Prefer setting this flag on fresh installations.
+-   Disabling per-day index on installations with historical data is Ok.
+-   Re-enabling per-day index on installations with historical data will make it unsearchable.
+
 ## Data migration
 
 ### From VictoriaMetrics
diff --git a/lib/storage/index_db.go b/lib/storage/index_db.go
index e60138a26..884b4f8a0 100644
--- a/lib/storage/index_db.go
+++ b/lib/storage/index_db.go
@@ -33,16 +33,19 @@ import (
 const (
 	// Prefix for MetricName->TSID entries.
 	//
-	// This index was substituted with nsPrefixDateMetricNameToTSID,
-	// since the MetricName->TSID index may require big amounts of memory for indexdb/dataBlocks cache
-	// when it grows big on the configured retention under high churn rate
-	// (e.g. when new time series are constantly registered).
+	// This index is used only when -disablePerDayIndex flag is set.
+	//
+	// Otherwise, this index is substituted with nsPrefixDateMetricNameToTSID,
+	// since the MetricName->TSID index may require big amounts of memory for
+	// indexdb/dataBlocks cache when it grows big on the configured retention
+	// under high churn rate (e.g. when new time series are constantly
+	// registered).
 	//
 	// It is much more efficient from memory usage PoV to query per-day MetricName->TSID index
 	// (aka nsPrefixDateMetricNameToTSID) when the TSID must be obtained for the given MetricName
 	// during data ingestion under high churn rate and big retention.
 	//
-	// nsPrefixMetricNameToTSID = 0
+	nsPrefixMetricNameToTSID = 0
 
 	// Prefix for Tag->MetricID entries.
 	nsPrefixTagToMetricIDs = 1
@@ -147,7 +150,7 @@ func getTagFiltersCacheSize() int {
 // will be then used as indexDB.generation
 func mustOpenIndexDB(path string, s *Storage, isReadOnly *atomic.Bool) *indexDB {
 	if s == nil {
-		logger.Panicf("BUG: Storage must be nin-nil")
+		logger.Panicf("BUG: Storage must be non-nil")
 	}
 
 	name := filepath.Base(path)
@@ -420,8 +423,7 @@ func marshalTagFiltersKey(dst []byte, tfss []*TagFilters, tr TimeRange, versione
 		prefix = tagFiltersKeyGen.Load()
 	}
 	// Round start and end times to per-day granularity according to per-day inverted index.
-	startDate := uint64(tr.MinTimestamp) / msecPerDay
-	endDate := uint64(tr.MaxTimestamp-1) / msecPerDay
+	startDate, endDate := tr.DateRange()
 	dst = encoding.MarshalUint64(dst, prefix)
 	dst = encoding.MarshalUint64(dst, startDate)
 	dst = encoding.MarshalUint64(dst, endDate)
@@ -573,6 +575,18 @@ func (is *indexSearch) createGlobalIndexes(tsid *TSID, mn *MetricName) {
 	ii := getIndexItems()
 	defer putIndexItems(ii)
 
+	if is.db.s.disablePerDayIndex {
+		// Create metricName -> TSID entry.
+		// This index is used for searching a TSID by metric name during data
+		// ingestion or metric name registration when -disablePerDayIndex flag
+		// is set.
+		ii.B = marshalCommonPrefix(ii.B, nsPrefixMetricNameToTSID)
+		ii.B = mn.Marshal(ii.B)
+		ii.B = append(ii.B, kvSeparatorChar)
+		ii.B = tsid.Marshal(ii.B)
+		ii.Next()
+	}
+
 	// Create metricID -> metricName entry.
 	ii.B = marshalCommonPrefix(ii.B, nsPrefixMetricIDToMetricName)
 	ii.B = encoding.MarshalUint64(ii.B, tsid.MetricID)
@@ -627,8 +641,9 @@ func putIndexItems(ii *indexItems) {
 
 var indexItemsPool sync.Pool
 
-// SearchLabelNamesWithFiltersOnTimeRange returns all the label names, which match the given tfss on the given tr.
-func (db *indexDB) SearchLabelNamesWithFiltersOnTimeRange(qt *querytracer.Tracer, tfss []*TagFilters, tr TimeRange, maxLabelNames, maxMetrics int, deadline uint64) ([]string, error) {
+// SearchLabelNames returns all the label names, which match the given tfss on
+// the given tr.
+func (db *indexDB) SearchLabelNames(qt *querytracer.Tracer, tfss []*TagFilters, tr TimeRange, maxLabelNames, maxMetrics int, deadline uint64) ([]string, error) {
 	qt = qt.NewChild("search for label names: filters=%s, timeRange=%s, maxLabelNames=%d, maxMetrics=%d", tfss, &tr, maxLabelNames, maxMetrics)
 	defer qt.Done()
 
@@ -664,14 +679,14 @@ func (db *indexDB) SearchLabelNamesWithFiltersOnTimeRange(qt *querytracer.Tracer
 }
 
 func (is *indexSearch) searchLabelNamesWithFiltersOnTimeRange(qt *querytracer.Tracer, lns map[string]struct{}, tfss []*TagFilters, tr TimeRange, maxLabelNames, maxMetrics int) error {
-	minDate := uint64(tr.MinTimestamp) / msecPerDay
-	maxDate := uint64(tr.MaxTimestamp-1) / msecPerDay
-	if maxDate == 0 || minDate > maxDate || maxDate-minDate > maxDaysForPerDaySearch {
+	if tr == globalIndexTimeRange {
 		qtChild := qt.NewChild("search for label names in global index: filters=%s", tfss)
-		err := is.searchLabelNamesWithFiltersOnDate(qtChild, lns, tfss, 0, maxLabelNames, maxMetrics)
+		err := is.searchLabelNamesWithFiltersOnDate(qtChild, lns, tfss, globalIndexDate, maxLabelNames, maxMetrics)
 		qtChild.Done()
 		return err
 	}
+
+	minDate, maxDate := tr.DateRange()
 	var mu sync.Mutex
 	wg := getWaitGroup()
 	var errGlobal error
@@ -734,7 +749,7 @@ func (is *indexSearch) searchLabelNamesWithFiltersOnDate(qt *querytracer.Tracer,
 	loopsPaceLimiter := 0
 	underscoreNameSeen := false
 	nsPrefixExpected := byte(nsPrefixDateTagToMetricIDs)
-	if date == 0 {
+	if date == globalIndexDate {
 		nsPrefixExpected = nsPrefixTagToMetricIDs
 	}
 
@@ -845,8 +860,8 @@ func (is *indexSearch) getLabelNamesForMetricIDs(qt *querytracer.Tracer, metricI
 	qt.Printf("get %d distinct label names from %d metricIDs", foundLabelNames, len(metricIDs))
 }
 
-// SearchLabelValuesWithFiltersOnTimeRange returns label values for the given labelName, tfss and tr.
-func (db *indexDB) SearchLabelValuesWithFiltersOnTimeRange(qt *querytracer.Tracer, labelName string, tfss []*TagFilters, tr TimeRange,
+// SearchLabelValues returns label values for the given labelName, tfss and tr.
+func (db *indexDB) SearchLabelValues(qt *querytracer.Tracer, labelName string, tfss []*TagFilters, tr TimeRange,
 	maxLabelValues, maxMetrics int, deadline uint64) ([]string, error) {
 	qt = qt.NewChild("search for label values: labelName=%q, filters=%s, timeRange=%s, maxLabelNames=%d, maxMetrics=%d", labelName, tfss, &tr, maxLabelValues, maxMetrics)
 	defer qt.Done()
@@ -888,21 +903,21 @@ func (db *indexDB) SearchLabelValuesWithFiltersOnTimeRange(qt *querytracer.Trace
 
 func (is *indexSearch) searchLabelValuesWithFiltersOnTimeRange(qt *querytracer.Tracer, lvs map[string]struct{}, labelName string, tfss []*TagFilters,
 	tr TimeRange, maxLabelValues, maxMetrics int) error {
-	minDate := uint64(tr.MinTimestamp) / msecPerDay
-	maxDate := uint64(tr.MaxTimestamp-1) / msecPerDay
-	if maxDate == 0 || minDate > maxDate || maxDate-minDate > maxDaysForPerDaySearch {
+	if tr == globalIndexTimeRange {
 		qtChild := qt.NewChild("search for label values in global index: labelName=%q, filters=%s", labelName, tfss)
-		err := is.searchLabelValuesWithFiltersOnDate(qtChild, lvs, labelName, tfss, 0, maxLabelValues, maxMetrics)
+		err := is.searchLabelValuesWithFiltersOnDate(qtChild, lvs, labelName, tfss, globalIndexDate, maxLabelValues, maxMetrics)
 		qtChild.Done()
 		return err
 	}
+
+	minDate, maxDate := tr.DateRange()
 	var mu sync.Mutex
 	wg := getWaitGroup()
 	var errGlobal error
 	qt = qt.NewChild("parallel search for label values: labelName=%q, filters=%s, timeRange=%s", labelName, tfss, &tr)
 	for date := minDate; date <= maxDate; date++ {
 		wg.Add(1)
-		qtChild := qt.NewChild("search for label names: filters=%s, date=%s", tfss, dateToString(date))
+		qtChild := qt.NewChild("search for label values: filters=%s, date=%s", tfss, dateToString(date))
 		go func(date uint64) {
 			defer func() {
 				qtChild.Done()
@@ -967,7 +982,7 @@ func (is *indexSearch) searchLabelValuesWithFiltersOnDate(qt *querytracer.Tracer
 	dmis := is.db.s.getDeletedMetricIDs()
 	loopsPaceLimiter := 0
 	nsPrefixExpected := byte(nsPrefixDateTagToMetricIDs)
-	if date == 0 {
+	if date == globalIndexDate {
 		nsPrefixExpected = nsPrefixTagToMetricIDs
 	}
 	kb.B = is.marshalCommonPrefixForDate(kb.B[:0], date)
@@ -1096,11 +1111,11 @@ func (db *indexDB) SearchTagValueSuffixes(qt *querytracer.Tracer, tr TimeRange,
 }
 
 func (is *indexSearch) searchTagValueSuffixesForTimeRange(tvss map[string]struct{}, tr TimeRange, tagKey, tagValuePrefix string, delimiter byte, maxTagValueSuffixes int) error {
-	minDate := uint64(tr.MinTimestamp) / msecPerDay
-	maxDate := uint64(tr.MaxTimestamp-1) / msecPerDay
-	if minDate > maxDate || maxDate-minDate > maxDaysForPerDaySearch {
+	if tr == globalIndexTimeRange {
 		return is.searchTagValueSuffixesAll(tvss, tagKey, tagValuePrefix, delimiter, maxTagValueSuffixes)
 	}
+
+	minDate, maxDate := tr.DateRange()
 	// Query over multiple days in parallel.
 	wg := getWaitGroup()
 	var errGlobal error
@@ -1329,7 +1344,7 @@ func (is *indexSearch) getTSDBStatus(qt *querytracer.Tracer, tfss []*TagFilters,
 
 	loopsPaceLimiter := 0
 	nsPrefixExpected := byte(nsPrefixDateTagToMetricIDs)
-	if date == 0 {
+	if date == globalIndexDate {
 		nsPrefixExpected = nsPrefixTagToMetricIDs
 	}
 	kb.B = is.marshalCommonPrefixForDate(kb.B[:0], date)
@@ -1594,12 +1609,8 @@ func (db *indexDB) DeleteTSIDs(qt *querytracer.Tracer, tfss []*TagFilters, maxMe
 	}
 
 	// Obtain metricIDs to delete.
-	tr := TimeRange{
-		MinTimestamp: 0,
-		MaxTimestamp: (1 << 63) - 1,
-	}
 	is := db.getIndexSearch(noDeadline)
-	metricIDs, err := is.searchMetricIDs(qt, tfss, tr, maxMetrics)
+	metricIDs, err := is.searchMetricIDs(qt, tfss, globalIndexTimeRange, maxMetrics)
 	db.putIndexSearch(is)
 	if err != nil {
 		return 0, err
@@ -1888,8 +1899,13 @@ func (is *indexSearch) getTSIDByMetricNameNoExtDB(dst *TSID, metricName []byte,
 	dmis := is.db.s.getDeletedMetricIDs()
 	ts := &is.ts
 	kb := &is.kb
-	kb.B = marshalCommonPrefix(kb.B[:0], nsPrefixDateMetricNameToTSID)
-	kb.B = encoding.MarshalUint64(kb.B, date)
+
+	if is.db.s.disablePerDayIndex {
+		kb.B = marshalCommonPrefix(kb.B[:0], nsPrefixMetricNameToTSID)
+	} else {
+		kb.B = marshalCommonPrefix(kb.B[:0], nsPrefixDateMetricNameToTSID)
+		kb.B = encoding.MarshalUint64(kb.B, date)
+	}
 	kb.B = append(kb.B, metricName...)
 	kb.B = append(kb.B, kvSeparatorChar)
 	ts.Seek(kb.B)
@@ -1953,8 +1969,11 @@ func (is *indexSearch) searchMetricName(dst []byte, metricID uint64) ([]byte, bo
 }
 
 func (is *indexSearch) containsTimeRange(tr TimeRange) bool {
-	db := is.db
+	if tr == globalIndexTimeRange {
+		return true
+	}
 
+	db := is.db
 	if db.hasExtDB() {
 		// The db corresponds to the current indexDB, which is used for storing index data for newly registered time series.
 		// This means that it may contain data for the given tr with probability close to 100%.
@@ -2247,14 +2266,17 @@ func (is *indexSearch) searchMetricIDsWithFiltersOnDate(qt *querytracer.Tracer,
 	if len(tfss) == 0 {
 		return nil, nil
 	}
-	tr := TimeRange{
-		MinTimestamp: int64(date) * msecPerDay,
-		MaxTimestamp: int64(date+1)*msecPerDay - 1,
-	}
-	if date == 0 {
-		// Search for metricIDs on the whole time range.
-		tr.MaxTimestamp = timestampFromTime(time.Now())
+
+	var tr TimeRange
+	if date == globalIndexDate {
+		tr = globalIndexTimeRange
+	} else {
+		tr = TimeRange{
+			MinTimestamp: int64(date) * msecPerDay,
+			MaxTimestamp: int64(date+1)*msecPerDay - 1,
+		}
 	}
+
 	metricIDs, err := is.searchMetricIDsInternal(qt, tfss, tr, maxMetrics)
 	if err != nil {
 		return nil, err
@@ -2307,6 +2329,8 @@ func (is *indexSearch) searchMetricIDsInternal(qt *querytracer.Tracer, tfss []*T
 
 	metricIDs := &uint64set.Set{}
 
+	// Always returns (true, nil) for zero time range used to indicate global
+	// index search.
 	if !is.containsTimeRange(tr) {
 		qt.Printf("indexdb doesn't contain data for the given timeRange=%s", &tr)
 		return metricIDs, nil
@@ -2339,23 +2363,20 @@ func (is *indexSearch) searchMetricIDsInternal(qt *querytracer.Tracer, tfss []*T
 	return metricIDs, nil
 }
 
-const maxDaysForPerDaySearch = 40
-
 func (is *indexSearch) updateMetricIDsForTagFilters(qt *querytracer.Tracer, metricIDs *uint64set.Set, tfs *TagFilters, tr TimeRange, maxMetrics int) error {
-	minDate := uint64(tr.MinTimestamp) / msecPerDay
-	maxDate := uint64(tr.MaxTimestamp-1) / msecPerDay
-	if minDate <= maxDate && maxDate-minDate <= maxDaysForPerDaySearch {
+	if tr != globalIndexTimeRange {
 		// Fast path - search metricIDs by date range in the per-day inverted
 		// index.
-		is.db.dateRangeSearchCalls.Add(1)
 		qt.Printf("search metric ids in the per-day index")
+		is.db.dateRangeSearchCalls.Add(1)
+		minDate, maxDate := tr.DateRange()
 		return is.updateMetricIDsForDateRange(qt, metricIDs, tfs, minDate, maxDate, maxMetrics)
 	}
 
 	// Slow path - search metricIDs in the global inverted index.
 	qt.Printf("search metric ids in the global index")
 	is.db.globalSearchCalls.Add(1)
-	m, err := is.getMetricIDsForDateAndFilters(qt, 0, tfs, maxMetrics)
+	m, err := is.getMetricIDsForDateAndFilters(qt, globalIndexDate, tfs, maxMetrics)
 	if err != nil {
 		return err
 	}
@@ -2594,6 +2615,7 @@ func (is *indexSearch) getMetricIDsForDateAndFilters(qt *querytracer.Tracer, dat
 		qt = qt.NewChild("search for metric ids on a particular day: filters=%s, date=%s, maxMetrics=%d", tfs, dateToString(date), maxMetrics)
 		defer qt.Done()
 	}
+
 	// Sort tfs by loopsCount needed for performing each filter.
 	// This stats is usually collected from the previous queries.
 	// This way we limit the amount of work below by applying fast filters at first.
@@ -2801,6 +2823,9 @@ const (
 )
 
 func (is *indexSearch) createPerDayIndexes(date uint64, tsid *TSID, mn *MetricName) {
+	if is.db.s.disablePerDayIndex {
+		return
+	}
 	ii := getIndexItems()
 	defer putIndexItems(ii)
 
@@ -2936,6 +2961,10 @@ func reverseBytes(dst, src []byte) []byte {
 }
 
 func (is *indexSearch) hasDateMetricIDNoExtDB(date, metricID uint64) bool {
+	if date == globalIndexDate {
+		return is.hasMetricIDNoExtDB(metricID)
+	}
+
 	ts := &is.ts
 	kb := &is.kb
 	kb.B = marshalCommonPrefix(kb.B[:0], nsPrefixDateToMetricID)
@@ -2955,12 +2984,27 @@ func (is *indexSearch) hasDateMetricIDNoExtDB(date, metricID uint64) bool {
 	return false
 }
 
+func (is *indexSearch) hasMetricIDNoExtDB(metricID uint64) bool {
+	ts := &is.ts
+	kb := &is.kb
+	kb.B = marshalCommonPrefix(kb.B[:0], nsPrefixMetricIDToTSID)
+	kb.B = encoding.MarshalUint64(kb.B, metricID)
+	if err := ts.FirstItemWithPrefix(kb.B); err != nil {
+		if err == io.EOF {
+			return false
+		}
+		logger.Panicf("FATAL: error when for metricID=%d; searchPrefix %q: %s", metricID, kb.B, err)
+	}
+	return true
+}
+
 func (is *indexSearch) getMetricIDsForDateTagFilter(qt *querytracer.Tracer, tf *tagFilter, date uint64, commonPrefix []byte,
 	maxMetrics int, maxLoopsCount int64) (*uint64set.Set, int64, error) {
 	if qt.Enabled() {
 		qt = qt.NewChild("get metric ids for filter and date: filter={%s}, date=%s, maxMetrics=%d, maxLoopsCount=%d", tf, dateToString(date), maxMetrics, maxLoopsCount)
 		defer qt.Done()
 	}
+
 	if !bytes.HasPrefix(tf.prefix, commonPrefix) {
 		logger.Panicf("BUG: unexpected tf.prefix %q; must start with commonPrefix %q", tf.prefix, commonPrefix)
 	}
@@ -3115,7 +3159,7 @@ func (is *indexSearch) marshalCommonPrefix(dst []byte, nsPrefix byte) []byte {
 }
 
 func (is *indexSearch) marshalCommonPrefixForDate(dst []byte, date uint64) []byte {
-	if date == 0 {
+	if date == globalIndexDate {
 		// Global index
 		return is.marshalCommonPrefix(dst, nsPrefixTagToMetricIDs)
 	}
diff --git a/lib/storage/search.go b/lib/storage/search.go
index 2ae10d7ec..1b4af28d2 100644
--- a/lib/storage/search.go
+++ b/lib/storage/search.go
@@ -144,6 +144,10 @@ func (s *Search) reset() {
 func (s *Search) Init(qt *querytracer.Tracer, storage *Storage, tfss []*TagFilters, tr TimeRange, maxMetrics int, deadline uint64) int {
 	qt = qt.NewChild("init series search: filters=%s, timeRange=%s", tfss, &tr)
 	defer qt.Done()
+
+	indexTR := storage.adjustTimeRange(tr)
+	dataTR := tr
+
 	if s.needClosing {
 		logger.Panicf("BUG: missing MustClose call before the next call to Init")
 	}
@@ -158,7 +162,7 @@ func (s *Search) Init(qt *querytracer.Tracer, storage *Storage, tfss []*TagFilte
 	s.needClosing = true
 
 	var tsids []TSID
-	metricIDs, err := s.idb.searchMetricIDs(qt, tfss, tr, maxMetrics, deadline)
+	metricIDs, err := s.idb.searchMetricIDs(qt, tfss, indexTR, maxMetrics, deadline)
 	if err == nil {
 		tsids, err = s.idb.getTSIDsFromMetricIDs(qt, metricIDs, deadline)
 		if err == nil {
@@ -168,7 +172,7 @@ func (s *Search) Init(qt *querytracer.Tracer, storage *Storage, tfss []*TagFilte
 	// It is ok to call Init on non-nil err.
 	// Init must be called before returning because it will fail
 	// on Search.MustClose otherwise.
-	s.ts.Init(storage.tb, tsids, tr)
+	s.ts.Init(storage.tb, tsids, dataTR)
 	qt.Printf("search for parts with data for %d series", len(tsids))
 	if err != nil {
 		s.err = err
diff --git a/lib/storage/storage.go b/lib/storage/storage.go
index b775e6508..70e5612ba 100644
--- a/lib/storage/storage.go
+++ b/lib/storage/storage.go
@@ -82,6 +82,8 @@ type Storage struct {
 	// See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1401
 	idbNext atomic.Pointer[indexDB]
 
+	disablePerDayIndex bool
+
 	tb *table
 
 	// Series cardinality limiters.
@@ -167,9 +169,10 @@ type Storage struct {
 
 // OpenOptions optional args for MustOpenStorage
 type OpenOptions struct {
-	Retention       time.Duration
-	MaxHourlySeries int
-	MaxDailySeries  int
+	Retention          time.Duration
+	MaxHourlySeries    int
+	MaxDailySeries     int
+	DisablePerDayIndex bool
 }
 
 // MustOpenStorage opens storage on the given path with the given retentionMsecs.
@@ -248,6 +251,8 @@ func MustOpenStorage(path string, opts OpenOptions) *Storage {
 	fs.MustMkdirIfNotExist(metadataDir)
 	s.minTimestampForCompositeIndex = mustGetMinTimestampForCompositeIndex(metadataDir, isEmptyDB)
 
+	s.disablePerDayIndex = opts.DisablePerDayIndex
+
 	// Load indexdb
 	idbPath := filepath.Join(path, indexdbDirname)
 	idbSnapshotsPath := filepath.Join(idbPath, snapshotsDirname)
@@ -1139,10 +1144,21 @@ func nextRetentionDeadlineSeconds(atSecs, retentionSecs, offsetSecs int64) int64
 	return deadline
 }
 
-// SearchMetricNames returns marshaled metric names matching the given tfss on the given tr.
+// SearchMetricNames returns marshaled metric names matching the given tfss on
+// the given tr.
+//
+// The marshaled metric names must be unmarshaled via
+// MetricName.UnmarshalString().
 //
-// The marshaled metric names must be unmarshaled via MetricName.UnmarshalString().
+// If -disablePerDayIndex flag is not set, the metric names are searched
+// within the given time range (as long as the time range is no more than 40
+// days), i.e. the per-day index are used for searching.
+//
+// If -disablePerDayIndex is set or the time range is more than 40 days, the
+// time range is ignored and the metrics are searched within the entire
+// retention period, i.e. the global index are used for searching.
 func (s *Storage) SearchMetricNames(qt *querytracer.Tracer, tfss []*TagFilters, tr TimeRange, maxMetrics int, deadline uint64) ([]string, error) {
+	tr = s.adjustTimeRange(tr)
 	qt = qt.NewChild("search for matching metric names: filters=%s, timeRange=%s", tfss, &tr)
 	defer qt.Done()
 
@@ -1291,16 +1307,33 @@ func (s *Storage) DeleteSeries(qt *querytracer.Tracer, tfss []*TagFilters, maxMe
 	return deletedCount, nil
 }
 
-// SearchLabelNamesWithFiltersOnTimeRange searches for label names matching the given tfss on tr.
-func (s *Storage) SearchLabelNamesWithFiltersOnTimeRange(qt *querytracer.Tracer, tfss []*TagFilters, tr TimeRange, maxLabelNames, maxMetrics int, deadline uint64,
-) ([]string, error) {
-	return s.idb().SearchLabelNamesWithFiltersOnTimeRange(qt, tfss, tr, maxLabelNames, maxMetrics, deadline)
+// SearchLabelNames searches for label names matching the given tfss on tr.
+//
+// If -disablePerDayIndex flag is not set, the label names are searched
+// within the given time range (as long as the time range is no more than 40
+// days), i.e. the per-day index are used for searching.
+//
+// If -disablePerDayIndex is set or the time range is more than 40 days, the
+// time range is ignored and the label names are searched within the entire
+// retention period, i.e. the global index are used for searching.
+func (s *Storage) SearchLabelNames(qt *querytracer.Tracer, tfss []*TagFilters, tr TimeRange, maxLabelNames, maxMetrics int, deadline uint64) ([]string, error) {
+	tr = s.adjustTimeRange(tr)
+	return s.idb().SearchLabelNames(qt, tfss, tr, maxLabelNames, maxMetrics, deadline)
 }
 
-// SearchLabelValuesWithFiltersOnTimeRange searches for label values for the given labelName, filters and tr.
-func (s *Storage) SearchLabelValuesWithFiltersOnTimeRange(qt *querytracer.Tracer, labelName string, tfss []*TagFilters,
-	tr TimeRange, maxLabelValues, maxMetrics int, deadline uint64,
-) ([]string, error) {
+// SearchLabelValues searches for label values for the given labelName, filters
+// and tr.
+//
+// If -disablePerDayIndex flag is not set, the label values are searched
+// within the given time range (as long as the time range is no more than 40
+// days), i.e. the per-day index are used for searching.
+//
+// If -disablePerDayIndex is set or the time range is more than 40 days, the
+// time range is ignored and the label values are searched within the entire
+// retention period, i.e. the global index are used for searching.
+func (s *Storage) SearchLabelValues(qt *querytracer.Tracer, labelName string, tfss []*TagFilters, tr TimeRange, maxLabelValues, maxMetrics int, deadline uint64) ([]string, error) {
+	tr = s.adjustTimeRange(tr)
+
 	idb := s.idb()
 
 	key := labelName
@@ -1311,7 +1344,8 @@ func (s *Storage) SearchLabelValuesWithFiltersOnTimeRange(qt *querytracer.Tracer
 		// tfss contains only a single filter on labelName. It is faster searching for label values
 		// without any filters and limits and then later applying the filter and the limit to the found label values.
 		qt.Printf("search for up to %d values for the label %q on the time range %s", maxMetrics, labelName, &tr)
-		lvs, err := idb.SearchLabelValuesWithFiltersOnTimeRange(qt, labelName, nil, tr, maxMetrics, maxMetrics, deadline)
+
+		lvs, err := idb.SearchLabelValues(qt, labelName, nil, tr, maxMetrics, maxMetrics, deadline)
 		if err != nil {
 			return nil, err
 		}
@@ -1334,7 +1368,7 @@ func (s *Storage) SearchLabelValuesWithFiltersOnTimeRange(qt *querytracer.Tracer
 		qt.Printf("fall back to slow search because only a subset of label values is found")
 	}
 
-	return idb.SearchLabelValuesWithFiltersOnTimeRange(qt, labelName, tfss, tr, maxLabelValues, maxMetrics, deadline)
+	return idb.SearchLabelValues(qt, labelName, tfss, tr, maxLabelValues, maxMetrics, deadline)
 }
 
 func filterLabelValues(lvs []string, tf *tagFilter, key string) []string {
@@ -1355,19 +1389,42 @@ func filterLabelValues(lvs []string, tf *tagFilter, key string) []string {
 	return result
 }
 
-// SearchTagValueSuffixes returns all the tag value suffixes for the given tagKey and tagValuePrefix on the given tr.
+// SearchTagValueSuffixes returns all the tag value suffixes for the given
+// tagKey and tagValuePrefix on the given tr.
+//
+// This allows implementing
+// https://graphite-api.readthedocs.io/en/latest/api.html#metrics-find
+// or similar APIs.
+//
+// If more than maxTagValueSuffixes suffixes is found, then only the first
+// maxTagValueSuffixes suffixes is returned.
 //
-// This allows implementing https://graphite-api.readthedocs.io/en/latest/api.html#metrics-find or similar APIs.
+// If -disablePerDayIndex flag is not set, the tag value suffixes are searched
+// within the given time range (as long as the time range is no more than 40
+// days), i.e. the per-day index are used for searching.
 //
-// If more than maxTagValueSuffixes suffixes is found, then only the first maxTagValueSuffixes suffixes is returned.
+// If -disablePerDayIndex is set or the time range is more than 40 days, the
+// time range is ignored and the tag value suffixes are searched within the
+// entire retention period, i.e. the global index are used for searching.
 func (s *Storage) SearchTagValueSuffixes(qt *querytracer.Tracer, tr TimeRange, tagKey, tagValuePrefix string,
 	delimiter byte, maxTagValueSuffixes int, deadline uint64,
 ) ([]string, error) {
+	tr = s.adjustTimeRange(tr)
 	return s.idb().SearchTagValueSuffixes(qt, tr, tagKey, tagValuePrefix, delimiter, maxTagValueSuffixes, deadline)
 }
 
-// SearchGraphitePaths returns all the matching paths for the given graphite query on the given tr.
+// SearchGraphitePaths returns all the matching paths for the given graphite
+// query on the given tr.
+//
+// If -disablePerDayIndex flag is not set, the graphite paths are searched
+// within the given time range (as long as the time range is no more than 40
+// days), i.e. the per-day index are used for searching.
+//
+// If -disablePerDayIndex is set or the time range is more than 40 days, the
+// time range is ignored and the graphite paths are searched within the entire
+// retention period, i.e. global index are used for searching.
 func (s *Storage) SearchGraphitePaths(qt *querytracer.Tracer, tr TimeRange, query []byte, maxPaths int, deadline uint64) ([]string, error) {
+	tr = s.adjustTimeRange(tr)
 	query = replaceAlternateRegexpsWithGraphiteWildcards(query)
 	return s.searchGraphitePaths(qt, tr, nil, query, maxPaths, deadline)
 }
@@ -1551,7 +1608,16 @@ func (s *Storage) GetSeriesCount(deadline uint64) (uint64, error) {
 }
 
 // GetTSDBStatus returns TSDB status data for /api/v1/status/tsdb
+//
+// If -disablePerDayIndex flag is not set, the status is calculated for the
+// given date, i.e. the per-day index are used for calculation.
+//
+// Otherwise, the date is ignored and the status is calculated for the entire
+// retention period, i.e. the global index are used for calculation.
 func (s *Storage) GetTSDBStatus(qt *querytracer.Tracer, tfss []*TagFilters, date uint64, focusLabel string, topN, maxMetrics int, deadline uint64) (*TSDBStatus, error) {
+	if s.disablePerDayIndex {
+		date = globalIndexDate
+	}
 	return s.idb().GetTSDBStatus(qt, tfss, date, focusLabel, topN, maxMetrics, deadline)
 }
 
@@ -1684,6 +1750,35 @@ var metricRowsInsertCtxPool sync.Pool
 
 const maxMetricRowsPerBlock = 8000
 
+func (s *Storage) date(millis int64) uint64 {
+	if s.disablePerDayIndex {
+		return globalIndexDate
+	}
+	return uint64(millis) / msecPerDay
+}
+
+// It has been found empirically, that once the time range is bigger than 40
+// days searching using per-day index becomes slower than using global index.
+//
+// TODO(rtm0): Extract into a flag?
+const maxDaysForPerDaySearch = 40
+
+// adjustTimeRange decides whether to use the time range as is or use
+// globalIndexTimeRange based on the time range length and -disablePerDayIndex
+// flag.
+func (s *Storage) adjustTimeRange(tr TimeRange) TimeRange {
+	if s.disablePerDayIndex {
+		return globalIndexTimeRange
+	}
+
+	minDate, maxDate := tr.DateRange()
+	if maxDate-minDate > maxDaysForPerDaySearch {
+		return globalIndexTimeRange
+	}
+
+	return tr
+}
+
 // RegisterMetricNames registers all the metric names from mrs in the indexdb, so they can be queried later.
 //
 // The the MetricRow.Timestamp is used for registering the metric name at the given day according to the timestamp.
@@ -1706,7 +1801,7 @@ func (s *Storage) RegisterMetricNames(qt *querytracer.Tracer, mrs []MetricRow) {
 	var firstWarn error
 	for i := range mrs {
 		mr := &mrs[i]
-		date := uint64(mr.Timestamp) / msecPerDay
+		date := s.date(mr.Timestamp)
 		if s.getTSIDFromCache(&genTSID, mr.MetricNameRaw) {
 			// Fast path - mr.MetricNameRaw has been already registered in the current idb.
 			if !s.registerSeriesCardinality(genTSID.TSID.MetricID, mr.MetricNameRaw) {
@@ -1873,6 +1968,7 @@ func (s *Storage) add(rows []rawRow, dstMrs []*MetricRow, mrs []MetricRow, preci
 		r.Timestamp = mr.Timestamp
 		r.Value = mr.Value
 		r.PrecisionBits = precisionBits
+		date := s.date(r.Timestamp)
 
 		// Search for TSID for the given mr.MetricNameRaw and store it at r.TSID.
 		if string(mr.MetricNameRaw) == string(prevMetricNameRaw) {
@@ -1898,8 +1994,6 @@ func (s *Storage) add(rows []rawRow, dstMrs []*MetricRow, mrs []MetricRow, preci
 
 			if genTSID.generation < generation {
 				// The found TSID is from the previous indexdb. Create it in the current indexdb.
-				date := uint64(r.Timestamp) / msecPerDay
-
 				if err := mn.UnmarshalRaw(mr.MetricNameRaw); err != nil {
 					if firstWarn == nil {
 						firstWarn = fmt.Errorf("cannot unmarshal MetricNameRaw %q: %w", mr.MetricNameRaw, err)
@@ -1922,8 +2016,6 @@ func (s *Storage) add(rows []rawRow, dstMrs []*MetricRow, mrs []MetricRow, preci
 		// Slow path - the TSID for the given mr.MetricNameRaw is missing in the cache.
 		slowInsertsCount++
 
-		date := uint64(r.Timestamp) / msecPerDay
-
 		// Construct canonical metric name - it is used below.
 		if err := mn.UnmarshalRaw(mr.MetricNameRaw); err != nil {
 			if firstWarn == nil {
@@ -2105,7 +2197,7 @@ func (s *Storage) prefillNextIndexDB(rows []rawRow, mrs []*MetricRow) error {
 		}
 
 		// Check whether the given MetricID is already present in dateMetricIDCache.
-		date := uint64(r.Timestamp) / msecPerDay
+		date := s.date(r.Timestamp)
 		metricID := r.TSID.MetricID
 		if s.dateMetricIDCache.Has(generation, date, metricID) {
 			// Indexes are already pre-filled.
@@ -2145,6 +2237,10 @@ func (s *Storage) prefillNextIndexDB(rows []rawRow, mrs []*MetricRow) error {
 }
 
 func (s *Storage) updatePerDateData(rows []rawRow, mrs []*MetricRow) error {
+	if s.disablePerDayIndex {
+		return nil
+	}
+
 	var date uint64
 	var hour uint64
 	var prevTimestamp int64
@@ -2267,7 +2363,7 @@ func (s *Storage) updatePerDateData(rows []rawRow, mrs []*MetricRow) error {
 		date := dmid.date
 		metricID := dmid.tsid.MetricID
 		if !is.hasDateMetricIDNoExtDB(date, metricID) {
-			// The (date, metricID) entry is missing in the indexDB. Add it there together with per-day indexes.
+			// The (date, metricID) entry is missing in the indexDB. Add it there together with per-day index.
 			// It is OK if the (date, metricID) entry is added multiple times to indexdb
 			// by concurrent goroutines.
 			if err := mn.UnmarshalRaw(dmid.mr.MetricNameRaw); err != nil {
diff --git a/lib/storage/time.go b/lib/storage/time.go
index e3d013ad3..78ca162cc 100644
--- a/lib/storage/time.go
+++ b/lib/storage/time.go
@@ -5,9 +5,10 @@ import (
 	"time"
 )
 
+// dateToString returns human readable representation of the date.
 func dateToString(date uint64) string {
 	if date == 0 {
-		return "1970-01-01"
+		return "[entire retention period]"
 	}
 	t := time.Unix(int64(date*24*3600), 0).UTC()
 	return t.Format("2006-01-02")
@@ -33,7 +34,34 @@ type TimeRange struct {
 	MaxTimestamp int64
 }
 
+// Zero time range and zero date are used to force global index search.
+var (
+	globalIndexTimeRange = TimeRange{}
+	globalIndexDate      = uint64(0)
+)
+
+// DateRange returns the date range for the given time range.
+func (tr *TimeRange) DateRange() (uint64, uint64) {
+	minDate := uint64(tr.MinTimestamp) / msecPerDay
+	// Max timestamp may point to the first millisecond of the next day. As the
+	// result, the returned date range will cover one more day than needed.
+	// Decrementing by 1 removes this extra day.
+	maxDate := uint64(tr.MaxTimestamp-1) / msecPerDay
+
+	// However, if both timestamps are the same and point to the beginning of
+	// the day, then maxDate will be smaller that the minDate. In this case
+	// maxDate is set to minDate.
+	if maxDate < minDate {
+		maxDate = minDate
+	}
+
+	return minDate, maxDate
+}
+
 func (tr *TimeRange) String() string {
+	if *tr == globalIndexTimeRange {
+		return "[entire retention period]"
+	}
 	start := TimestampToHumanReadableFormat(tr.MinTimestamp)
 	end := TimestampToHumanReadableFormat(tr.MaxTimestamp)
 	return fmt.Sprintf("[%s..%s]", start, end)


Test patch:
diff --git a/apptest/testcase.go b/apptest/testcase.go
index d1aea7d3c..3f111c5f4 100644
--- a/apptest/testcase.go
+++ b/apptest/testcase.go
@@ -233,6 +233,23 @@ func (tc *TestCase) StopApp(instance string) {
 	}
 }
 
+// StopPrometheusWriteQuerier stop all apps that are a part of the pwq.
+func (tc *TestCase) StopPrometheusWriteQuerier(pwq PrometheusWriteQuerier) {
+	tc.t.Helper()
+	switch t := pwq.(type) {
+	case *Vmsingle:
+		tc.StopApp(t.Name())
+	case *vmcluster:
+		tc.StopApp(t.Vminsert.Name())
+		tc.StopApp(t.Vmselect.Name())
+		for _, vmstorage := range t.vmstorages {
+			tc.StopApp(vmstorage.Name())
+		}
+	default:
+		tc.t.Fatalf("Unsupported type: %v", t)
+	}
+}
+
 // ForceFlush flushes zero or more storages.
 func (tc *TestCase) ForceFlush(apps ...*Vmstorage) {
 	tc.t.Helper()
diff --git a/apptest/tests/per_day_index_test.go b/apptest/tests/per_day_index_test.go
new file mode 100644
index 000000000..72444a913
--- /dev/null
+++ b/apptest/tests/per_day_index_test.go
@@ -0,0 +1,220 @@
+package tests
+
+import (
+	"fmt"
+	"testing"
+
+	at "github.com/VictoriaMetrics/VictoriaMetrics/apptest"
+)
+
+func TestSingleSearchWithDisabledPerDayIndex(t *testing.T) {
+	tc := at.NewTestCase(t)
+	defer tc.Stop()
+
+	testSearchWithDisabledPerDayIndex(tc, func(name string, disablePerDayIndex bool) at.PrometheusWriteQuerier {
+		return tc.MustStartVmsingle("vmsingle-"+name, []string{
+			"-storageDataPath=" + tc.Dir() + "/vmsingle",
+			"-retentionPeriod=100y",
+			"-search.maxStalenessInterval=1m",
+			fmt.Sprintf("-disablePerDayIndex=%t", disablePerDayIndex),
+		})
+	})
+}
+
+func TestClusterSearchWithDisabledPerDayIndex(t *testing.T) {
+	tc := at.NewTestCase(t)
+	defer tc.Stop()
+
+	testSearchWithDisabledPerDayIndex(tc, func(name string, disablePerDayIndex bool) at.PrometheusWriteQuerier {
+		return tc.MustStartCluster(&at.ClusterOptions{
+			Vmstorage1Instance: "vmstorage1-" + name,
+			Vmstorage1Flags: []string{
+				fmt.Sprintf("-disablePerDayIndex=%t", disablePerDayIndex),
+			},
+			Vmstorage2Instance: "vmstorage2-" + name,
+			Vmstorage2Flags: []string{
+				fmt.Sprintf("-disablePerDayIndex=%t", disablePerDayIndex),
+			},
+			VminsertInstance: "vminsert",
+			VmselectInstance: "vmselect",
+		})
+	})
+}
+
+type startSUTFunc func(name string, disablePerDayIndex bool) at.PrometheusWriteQuerier
+
+// testDisablePerDayIndex_Search shows what search results to expect when data
+// is first inserted with per-day index enabled and then with per-day index
+// disabled.
+//
+// The data inserted with enabled per-day index must be searchable with disabled
+// per-day index.
+//
+// The data inserted with disabled per-day index is not searcheable with per-day
+// index enabled unless the search time range is > 40 days.
+func testSearchWithDisabledPerDayIndex(tc *at.TestCase, start startSUTFunc) {
+	t := tc.T()
+
+	type opts struct {
+		start, end       string
+		wantSeries       []map[string]string
+		wantQueryResults []*at.QueryResult
+	}
+	assertSearchResults := func(sut at.PrometheusQuerier, opts *opts) {
+		t.Helper()
+		tc.Assert(&at.AssertOptions{
+			Msg: "unexpected /api/v1/series response",
+			Got: func() any {
+				return sut.PrometheusAPIV1Series(t, `{__name__=~".*"}`, at.QueryOpts{
+					Start: opts.start,
+					End:   opts.end,
+				}).Sort()
+			},
+			Want: &at.PrometheusAPIV1SeriesResponse{
+				Status: "success",
+				Data:   opts.wantSeries,
+			},
+		})
+		tc.Assert(&at.AssertOptions{
+			Msg: "unexpected /api/v1/query_range response",
+			Got: func() any {
+				return sut.PrometheusAPIV1QueryRange(t, `{__name__=~".*"}`, at.QueryOpts{
+					Start: opts.start,
+					End:   opts.end,
+					Step:  "1d",
+				})
+			},
+			Want: &at.PrometheusAPIV1QueryResponse{
+				Status: "success",
+				Data: &at.QueryData{
+					ResultType: "matrix",
+					Result:     opts.wantQueryResults,
+				},
+			},
+		})
+	}
+
+	// Start vmsingle with enabled per-day index, insert sample1, and confirm it
+	// is searcheable.
+	sut := start("with-per-day-index", false)
+	sample1 := []string{"metric1 111 1704067200000"} // 2024-01-01T00:00:00Z
+	sut.PrometheusAPIV1ImportPrometheus(t, sample1, at.QueryOpts{})
+	sut.ForceFlush(t)
+	assertSearchResults(sut, &opts{
+		start:      "2024-01-01T00:00:00Z",
+		end:        "2024-01-01T23:59:59Z",
+		wantSeries: []map[string]string{{"__name__": "metric1"}},
+		wantQueryResults: []*at.QueryResult{
+			{
+				Metric:  map[string]string{"__name__": "metric1"},
+				Samples: []*at.Sample{{Timestamp: 1704067200000, Value: float64(111)}},
+			},
+		},
+	})
+
+	// Restart vmsingle with disabled per-day index, insert sample2, and confirm
+	// that both sample1 and sample2 is searcheable.
+	tc.StopPrometheusWriteQuerier(sut)
+	sut = start("without-per-day-index", true)
+	sample2 := []string{"metric2 222 1704067200000"} // 2024-01-01T00:00:00Z
+	sut.PrometheusAPIV1ImportPrometheus(t, sample2, at.QueryOpts{})
+	sut.ForceFlush(t)
+	assertSearchResults(sut, &opts{
+		start: "2024-01-01T00:00:00Z",
+		end:   "2024-01-01T23:59:59Z",
+		wantSeries: []map[string]string{
+			{"__name__": "metric1"},
+			{"__name__": "metric2"},
+		},
+		wantQueryResults: []*at.QueryResult{
+			{
+				Metric:  map[string]string{"__name__": "metric1"},
+				Samples: []*at.Sample{{Timestamp: 1704067200000, Value: float64(111)}},
+			},
+			{
+				Metric:  map[string]string{"__name__": "metric2"},
+				Samples: []*at.Sample{{Timestamp: 1704067200000, Value: float64(222)}},
+			},
+		},
+	})
+
+	// Insert sample1 but for a different date, restart vmsingle with enabled
+	// per-day index and confirm that:
+	// - sample1 is searcheable within the time range of Jan 1st
+	// - sample1 is not searcheable within the time range of Jan 20th
+	// - sample1 is searcheable within the time range of Jan 1st-20th (because
+	//   the metric1 metricID will be found in the per-day index for Jan 1st).
+	// - sample2 is not searcheable when the time range is <= 40 days
+	// - sample2 becomes searcheable when the time range is > 40 days
+	sample3 := []string{"metric1 333 1705708800000"} // 2024-01-20T00:00:00Z
+	sut.PrometheusAPIV1ImportPrometheus(t, sample3, at.QueryOpts{})
+	sut.ForceFlush(t)
+	tc.StopPrometheusWriteQuerier(sut)
+	sut = start("with-per-day-index2", false)
+
+	// Time range is 1 day (Jan 1st) <= 40 days
+	assertSearchResults(sut, &opts{
+		start: "2024-01-01T00:00:00Z",
+		end:   "2024-01-01T23:59:59Z",
+		wantSeries: []map[string]string{
+			{"__name__": "metric1"},
+		},
+		wantQueryResults: []*at.QueryResult{
+			{
+				Metric:  map[string]string{"__name__": "metric1"},
+				Samples: []*at.Sample{{Timestamp: 1704067200000, Value: float64(111)}},
+			},
+		},
+	})
+
+	// Time range is 1 day (Jan 20th) <= 40 days
+	assertSearchResults(sut, &opts{
+		start:            "2024-01-20T00:00:00Z",
+		end:              "2024-01-20T23:59:59Z",
+		wantSeries:       []map[string]string{},
+		wantQueryResults: []*at.QueryResult{},
+	})
+
+	// Time range is 20 days (Jan 1st-20th) <= 40 days
+	assertSearchResults(sut, &opts{
+		start: "2024-01-01T00:00:00Z",
+		end:   "2024-01-20T23:59:59Z",
+		wantSeries: []map[string]string{
+			{"__name__": "metric1"},
+		},
+		wantQueryResults: []*at.QueryResult{
+			{
+				Metric: map[string]string{"__name__": "metric1"},
+				Samples: []*at.Sample{
+					{Timestamp: 1704067200000, Value: float64(111)},
+					{Timestamp: 1705708800000, Value: float64(333)},
+				},
+			},
+		},
+	})
+
+	// Time range > 40 days
+	assertSearchResults(sut, &opts{
+		start: "2024-01-01T00:00:00Z",
+		end:   "2024-02-29T23:59:59Z",
+		wantSeries: []map[string]string{
+			{"__name__": "metric1"},
+			{"__name__": "metric2"},
+		},
+		wantQueryResults: []*at.QueryResult{
+			{
+				Metric: map[string]string{"__name__": "metric1"},
+				Samples: []*at.Sample{
+					{Timestamp: 1704067200000, Value: float64(111)},
+					{Timestamp: 1705708800000, Value: float64(333)},
+				},
+			},
+			{
+				Metric: map[string]string{"__name__": "metric2"},
+				Samples: []*at.Sample{
+					{Timestamp: 1704067200000, Value: float64(222)},
+				},
+			},
+		},
+	})
+}
diff --git a/lib/storage/index_db_test.go b/lib/storage/index_db_test.go
index 574605810..ab3404d31 100644
--- a/lib/storage/index_db_test.go
+++ b/lib/storage/index_db_test.go
@@ -746,31 +746,31 @@ func testIndexDBCheckTSIDByName(db *indexDB, mns []MetricName, tsids []TSID, isC
 			return fmt.Errorf("expecting empty buf when searching for non-existent metricID; got %X", buf)
 		}
 
-		// Test SearchLabelValuesWithFiltersOnTimeRange
-		lvs, err := db.SearchLabelValuesWithFiltersOnTimeRange(nil, "__name__", nil, TimeRange{}, 1e5, 1e9, noDeadline)
+		// Test SearchLabelValues
+		lvs, err := db.SearchLabelValues(nil, "__name__", nil, TimeRange{}, 1e5, 1e9, noDeadline)
 		if err != nil {
-			return fmt.Errorf("error in SearchLabelValuesWithFiltersOnTimeRange(labelName=%q): %w", "__name__", err)
+			return fmt.Errorf("error in SearchLabelValues(labelName=%q): %w", "__name__", err)
 		}
 		if !hasValue(lvs, mn.MetricGroup) {
-			return fmt.Errorf("SearchLabelValuesWithFiltersOnTimeRange(labelName=%q): couldn't find %q; found %q", "__name__", mn.MetricGroup, lvs)
+			return fmt.Errorf("SearchLabelValues(labelName=%q): couldn't find %q; found %q", "__name__", mn.MetricGroup, lvs)
 		}
 		for i := range mn.Tags {
 			tag := &mn.Tags[i]
-			lvs, err := db.SearchLabelValuesWithFiltersOnTimeRange(nil, string(tag.Key), nil, TimeRange{}, 1e5, 1e9, noDeadline)
+			lvs, err := db.SearchLabelValues(nil, string(tag.Key), nil, TimeRange{}, 1e5, 1e9, noDeadline)
 			if err != nil {
-				return fmt.Errorf("error in SearchLabelValuesWithFiltersOnTimeRange(labelName=%q): %w", tag.Key, err)
+				return fmt.Errorf("error in SearchLabelValues(labelName=%q): %w", tag.Key, err)
 			}
 			if !hasValue(lvs, tag.Value) {
-				return fmt.Errorf("SearchLabelValuesWithFiltersOnTimeRange(labelName=%q): couldn't find %q; found %q", tag.Key, tag.Value, lvs)
+				return fmt.Errorf("SearchLabelValues(labelName=%q): couldn't find %q; found %q", tag.Key, tag.Value, lvs)
 			}
 			allLabelNames[string(tag.Key)] = true
 		}
 	}
 
-	// Test SearchLabelNamesWithFiltersOnTimeRange (empty filters, global time range)
-	lns, err := db.SearchLabelNamesWithFiltersOnTimeRange(nil, nil, TimeRange{}, 1e5, 1e9, noDeadline)
+	// Test SearchLabelNames (empty filters, global time range)
+	lns, err := db.SearchLabelNames(nil, nil, TimeRange{}, 1e5, 1e9, noDeadline)
 	if err != nil {
-		return fmt.Errorf("error in SearchLabelNamesWithFiltersOnTimeRange(empty filter, global time range): %w", err)
+		return fmt.Errorf("error in SearchLabelNames(empty filter, global time range): %w", err)
 	}
 	if !hasValue(lns, []byte("__name__")) {
 		return fmt.Errorf("cannot find __name__ in %q", lns)
@@ -1702,24 +1702,24 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 	db.putIndexSearch(is3)
 	s.DebugFlush()
 
-	// Check SearchLabelNamesWithFiltersOnTimeRange with the specified time range.
+	// Check SearchLabelNames with the specified time range.
 	tr := TimeRange{
 		MinTimestamp: int64(now) - msecPerDay,
 		MaxTimestamp: int64(now),
 	}
-	lns, err := db.SearchLabelNamesWithFiltersOnTimeRange(nil, nil, tr, 10000, 1e9, noDeadline)
+	lns, err := db.SearchLabelNames(nil, nil, tr, 10000, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("unexpected error in SearchLabelNamesWithFiltersOnTimeRange(timeRange=%s): %s", &tr, err)
+		t.Fatalf("unexpected error in SearchLabelNames(timeRange=%s): %s", &tr, err)
 	}
 	sort.Strings(lns)
 	if !reflect.DeepEqual(lns, labelNames) {
 		t.Fatalf("unexpected labelNames; got\n%s\nwant\n%s", lns, labelNames)
 	}
 
-	// Check SearchLabelValuesWithFiltersOnTimeRange with the specified time range.
-	lvs, err := db.SearchLabelValuesWithFiltersOnTimeRange(nil, "", nil, tr, 10000, 1e9, noDeadline)
+	// Check SearchLabelValues with the specified time range.
+	lvs, err := db.SearchLabelValues(nil, "", nil, tr, 10000, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("unexpected error in SearchLabelValuesWithFiltersOnTimeRange(timeRange=%s): %s", &tr, err)
+		t.Fatalf("unexpected error in SearchLabelValues(timeRange=%s): %s", &tr, err)
 	}
 	sort.Strings(lvs)
 	if !reflect.DeepEqual(lvs, labelValues) {
@@ -1753,60 +1753,60 @@ func TestSearchTSIDWithTimeRange(t *testing.T) {
 		t.Fatalf("expected %d time series for current day, got %d time series", metricsPerDay, len(matchedTSIDs))
 	}
 
-	// Check SearchLabelNamesWithFiltersOnTimeRange with the specified filter.
-	lns, err = db.SearchLabelNamesWithFiltersOnTimeRange(nil, []*TagFilters{tfs}, TimeRange{}, 10000, 1e9, noDeadline)
+	// Check SearchLabelNames with the specified filter.
+	lns, err = db.SearchLabelNames(nil, []*TagFilters{tfs}, TimeRange{}, 10000, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("unexpected error in SearchLabelNamesWithFiltersOnTimeRange(filters=%s): %s", tfs, err)
+		t.Fatalf("unexpected error in SearchLabelNames(filters=%s): %s", tfs, err)
 	}
 	sort.Strings(lns)
 	if !reflect.DeepEqual(lns, labelNames) {
 		t.Fatalf("unexpected labelNames; got\n%s\nwant\n%s", lns, labelNames)
 	}
 
-	// Check SearchLabelNamesWithFiltersOnTimeRange with the specified filter and time range.
-	lns, err = db.SearchLabelNamesWithFiltersOnTimeRange(nil, []*TagFilters{tfs}, tr, 10000, 1e9, noDeadline)
+	// Check SearchLabelNames with the specified filter and time range.
+	lns, err = db.SearchLabelNames(nil, []*TagFilters{tfs}, tr, 10000, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("unexpected error in SearchLabelNamesWithFiltersOnTimeRange(filters=%s, timeRange=%s): %s", tfs, &tr, err)
+		t.Fatalf("unexpected error in SearchLabelNames(filters=%s, timeRange=%s): %s", tfs, &tr, err)
 	}
 	sort.Strings(lns)
 	if !reflect.DeepEqual(lns, labelNames) {
 		t.Fatalf("unexpected labelNames; got\n%s\nwant\n%s", lns, labelNames)
 	}
 
-	// Check SearchLabelNamesWithFiltersOnTimeRange with filters on metric name and time range.
-	lns, err = db.SearchLabelNamesWithFiltersOnTimeRange(nil, []*TagFilters{tfsMetricName}, tr, 10000, 1e9, noDeadline)
+	// Check SearchLabelNames with filters on metric name and time range.
+	lns, err = db.SearchLabelNames(nil, []*TagFilters{tfsMetricName}, tr, 10000, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("unexpected error in SearchLabelNamesWithFiltersOnTimeRange(filters=%s, timeRange=%s): %s", tfs, &tr, err)
+		t.Fatalf("unexpected error in SearchLabelNames(filters=%s, timeRange=%s): %s", tfs, &tr, err)
 	}
 	sort.Strings(lns)
 	if !reflect.DeepEqual(lns, labelNames) {
 		t.Fatalf("unexpected labelNames; got\n%s\nwant\n%s", lns, labelNames)
 	}
 
-	// Check SearchLabelValuesWithFiltersOnTimeRange with the specified filter.
-	lvs, err = db.SearchLabelValuesWithFiltersOnTimeRange(nil, "", []*TagFilters{tfs}, TimeRange{}, 10000, 1e9, noDeadline)
+	// Check SearchLabelValues with the specified filter.
+	lvs, err = db.SearchLabelValues(nil, "", []*TagFilters{tfs}, TimeRange{}, 10000, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("unexpected error in SearchLabelValuesWithFiltersOnTimeRange(filters=%s): %s", tfs, err)
+		t.Fatalf("unexpected error in SearchLabelValues(filters=%s): %s", tfs, err)
 	}
 	sort.Strings(lvs)
 	if !reflect.DeepEqual(lvs, labelValues) {
 		t.Fatalf("unexpected labelValues; got\n%s\nwant\n%s", lvs, labelValues)
 	}
 
-	// Check SearchLabelValuesWithFiltersOnTimeRange with the specified filter and time range.
-	lvs, err = db.SearchLabelValuesWithFiltersOnTimeRange(nil, "", []*TagFilters{tfs}, tr, 10000, 1e9, noDeadline)
+	// Check SearchLabelValues with the specified filter and time range.
+	lvs, err = db.SearchLabelValues(nil, "", []*TagFilters{tfs}, tr, 10000, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("unexpected error in SearchLabelValuesWithFiltersOnTimeRange(filters=%s, timeRange=%s): %s", tfs, &tr, err)
+		t.Fatalf("unexpected error in SearchLabelValues(filters=%s, timeRange=%s): %s", tfs, &tr, err)
 	}
 	sort.Strings(lvs)
 	if !reflect.DeepEqual(lvs, labelValues) {
 		t.Fatalf("unexpected labelValues; got\n%s\nwant\n%s", lvs, labelValues)
 	}
 
-	// Check SearchLabelValuesWithFiltersOnTimeRange with filters on metric name and time range.
-	lvs, err = db.SearchLabelValuesWithFiltersOnTimeRange(nil, "", []*TagFilters{tfsMetricName}, tr, 10000, 1e9, noDeadline)
+	// Check SearchLabelValues with filters on metric name and time range.
+	lvs, err = db.SearchLabelValues(nil, "", []*TagFilters{tfsMetricName}, tr, 10000, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("unexpected error in SearchLabelValuesWithFiltersOnTimeRange(filters=%s, timeRange=%s): %s", tfs, &tr, err)
+		t.Fatalf("unexpected error in SearchLabelValues(filters=%s, timeRange=%s): %s", tfs, &tr, err)
 	}
 	sort.Strings(lvs)
 	if !reflect.DeepEqual(lvs, labelValues) {
diff --git a/lib/storage/search_test.go b/lib/storage/search_test.go
index 1f0dac2f4..001b29861 100644
--- a/lib/storage/search_test.go
+++ b/lib/storage/search_test.go
@@ -151,8 +151,7 @@ func TestSearch(t *testing.T) {
 	})
 }
 
-func testSearchInternal(st *Storage, tr TimeRange, mrs []MetricRow) error {
-	var s Search
+func testSearchInternal(s *Storage, tr TimeRange, mrs []MetricRow) error {
 	for i := 0; i < 10; i++ {
 		// Prepare TagFilters for search.
 		tfs := NewTagFilters()
@@ -185,76 +184,77 @@ func testSearchInternal(st *Storage, tr TimeRange, mrs []MetricRow) error {
 			expectedMrs = append(expectedMrs, *mr)
 		}
 
-		type metricBlock struct {
-			MetricName []byte
-			Block      *Block
+		if err := testAssertSearchResult(s, tr, tfs, expectedMrs); err != nil {
+			return err
 		}
+	}
+	return nil
+}
 
-		// Search
-		s.Init(nil, st, []*TagFilters{tfs}, tr, 1e5, noDeadline)
-		var mbs []metricBlock
-		for s.NextMetricBlock() {
-			var b Block
-			s.MetricBlockRef.BlockRef.MustReadBlock(&b)
+func testAssertSearchResult(st *Storage, tr TimeRange, tfs *TagFilters, want []MetricRow) error {
+	type metricBlock struct {
+		MetricName []byte
+		Block      *Block
+	}
 
-			var mb metricBlock
-			mb.MetricName = append(mb.MetricName, s.MetricBlockRef.MetricName...)
-			mb.Block = &b
-			mbs = append(mbs, mb)
-		}
-		if err := s.Error(); err != nil {
-			return fmt.Errorf("search error: %w", err)
-		}
-		s.MustClose()
+	var s Search
+	s.Init(nil, st, []*TagFilters{tfs}, tr, 1e5, noDeadline)
+	var mbs []metricBlock
+	for s.NextMetricBlock() {
+		var b Block
+		s.MetricBlockRef.BlockRef.MustReadBlock(&b)
 
-		// Build foundMrs.
-		var foundMrs []MetricRow
-		for _, mb := range mbs {
-			rb := newTestRawBlock(mb.Block, tr)
-			if err := mn.Unmarshal(mb.MetricName); err != nil {
-				return fmt.Errorf("cannot unmarshal MetricName: %w", err)
-			}
-			metricNameRaw := mn.marshalRaw(nil)
-			for i, timestamp := range rb.Timestamps {
-				mr := MetricRow{
-					MetricNameRaw: metricNameRaw,
-					Timestamp:     timestamp,
-					Value:         rb.Values[i],
-				}
-				foundMrs = append(foundMrs, mr)
-			}
-		}
+		var mb metricBlock
+		mb.MetricName = append(mb.MetricName, s.MetricBlockRef.MetricName...)
+		mb.Block = &b
+		mbs = append(mbs, mb)
+	}
+	if err := s.Error(); err != nil {
+		return fmt.Errorf("search error: %w", err)
+	}
+	s.MustClose()
 
-		// Compare expectedMrs to foundMrs.
-		sort.Slice(expectedMrs, func(i, j int) bool {
-			a, b := &expectedMrs[i], &expectedMrs[j]
-			cmp := bytes.Compare(a.MetricNameRaw, b.MetricNameRaw)
-			if cmp < 0 {
-				return true
-			}
-			if cmp > 0 {
-				return false
-			}
-			return a.Timestamp < b.Timestamp
-		})
-		sort.Slice(foundMrs, func(i, j int) bool {
-			a, b := &foundMrs[i], &foundMrs[j]
-			cmp := bytes.Compare(a.MetricNameRaw, b.MetricNameRaw)
-			if cmp < 0 {
-				return true
-			}
-			if cmp > 0 {
-				return false
+	var got []MetricRow
+	var mn MetricName
+	for _, mb := range mbs {
+		rb := newTestRawBlock(mb.Block, tr)
+		if err := mn.Unmarshal(mb.MetricName); err != nil {
+			return fmt.Errorf("cannot unmarshal MetricName: %w", err)
+		}
+		metricNameRaw := mn.marshalRaw(nil)
+		for i, timestamp := range rb.Timestamps {
+			mr := MetricRow{
+				MetricNameRaw: metricNameRaw,
+				Timestamp:     timestamp,
+				Value:         rb.Values[i],
 			}
-			return a.Timestamp < b.Timestamp
-		})
-		if !reflect.DeepEqual(expectedMrs, foundMrs) {
-			return fmt.Errorf("unexpected rows found;\ngot\n%s\nwant\n%s", mrsToString(foundMrs), mrsToString(expectedMrs))
+			got = append(got, mr)
 		}
 	}
+
+	testSortMetricRows(got)
+	testSortMetricRows(want)
+	if !reflect.DeepEqual(got, want) {
+		return fmt.Errorf("unexpected rows found;\ngot\n%s\nwant\n%s", mrsToString(got), mrsToString(want))
+	}
+
 	return nil
 }
 
+func testSortMetricRows(mrs []MetricRow) {
+	sort.Slice(mrs, func(i, j int) bool {
+		a, b := &mrs[i], &mrs[j]
+		cmp := bytes.Compare(a.MetricNameRaw, b.MetricNameRaw)
+		if cmp < 0 {
+			return true
+		}
+		if cmp > 0 {
+			return false
+		}
+		return a.Timestamp < b.Timestamp
+	})
+}
+
 func mrsToString(mrs []MetricRow) string {
 	var bb bytes.Buffer
 	fmt.Fprintf(&bb, "len=%d\n", len(mrs))
diff --git a/lib/storage/storage_test.go b/lib/storage/storage_test.go
index 2968fe5fc..d64ba2151 100644
--- a/lib/storage/storage_test.go
+++ b/lib/storage/storage_test.go
@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"slices"
 	"sort"
 	"sync"
 	"testing"
@@ -624,9 +625,9 @@ func TestStorageDeleteSeries(t *testing.T) {
 	s := MustOpenStorage(path, OpenOptions{})
 
 	// Verify no label names exist
-	lns, err := s.SearchLabelNamesWithFiltersOnTimeRange(nil, nil, TimeRange{}, 1e5, 1e9, noDeadline)
+	lns, err := s.SearchLabelNames(nil, nil, TimeRange{}, 1e5, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("error in SearchLabelNamesWithFiltersOnTimeRange() at the start: %s", err)
+		t.Fatalf("error in SearchLabelNames() at the start: %s", err)
 	}
 	if len(lns) != 0 {
 		t.Fatalf("found non-empty tag keys at the start: %q", lns)
@@ -673,9 +674,9 @@ func TestStorageDeleteSeries(t *testing.T) {
 	})
 
 	// Verify no more tag keys exist
-	lns, err = s.SearchLabelNamesWithFiltersOnTimeRange(nil, nil, TimeRange{}, 1e5, 1e9, noDeadline)
+	lns, err = s.SearchLabelNames(nil, nil, TimeRange{}, 1e5, 1e9, noDeadline)
 	if err != nil {
-		t.Fatalf("error in SearchLabelNamesWithFiltersOnTimeRange after the test: %s", err)
+		t.Fatalf("error in SearchLabelNames after the test: %s", err)
 	}
 	if len(lns) != 0 {
 		t.Fatalf("found non-empty tag keys after the test: %q", lns)
@@ -728,18 +729,18 @@ func testStorageDeleteSeries(s *Storage, workerNum int) error {
 	s.DebugFlush()
 
 	// Verify tag values exist
-	tvs, err := s.SearchLabelValuesWithFiltersOnTimeRange(nil, string(workerTag), nil, TimeRange{}, 1e5, 1e9, noDeadline)
+	tvs, err := s.SearchLabelValues(nil, string(workerTag), nil, TimeRange{}, 1e5, 1e9, noDeadline)
 	if err != nil {
-		return fmt.Errorf("error in SearchLabelValuesWithFiltersOnTimeRange before metrics removal: %w", err)
+		return fmt.Errorf("error in SearchLabelValues before metrics removal: %w", err)
 	}
 	if len(tvs) == 0 {
 		return fmt.Errorf("unexpected empty number of tag values for workerTag")
 	}
 
 	// Verify tag keys exist
-	lns, err := s.SearchLabelNamesWithFiltersOnTimeRange(nil, nil, TimeRange{}, 1e5, 1e9, noDeadline)
+	lns, err := s.SearchLabelNames(nil, nil, TimeRange{}, 1e5, 1e9, noDeadline)
 	if err != nil {
-		return fmt.Errorf("error in SearchLabelNamesWithFiltersOnTimeRange before metrics removal: %w", err)
+		return fmt.Errorf("error in SearchLabelNames before metrics removal: %w", err)
 	}
 	if err := checkLabelNames(lns, lnsAll); err != nil {
 		return fmt.Errorf("unexpected label names before metrics removal: %w", err)
@@ -801,9 +802,9 @@ func testStorageDeleteSeries(s *Storage, workerNum int) error {
 	if n := metricBlocksCount(tfs); n != 0 {
 		return fmt.Errorf("expecting zero metric blocks after deleting all the metrics; got %d blocks", n)
 	}
-	tvs, err = s.SearchLabelValuesWithFiltersOnTimeRange(nil, string(workerTag), nil, TimeRange{}, 1e5, 1e9, noDeadline)
+	tvs, err = s.SearchLabelValues(nil, string(workerTag), nil, TimeRange{}, 1e5, 1e9, noDeadline)
 	if err != nil {
-		return fmt.Errorf("error in SearchLabelValuesWithFiltersOnTimeRange after all the metrics are removed: %w", err)
+		return fmt.Errorf("error in SearchLabelValues after all the metrics are removed: %w", err)
 	}
 	if len(tvs) != 0 {
 		return fmt.Errorf("found non-empty tag values for %q after metrics removal: %q", workerTag, tvs)
@@ -864,8 +865,11 @@ func TestStorageDeleteSeries_TooManyTimeseries(t *testing.T) {
 func TestStorageDeleteSeries_CachesAreUpdatedOrReset(t *testing.T) {
 	defer testRemoveAll(t)
 
+	// For this test, the time range must be < 40 days. Otherwise the global
+	// index will be searched and the actual time range used in tag filters
+	// cache will be the globalTimeRange.
 	tr := TimeRange{
-		MinTimestamp: time.Now().Add(-100 * 24 * time.Hour).UnixMilli(),
+		MinTimestamp: time.Now().Add(-30 * 24 * time.Hour).UnixMilli(),
 		MaxTimestamp: time.Now().UnixMilli(),
 	}
 	mn := MetricName{MetricGroup: []byte("metric")}
@@ -1020,23 +1024,23 @@ func testStorageRegisterMetricNames(s *Storage) error {
 	// Verify the storage contains the added metric names.
 	s.DebugFlush()
 
-	// Verify that SearchLabelNamesWithFiltersOnTimeRange returns correct result.
+	// Verify that SearchLabelNames returns correct result.
 	lnsExpected := []string{
 		"__name__",
 		"add_id",
 		"instance",
 		"job",
 	}
-	lns, err := s.SearchLabelNamesWithFiltersOnTimeRange(nil, nil, TimeRange{}, 100, 1e9, noDeadline)
+	lns, err := s.SearchLabelNames(nil, nil, TimeRange{}, 100, 1e9, noDeadline)
 	if err != nil {
-		return fmt.Errorf("error in SearchLabelNamesWithFiltersOnTimeRange: %w", err)
+		return fmt.Errorf("error in SearchLabelNames: %w", err)
 	}
 	sort.Strings(lns)
 	if !reflect.DeepEqual(lns, lnsExpected) {
-		return fmt.Errorf("unexpected label names returned from SearchLabelNamesWithFiltersOnTimeRange;\ngot\n%q\nwant\n%q", lns, lnsExpected)
+		return fmt.Errorf("unexpected label names returned from SearchLabelNames;\ngot\n%q\nwant\n%q", lns, lnsExpected)
 	}
 
-	// Verify that SearchLabelNamesWithFiltersOnTimeRange with the specified time range returns correct result.
+	// Verify that SearchLabelNames with the specified time range returns correct result.
 	now := timestampFromTime(time.Now())
 	start := now - msecPerDay
 	end := now + 60*1000
@@ -1044,33 +1048,33 @@ func testStorageRegisterMetricNames(s *Storage) error {
 		MinTimestamp: start,
 		MaxTimestamp: end,
 	}
-	lns, err = s.SearchLabelNamesWithFiltersOnTimeRange(nil, nil, tr, 100, 1e9, noDeadline)
+	lns, err = s.SearchLabelNames(nil, nil, tr, 100, 1e9, noDeadline)
 	if err != nil {
-		return fmt.Errorf("error in SearchLabelNamesWithFiltersOnTimeRange: %w", err)
+		return fmt.Errorf("error in SearchLabelNames: %w", err)
 	}
 	sort.Strings(lns)
 	if !reflect.DeepEqual(lns, lnsExpected) {
-		return fmt.Errorf("unexpected label names returned from SearchLabelNamesWithFiltersOnTimeRange;\ngot\n%q\nwant\n%q", lns, lnsExpected)
+		return fmt.Errorf("unexpected label names returned from SearchLabelNames;\ngot\n%q\nwant\n%q", lns, lnsExpected)
 	}
 
-	// Verify that SearchLabelValuesWithFiltersOnTimeRange returns correct result.
-	addIDs, err := s.SearchLabelValuesWithFiltersOnTimeRange(nil, "add_id", nil, TimeRange{}, addsCount+100, 1e9, noDeadline)
+	// Verify that SearchLabelValues returns correct result.
+	addIDs, err := s.SearchLabelValues(nil, "add_id", nil, TimeRange{}, addsCount+100, 1e9, noDeadline)
 	if err != nil {
-		return fmt.Errorf("error in SearchLabelValuesWithFiltersOnTimeRange: %w", err)
+		return fmt.Errorf("error in SearchLabelValues: %w", err)
 	}
 	sort.Strings(addIDs)
 	if !reflect.DeepEqual(addIDs, addIDsExpected) {
-		return fmt.Errorf("unexpected tag values returned from SearchLabelValuesWithFiltersOnTimeRange;\ngot\n%q\nwant\n%q", addIDs, addIDsExpected)
+		return fmt.Errorf("unexpected tag values returned from SearchLabelValues;\ngot\n%q\nwant\n%q", addIDs, addIDsExpected)
 	}
 
-	// Verify that SearchLabelValuesWithFiltersOnTimeRange with the specified time range returns correct result.
-	addIDs, err = s.SearchLabelValuesWithFiltersOnTimeRange(nil, "add_id", nil, tr, addsCount+100, 1e9, noDeadline)
+	// Verify that SearchLabelValues with the specified time range returns correct result.
+	addIDs, err = s.SearchLabelValues(nil, "add_id", nil, tr, addsCount+100, 1e9, noDeadline)
 	if err != nil {
-		return fmt.Errorf("error in SearchLabelValuesWithFiltersOnTimeRange: %w", err)
+		return fmt.Errorf("error in SearchLabelValues: %w", err)
 	}
 	sort.Strings(addIDs)
 	if !reflect.DeepEqual(addIDs, addIDsExpected) {
-		return fmt.Errorf("unexpected tag values returned from SearchLabelValuesWithFiltersOnTimeRange;\ngot\n%q\nwant\n%q", addIDs, addIDsExpected)
+		return fmt.Errorf("unexpected tag values returned from SearchLabelValues;\ngot\n%q\nwant\n%q", addIDs, addIDsExpected)
 	}
 
 	// Verify that SearchMetricNames returns correct result.
@@ -1604,6 +1608,23 @@ func testCountAllMetricNames(s *Storage, tr TimeRange) int {
 	return len(names)
 }
 
+// testCountAllMetricIDs is a test helper function that counts the IDs of
+// all time series within the given time range.
+func testCountAllMetricIDs(s *Storage, tr TimeRange) int {
+	tfsAll := NewTagFilters()
+	if err := tfsAll.Add([]byte("__name__"), []byte(".*"), false, true); err != nil {
+		panic(fmt.Sprintf("unexpected error in TagFilters.Add: %v", err))
+	}
+	if s.disablePerDayIndex {
+		tr = globalIndexTimeRange
+	}
+	ids, err := s.idb().searchMetricIDs(nil, []*TagFilters{tfsAll}, tr, 1e9, noDeadline)
+	if err != nil {
+		panic(fmt.Sprintf("seachMetricIDs() failed unexpectedly: %s", err))
+	}
+	return len(ids)
+}
+
 func TestStorageSearchMetricNames_TooManyTimeseries(t *testing.T) {
 	defer testRemoveAll(t)
 
@@ -1819,18 +1840,620 @@ func TestStorageSearchMetricNames_TooManyTimeseries(t *testing.T) {
 	})
 }
 
-// testCountAllMetricIDs is a test helper function that counts the IDs of
-// all time series within the given time range.
-func testCountAllMetricIDs(s *Storage, tr TimeRange) int {
-	tfsAll := NewTagFilters()
-	if err := tfsAll.Add([]byte("__name__"), []byte(".*"), false, true); err != nil {
-		panic(fmt.Sprintf("unexpected error in TagFilters.Add: %v", err))
+func TestStorageDate(t *testing.T) {
+	defer testRemoveAll(t)
+
+	f := func(disablePerDayIndex bool, millis int64, want uint64) {
+		t.Helper()
+		s := MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: disablePerDayIndex,
+		})
+		defer s.MustClose()
+		if got := s.date(millis); got != want {
+			t.Errorf("unexpected date: got %d, want %d", got, want)
+		}
 	}
-	ids, err := s.idb().searchMetricIDs(nil, []*TagFilters{tfsAll}, tr, 1e9, noDeadline)
-	if err != nil {
-		panic(fmt.Sprintf("seachMetricIDs() failed unexpectedly: %s", err))
+
+	// Zero millis are converted to zero date regardless whether
+	// -disablePerDayIndex flag is set or not.
+	f(false, 0, 0)
+	f(true, 0, 0)
+
+	// When per-day index is enabled, positive millis are converted to the
+	// corresponding date.
+	f(false, 10*msecPerDay, 10)
+
+	// When per-day index is disabled, positive millis are converted to
+	// globalIndexDate.
+	f(true, 10*msecPerDay, globalIndexDate)
+}
+
+func TestStorageDate_negativeMillis(t *testing.T) {
+	defer testRemoveAll(t)
+
+	// Negative millis won't be converted to a negative date because the date is
+	// always positive. As a result, dates earlier than 1970-01-01 are not
+	// supported. However, when the -disablePerDayIndex flag is set, negative
+	// millis must be converted to globalIndexDate.
+	s := MustOpenStorage(t.Name(), OpenOptions{
+		DisablePerDayIndex: true,
+	})
+	millis := int64(-10 * msecPerDay)
+	want := globalIndexDate
+	if got := s.date(millis); got != want {
+		t.Errorf("unexpected date: got %d, want %d", got, want)
 	}
-	return len(ids)
+	s.MustClose()
+}
+
+func TestStorageAdjustTimeRange(t *testing.T) {
+	defer testRemoveAll(t)
+
+	f := func(disablePerDayIndex bool, tr TimeRange, want TimeRange) {
+		t.Helper()
+
+		s := MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: disablePerDayIndex,
+		})
+		defer s.MustClose()
+		if got := s.adjustTimeRange(tr); got != want {
+			t.Errorf("unexpected time range: got %v, want %v", &got, &want)
+		}
+	}
+
+	var tr TimeRange
+
+	// Zero time range is adjusted to globalIndexTimeRange regardless whether
+	// the -disablePerDayIndex flag is set or not.
+	tr = TimeRange{}
+	f(false, tr, globalIndexTimeRange)
+	f(true, tr, globalIndexTimeRange)
+
+	// Time range is smaller than 40 days. When the -disablePerDayIndex flag is
+	// unset, the time range will not be adjusted. When the flag is set, the
+	// adjusted time range will be globalIndexTimeRange.
+	tr = TimeRange{10 * msecPerDay, 50 * msecPerDay}
+	f(false, tr, tr)
+	f(true, tr, globalIndexTimeRange)
+
+	// Time range is exactly 40 days. In this case, the TimeRange.MaxTimestamp
+	// is set to the minimum possible value when the time range becomes 40 days.
+	// When the -disablePerDayIndex flag is unset, the time range will not be
+	// adjusted. When the flag is set, the adjusted time range will be
+	// globalIndexTimeRange.
+	tr = TimeRange{10 * msecPerDay, 50*msecPerDay + 1}
+	f(false, tr, tr)
+	f(true, tr, globalIndexTimeRange)
+
+	// Another case when time range is exactly 40 days. This time the the
+	// TimeRange.MaxTimestamp is set to its max value when the time range is
+	// still 40 days.
+	// When the -disablePerDayIndex flag is unset, the time range will not be
+	// adjusted. When the flag is set, the adjusted time range will be
+	// globalIndexTimeRange.
+	tr = TimeRange{10 * msecPerDay, 51 * msecPerDay}
+	f(false, tr, tr)
+	f(true, tr, globalIndexTimeRange)
+
+	// Time range is more than 40 days. The time range is adjusted to
+	// globalIndexTimeRange regardless whether the -disablePerDayIndex flag is
+	// set or not.
+	tr = TimeRange{10 * msecPerDay, 51*msecPerDay + 1}
+	f(false, tr, globalIndexTimeRange)
+	f(true, tr, globalIndexTimeRange)
+
+}
+
+type testStorageSearchWithoutPerDayIndexOptions struct {
+	mrs                []MetricRow
+	assertSearchResult func(t *testing.T, s *Storage, tr TimeRange, want any)
+	alwaysPerTimeRange bool // If true, use wantPerTimeRange instead of wantAll
+	wantPerTimeRange   map[TimeRange]any
+	wantAll            any
+	wantEmpty          any
+}
+
+// testStorageSearchWithoutPerDayIndex tests how the search behaves when the
+// per-day index is disabled. This function is expected to be called by
+// functions that test a particular search operation, such as GetTSDBStatus(),
+// SearchMetricNames(), etc.
+func testStorageSearchWithoutPerDayIndex(t *testing.T, opts *testStorageSearchWithoutPerDayIndexOptions) {
+	defer testRemoveAll(t)
+
+	// The data is inserted and the search is performed when the per-day index
+	// is enabled.
+	t.Run("InsertAndSearchWithPerDayIndex", func(t *testing.T) {
+		s := MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: false,
+		})
+		s.AddRows(opts.mrs, defaultPrecisionBits)
+		s.DebugFlush()
+		for tr, want := range opts.wantPerTimeRange {
+			opts.assertSearchResult(t, s, tr, want)
+		}
+		s.MustClose()
+	})
+
+	//  The data is inserted and the search is performed when the per-day index
+	//  is disabled.
+	t.Run("InsertAndSearchWithoutPerDayIndex", func(t *testing.T) {
+		s := MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: true,
+		})
+		s.AddRows(opts.mrs, defaultPrecisionBits)
+		s.DebugFlush()
+		for tr, want := range opts.wantPerTimeRange {
+			if !opts.alwaysPerTimeRange {
+				want = opts.wantAll
+			}
+			opts.assertSearchResult(t, s, tr, want)
+		}
+		s.MustClose()
+	})
+
+	// The data is inserted when the per-day index is enabled but the search is
+	// performed when the per-day index is disabled.
+	t.Run("InsertWithPerDayIndexSearchWithout", func(t *testing.T) {
+		s := MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: false,
+		})
+		s.AddRows(opts.mrs, defaultPrecisionBits)
+		s.DebugFlush()
+		s.MustClose()
+
+		s = MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: true,
+		})
+		for tr, want := range opts.wantPerTimeRange {
+			if !opts.alwaysPerTimeRange {
+				want = opts.wantAll
+			}
+			opts.assertSearchResult(t, s, tr, want)
+		}
+		s.MustClose()
+	})
+
+	// The data is inserted when the per-day index is disabled but the search is
+	// performed when the per-day index is enabled. This case also shows that
+	// registering metric names recovers the per-day index.
+	t.Run("InsertWithoutPerDayIndexSearchWith", func(t *testing.T) {
+		s := MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: true,
+		})
+		s.AddRows(opts.mrs, defaultPrecisionBits)
+		s.DebugFlush()
+		s.MustClose()
+
+		s = MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: false,
+		})
+		for tr := range opts.wantPerTimeRange {
+			opts.assertSearchResult(t, s, tr, opts.wantEmpty)
+		}
+
+		// Verify that search result contains correct label values after populating
+		// per-day index by registering metric names.
+		s.RegisterMetricNames(nil, opts.mrs)
+		s.DebugFlush()
+		for tr, want := range opts.wantPerTimeRange {
+			opts.assertSearchResult(t, s, tr, want)
+		}
+		s.MustClose()
+	})
+}
+
+func TestStorageGetTSDBStatusWithoutPerDayIndex(t *testing.T) {
+	const (
+		days = 4
+		rows = 10
+	)
+	rng := rand.New(rand.NewSource(1))
+	opts := testStorageSearchWithoutPerDayIndexOptions{
+		wantEmpty:        &TSDBStatus{},
+		wantPerTimeRange: make(map[TimeRange]any),
+		wantAll:          &TSDBStatus{TotalSeries: days * rows},
+	}
+	for day := 1; day <= days; day++ {
+		tr := TimeRange{
+			MinTimestamp: time.Date(2024, 1, day, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2024, 1, day, 23, 59, 59, 999, time.UTC).UnixMilli(),
+		}
+		for row := 0; row < rows; row++ {
+			name := fmt.Sprintf("metric_%d", rows*day+row)
+			mn := &MetricName{
+				MetricGroup: []byte(name),
+			}
+			metricNameRaw := mn.marshalRaw(nil)
+			opts.mrs = append(opts.mrs, MetricRow{
+				MetricNameRaw: metricNameRaw,
+				Timestamp:     rng.Int63n(tr.MaxTimestamp-tr.MinTimestamp) + tr.MinTimestamp,
+				Value:         rng.NormFloat64() * 1e6,
+			})
+		}
+		opts.wantPerTimeRange[tr] = &TSDBStatus{TotalSeries: rows}
+	}
+
+	opts.assertSearchResult = func(t *testing.T, s *Storage, tr TimeRange, want any) {
+		t.Helper()
+
+		date := uint64(tr.MinTimestamp) / msecPerDay
+		gotStatus, err := s.GetTSDBStatus(nil, nil, date, "", 10, 1e6, noDeadline)
+		if err != nil {
+			t.Fatalf("GetTSDBStatus(%v) failed unexpectedly", &tr)
+		}
+
+		wantStatus := want.(*TSDBStatus)
+		if got, want := gotStatus.TotalSeries, wantStatus.TotalSeries; got != want {
+			t.Errorf("[%v] unexpected TSDBStatus.TotalSeries: got %d, want %d", &tr, got, want)
+		}
+	}
+
+	testStorageSearchWithoutPerDayIndex(t, &opts)
+}
+
+func TestStorageSearchMetricNamesWithoutPerDayIndex(t *testing.T) {
+	const (
+		days = 4
+		rows = 10
+	)
+	rng := rand.New(rand.NewSource(1))
+	opts := testStorageSearchWithoutPerDayIndexOptions{
+		wantEmpty:        []string(nil),
+		wantPerTimeRange: make(map[TimeRange]any),
+		wantAll:          []string{},
+	}
+	for day := 1; day <= days; day++ {
+		tr := TimeRange{
+			MinTimestamp: time.Date(2024, 1, day, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2024, 1, day, 23, 59, 59, 999, time.UTC).UnixMilli(),
+		}
+		var want []string
+		for row := 0; row < rows; row++ {
+			name := fmt.Sprintf("metric_%d", rows*day+row)
+			mn := &MetricName{
+				MetricGroup: []byte(name),
+			}
+			metricNameRaw := mn.marshalRaw(nil)
+			want = append(want, string(name))
+			opts.mrs = append(opts.mrs, MetricRow{
+				MetricNameRaw: metricNameRaw,
+				Timestamp:     rng.Int63n(tr.MaxTimestamp-tr.MinTimestamp) + tr.MinTimestamp,
+				Value:         rng.NormFloat64() * 1e6,
+			})
+		}
+		opts.wantPerTimeRange[tr] = want
+		opts.wantAll = append(opts.wantAll.([]string), want...)
+	}
+
+	opts.assertSearchResult = func(t *testing.T, s *Storage, tr TimeRange, want any) {
+		t.Helper()
+
+		tfsAll := NewTagFilters()
+		if err := tfsAll.Add([]byte("__name__"), []byte(".*"), false, true); err != nil {
+			panic(fmt.Sprintf("unexpected error in TagFilters.Add: %v", err))
+		}
+		got, err := s.SearchMetricNames(nil, []*TagFilters{tfsAll}, tr, 1e6, noDeadline)
+		if err != nil {
+			t.Fatalf("SearchMetricNames(%v) failed unexpectedly: %v", &tr, err)
+		}
+		for i, name := range got {
+			var mn MetricName
+			if err := mn.Unmarshal([]byte(name)); err != nil {
+				t.Fatalf("mn.Unmarshal(%q) failed unexpectedly: %v", name, err)
+			}
+			got[i] = string(mn.MetricGroup)
+		}
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("[%v] unexpected metric names: got %v, want %v", &tr, got, want)
+		}
+	}
+
+	testStorageSearchWithoutPerDayIndex(t, &opts)
+}
+
+func TestStorageSearchLabelNamesWithoutPerDayIndex(t *testing.T) {
+	const (
+		days = 4
+		rows = 10
+	)
+	rng := rand.New(rand.NewSource(1))
+	opts := testStorageSearchWithoutPerDayIndexOptions{
+		wantEmpty:        []string{},
+		wantPerTimeRange: make(map[TimeRange]any),
+		wantAll:          []string{},
+	}
+	for day := 1; day <= days; day++ {
+		tr := TimeRange{
+			MinTimestamp: time.Date(2024, 1, day, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2024, 1, day, 23, 59, 59, 999, time.UTC).UnixMilli(),
+		}
+		var want []string
+		for row := 0; row < rows; row++ {
+			labelName := fmt.Sprintf("job_%d", rows*day+row)
+			mn := &MetricName{
+				MetricGroup: []byte("metric"),
+				Tags: []Tag{
+					{[]byte(labelName), []byte("webservice")},
+				},
+			}
+			metricNameRaw := mn.marshalRaw(nil)
+			want = append(want, labelName)
+			opts.mrs = append(opts.mrs, MetricRow{
+				MetricNameRaw: metricNameRaw,
+				Timestamp:     rng.Int63n(tr.MaxTimestamp-tr.MinTimestamp) + tr.MinTimestamp,
+				Value:         rng.NormFloat64() * 1e6,
+			})
+		}
+		opts.wantAll = append(opts.wantAll.([]string), want...)
+		opts.wantPerTimeRange[tr] = append(want, "__name__")
+	}
+	opts.wantAll = append(opts.wantAll.([]string), "__name__")
+
+	opts.assertSearchResult = func(t *testing.T, s *Storage, tr TimeRange, want any) {
+		t.Helper()
+		got, err := s.SearchLabelNames(nil, []*TagFilters{}, tr, 1e6, 1e6, noDeadline)
+		if err != nil {
+			t.Fatalf("SearchLabelNames(%v) failed unexpectedly: %v", &tr, err)
+		}
+		slices.Sort(got)
+		slices.Sort(want.([]string))
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("[%v] unexpected label names: got %v, want %v", &tr, got, want)
+		}
+	}
+
+	testStorageSearchWithoutPerDayIndex(t, &opts)
+}
+
+func TestStorageSearchLabelValuesWithoutPerDayIndex(t *testing.T) {
+	const (
+		days      = 4
+		rows      = 10
+		labelName = "job"
+	)
+	rng := rand.New(rand.NewSource(1))
+	opts := testStorageSearchWithoutPerDayIndexOptions{
+		wantEmpty:        []string{},
+		wantPerTimeRange: make(map[TimeRange]any),
+		wantAll:          []string{},
+	}
+	for day := 1; day <= days; day++ {
+		tr := TimeRange{
+			MinTimestamp: time.Date(2024, 1, day, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2024, 1, day, 23, 59, 59, 999, time.UTC).UnixMilli(),
+		}
+		var want []string
+		for row := 0; row < rows; row++ {
+			labelValue := fmt.Sprintf("webservice_%d", rows*day+row)
+			mn := &MetricName{
+				MetricGroup: []byte("metric"),
+				Tags: []Tag{
+					{[]byte(labelName), []byte(labelValue)},
+				},
+			}
+			metricNameRaw := mn.marshalRaw(nil)
+			want = append(want, labelValue)
+			opts.mrs = append(opts.mrs, MetricRow{
+				MetricNameRaw: metricNameRaw,
+				Timestamp:     rng.Int63n(tr.MaxTimestamp-tr.MinTimestamp) + tr.MinTimestamp,
+				Value:         rng.NormFloat64() * 1e6,
+			})
+		}
+		opts.wantPerTimeRange[tr] = want
+		opts.wantAll = append(opts.wantAll.([]string), want...)
+	}
+
+	opts.assertSearchResult = func(t *testing.T, s *Storage, tr TimeRange, want any) {
+		t.Helper()
+		got, err := s.SearchLabelValues(nil, labelName, []*TagFilters{}, tr, 1e6, 1e6, noDeadline)
+		if err != nil {
+			t.Fatalf("SearchLabelValues(%v) failed unexpectedly: %v", &tr, err)
+		}
+		slices.Sort(got)
+		slices.Sort(want.([]string))
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("[%v] unexpected label values: got %v, want %v", &tr, got, want)
+		}
+	}
+
+	testStorageSearchWithoutPerDayIndex(t, &opts)
+}
+
+func TestStorageSearchTagValueSuffixesWithoutPerDayIndex(t *testing.T) {
+	const (
+		days           = 4
+		rows           = 10
+		tagValuePrefix = "metric."
+	)
+	rng := rand.New(rand.NewSource(1))
+	opts := testStorageSearchWithoutPerDayIndexOptions{
+		wantEmpty:        []string{},
+		wantPerTimeRange: make(map[TimeRange]any),
+		wantAll:          []string{},
+	}
+	for day := 1; day <= days; day++ {
+		tr := TimeRange{
+			MinTimestamp: time.Date(2024, 1, day, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2024, 1, day, 23, 59, 59, 999, time.UTC).UnixMilli(),
+		}
+		for row := 0; row < rows; row++ {
+			metricName := fmt.Sprintf("%sday%d.row%d", tagValuePrefix, day, row)
+			mn := &MetricName{
+				MetricGroup: []byte(metricName),
+			}
+			metricNameRaw := mn.marshalRaw(nil)
+			opts.mrs = append(opts.mrs, MetricRow{
+				MetricNameRaw: metricNameRaw,
+				Timestamp:     rng.Int63n(tr.MaxTimestamp-tr.MinTimestamp) + tr.MinTimestamp,
+				Value:         rng.NormFloat64() * 1e6,
+			})
+		}
+		want := fmt.Sprintf("day%d.", day)
+		opts.wantPerTimeRange[tr] = []string{want}
+		opts.wantAll = append(opts.wantAll.([]string), want)
+	}
+
+	opts.assertSearchResult = func(t *testing.T, s *Storage, tr TimeRange, want any) {
+		t.Helper()
+		got, err := s.SearchTagValueSuffixes(nil, tr, "", tagValuePrefix, '.', 1e6, noDeadline)
+		if err != nil {
+			t.Fatalf("SearchTagValueSuffixes(%v) failed unexpectedly: %v", &tr, err)
+		}
+		slices.Sort(got)
+		slices.Sort(want.([]string))
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("[%v] unexpected tag value suffixes: got %v, want %v", &tr, got, want)
+		}
+	}
+
+	testStorageSearchWithoutPerDayIndex(t, &opts)
+}
+
+func TestStorageSearchGraphitePathsWithoutPerDayIndex(t *testing.T) {
+	const (
+		days = 4
+		rows = 10
+	)
+	rng := rand.New(rand.NewSource(1))
+	opts := testStorageSearchWithoutPerDayIndexOptions{
+		wantEmpty:        []string(nil),
+		wantPerTimeRange: make(map[TimeRange]any),
+		wantAll:          []string{},
+	}
+	for day := 1; day <= days; day++ {
+		tr := TimeRange{
+			MinTimestamp: time.Date(2024, 1, day, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2024, 1, day, 23, 59, 59, 999, time.UTC).UnixMilli(),
+		}
+		want := make([]string, rows)
+		for row := 0; row < rows; row++ {
+			metricName := fmt.Sprintf("day%d.row%d", day, row)
+			mn := &MetricName{
+				MetricGroup: []byte(metricName),
+			}
+			metricNameRaw := mn.marshalRaw(nil)
+			opts.mrs = append(opts.mrs, MetricRow{
+				MetricNameRaw: metricNameRaw,
+				Timestamp:     rng.Int63n(tr.MaxTimestamp-tr.MinTimestamp) + tr.MinTimestamp,
+				Value:         rng.NormFloat64() * 1e6,
+			})
+			want[row] = metricName
+		}
+		opts.wantPerTimeRange[tr] = want
+		opts.wantAll = append(opts.wantAll.([]string), want...)
+	}
+
+	opts.assertSearchResult = func(t *testing.T, s *Storage, tr TimeRange, want any) {
+		t.Helper()
+		got, err := s.SearchGraphitePaths(nil, tr, []byte("*.*"), 1e6, noDeadline)
+		if err != nil {
+			t.Fatalf("SearchGraphitePaths(%v) failed unexpectedly: %v", &tr, err)
+		}
+		slices.Sort(got)
+		slices.Sort(want.([]string))
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("[%v] unexpected graphite paths: got %v, want %v", &tr, got, want)
+		}
+	}
+
+	testStorageSearchWithoutPerDayIndex(t, &opts)
+}
+
+func TestStorageQueryWithoutPerDayIndex(t *testing.T) {
+	const (
+		days = 4
+		rows = 10
+	)
+	rng := rand.New(rand.NewSource(1))
+	opts := testStorageSearchWithoutPerDayIndexOptions{
+		wantEmpty:          []MetricRow(nil),
+		wantPerTimeRange:   make(map[TimeRange]any),
+		alwaysPerTimeRange: true,
+	}
+	for day := 1; day <= days; day++ {
+		tr := TimeRange{
+			MinTimestamp: time.Date(2024, 1, day, 0, 0, 0, 0, time.UTC).UnixMilli(),
+			MaxTimestamp: time.Date(2024, 1, day, 23, 59, 59, 999, time.UTC).UnixMilli(),
+		}
+		var want []MetricRow
+		for row := 0; row < rows; row++ {
+			seqNumber := rows*day + row
+			name := fmt.Sprintf("metric_%d", seqNumber)
+			mn := &MetricName{
+				MetricGroup: []byte(name),
+			}
+			metricNameRaw := mn.marshalRaw(nil)
+			mr := MetricRow{
+				MetricNameRaw: metricNameRaw,
+				Timestamp:     rng.Int63n(tr.MaxTimestamp-tr.MinTimestamp) + tr.MinTimestamp,
+				Value:         float64(seqNumber),
+			}
+			opts.mrs = append(opts.mrs, mr)
+			want = append(want, mr)
+		}
+		opts.wantPerTimeRange[tr] = want
+	}
+
+	opts.assertSearchResult = func(t *testing.T, s *Storage, tr TimeRange, want any) {
+		t.Helper()
+
+		tfs := NewTagFilters()
+		if err := tfs.Add(nil, []byte(`metric_\d*`), false, true); err != nil {
+			t.Fatalf("unexpected error in TagFilters.Add: %v", err)
+		}
+		if err := testAssertSearchResult(s, tr, tfs, want.([]MetricRow)); err != nil {
+			t.Errorf("%v: %v", &tr, err)
+		}
+	}
+
+	testStorageSearchWithoutPerDayIndex(t, &opts)
+}
+
+func TestStorageAddRows_SamplesWithZeroDate(t *testing.T) {
+	defer testRemoveAll(t)
+
+	f := func(t *testing.T, disablePerDayIndex bool) {
+		t.Helper()
+
+		s := MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: disablePerDayIndex,
+		})
+		defer s.MustClose()
+
+		mn := MetricName{MetricGroup: []byte("metric")}
+		mr := MetricRow{MetricNameRaw: mn.marshalRaw(nil)}
+		for range 10 {
+			mr.Timestamp = rand.Int63n(msecPerDay)
+			mr.Value = float64(rand.Intn(1000))
+			s.AddRows([]MetricRow{mr}, defaultPrecisionBits)
+			s.DebugFlush()
+			// Reset TSID cache so that insertion takes the path that involves
+			// checking whether the index contains metricName->TSID mapping.
+			s.resetAndSaveTSIDCache()
+		}
+
+		want := 1
+		firstUnixDay := TimeRange{
+			MinTimestamp: 0,
+			MaxTimestamp: msecPerDay - 1,
+		}
+		if got := s.newTimeseriesCreated.Load(); got != uint64(want) {
+			t.Errorf("unexpected new timeseries count: got %d, want %d", got, want)
+		}
+		if got := testCountAllMetricNames(s, firstUnixDay); got != want {
+			t.Errorf("unexpected metric name count: got %d, want %d", got, want)
+		}
+		if got := testCountAllMetricIDs(s, firstUnixDay); got != want {
+			t.Errorf("unexpected metric id count: got %d, want %d", got, want)
+		}
+	}
+
+	t.Run("disablePerDayIndex=false", func(t *testing.T) {
+		f(t, false)
+	})
+	t.Run("disablePerDayIndex=true", func(t *testing.T) {
+		f(t, true)
+	})
 }
 
 func TestStorageRegisterMetricNamesForVariousDataPatternsConcurrently(t *testing.T) {
@@ -1846,7 +2469,8 @@ func TestStorageAddRowsForVariousDataPatternsConcurrently(t *testing.T) {
 }
 
 // testStorageVariousDataPatternsConcurrently tests different concurrency use
-// cases when ingesting data of different patterns.
+// cases when ingesting data of different patterns. Each concurrency use case
+// considered with and without the per-day index.
 //
 // The function is intended to be used by other tests that define which
 // operation (AddRows or RegisterMetricNames) is tested.
@@ -1855,14 +2479,26 @@ func testStorageVariousDataPatternsConcurrently(t *testing.T, registerOnly bool,
 
 	const concurrency = 4
 
-	t.Run("serial", func(t *testing.T) {
-		testStorageVariousDataPatterns(t, registerOnly, op, 1, false)
+	disablePerDayIndex := false
+	t.Run("perDayIndexes/serial", func(t *testing.T) {
+		testStorageVariousDataPatterns(t, disablePerDayIndex, registerOnly, op, 1, false)
 	})
-	t.Run("concurrentRows", func(t *testing.T) {
-		testStorageVariousDataPatterns(t, registerOnly, op, concurrency, true)
+	t.Run("perDayIndexes/concurrentRows", func(t *testing.T) {
+		testStorageVariousDataPatterns(t, disablePerDayIndex, registerOnly, op, concurrency, true)
 	})
-	t.Run("concurrentBatches", func(t *testing.T) {
-		testStorageVariousDataPatterns(t, registerOnly, op, concurrency, false)
+	t.Run("perDayIndexes/concurrentBatches", func(t *testing.T) {
+		testStorageVariousDataPatterns(t, disablePerDayIndex, registerOnly, op, concurrency, false)
+	})
+
+	disablePerDayIndex = true
+	t.Run("noPerDayIndexes/serial", func(t *testing.T) {
+		testStorageVariousDataPatterns(t, disablePerDayIndex, registerOnly, op, 1, false)
+	})
+	t.Run("noPerDayIndexes/concurrentRows", func(t *testing.T) {
+		testStorageVariousDataPatterns(t, disablePerDayIndex, registerOnly, op, concurrency, true)
+	})
+	t.Run("noPerDayIndexes/concurrentBatches", func(t *testing.T) {
+		testStorageVariousDataPatterns(t, disablePerDayIndex, registerOnly, op, concurrency, false)
 	})
 }
 
@@ -1870,12 +2506,14 @@ func testStorageVariousDataPatternsConcurrently(t *testing.T, registerOnly bool,
 // of metric names and dates.
 //
 // The function is intended to be used by other tests that define the
-// concurrency and the operation (AddRows or RegisterMetricNames) under test.
-func testStorageVariousDataPatterns(t *testing.T, registerOnly bool, op func(s *Storage, mrs []MetricRow), concurrency int, splitBatches bool) {
+// concurrency, the per-day index setting, and the operation (AddRows or
+// RegisterMetricNames) under test.
+func testStorageVariousDataPatterns(t *testing.T, disablePerDayIndex, registerOnly bool, op func(s *Storage, mrs []MetricRow), concurrency int, splitBatches bool) {
 	f := func(t *testing.T, sameBatchMetricNames, sameRowMetricNames, sameBatchDates, sameRowDates bool) {
 		batches, wantCounts := testGenerateMetricRowBatches(&batchOptions{
 			numBatches:           3,
 			numRowsPerBatch:      30,
+			disablePerDayIndex:   disablePerDayIndex,
 			registerOnly:         registerOnly,
 			sameBatchMetricNames: sameBatchMetricNames,
 			sameRowMetricNames:   sameRowMetricNames,
@@ -1885,7 +2523,9 @@ func testStorageVariousDataPatterns(t *testing.T, registerOnly bool, op func(s *
 		strict := concurrency == 1
 		rowsAddedTotal := wantCounts.metrics.RowsAddedTotal
 
-		s := MustOpenStorage(t.Name(), OpenOptions{})
+		s := MustOpenStorage(t.Name(), OpenOptions{
+			DisablePerDayIndex: disablePerDayIndex,
+		})
 
 		testDoConcurrently(s, op, concurrency, splitBatches, batches)
 		s.DebugFlush()
@@ -2151,6 +2791,7 @@ func assertCounts(t *testing.T, s *Storage, want *counts, strict bool) {
 type batchOptions struct {
 	numBatches           int
 	numRowsPerBatch      int
+	disablePerDayIndex   bool
 	registerOnly         bool
 	sameBatchMetricNames bool
 	sameRowMetricNames   bool
@@ -2222,6 +2863,7 @@ func testGenerateMetricRowBatches(opts *batchOptions) ([][]MetricRow, *counts) {
 
 	allTimeseries := len(names)
 	rowsAddedTotal := uint64(opts.numBatches * opts.numRowsPerBatch)
+
 	// When RegisterMetricNames() is called it only restisters the time series
 	// in IndexDB but no samples is written to the storage.
 	if opts.registerOnly {
@@ -2236,7 +2878,12 @@ func testGenerateMetricRowBatches(opts *batchOptions) ([][]MetricRow, *counts) {
 		dateTSDBStatuses: make(map[uint64]*TSDBStatus),
 	}
 	for tr, names := range trNames {
-		count := len(names)
+		var count int
+		if opts.disablePerDayIndex {
+			count = allTimeseries
+		} else {
+			count = len(names)
+		}
 		date := uint64(tr.MinTimestamp / msecPerDay)
 		want.timeRangeCounts[tr] = count
 		want.dateTSDBStatuses[date] = &TSDBStatus{
diff --git a/lib/storage/storage_timing_test.go b/lib/storage/storage_timing_test.go
index a60ab0c1b..0c94d94e7 100644
--- a/lib/storage/storage_timing_test.go
+++ b/lib/storage/storage_timing_test.go
@@ -2,9 +2,13 @@ package storage
 
 import (
 	"fmt"
+	"io/fs"
 	"os"
+	"slices"
 	"sync/atomic"
 	"testing"
+
+	vmfs "github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 )
 
 func BenchmarkStorageAddRows(b *testing.B) {
@@ -51,3 +55,111 @@ func benchmarkStorageAddRows(b *testing.B, rowsPerBatch int) {
 	})
 	b.StopTimer()
 }
+
+func BenchmarkStorageInsertWithAndWithoutPerDayIndex(b *testing.B) {
+	const (
+		numBatches      = 100
+		numRowsPerBatch = 10000
+		concurrency     = 1
+		splitBatches    = true
+	)
+
+	// Each batch corresponds to a unique date and has a unique set of metric
+	// names.
+	highChurnRateData, _ := testGenerateMetricRowBatches(&batchOptions{
+		numBatches:           numBatches,
+		numRowsPerBatch:      numRowsPerBatch,
+		sameBatchMetricNames: false, // Each batch has unique set of metric names.
+		sameRowMetricNames:   false, // Within a batch, each metric name is unique.
+		sameBatchDates:       false, // Each batch has a unique date.
+		sameRowDates:         true,  // Within a batch, the date is the same.
+	})
+
+	// Each batch corresponds to a unique date but has the same set of metric
+	// names.
+	noChurnRateData, _ := testGenerateMetricRowBatches(&batchOptions{
+		numBatches:           numBatches,
+		numRowsPerBatch:      numRowsPerBatch,
+		sameBatchMetricNames: true,  // Each batch has the same set of metric names.
+		sameRowMetricNames:   false, // Within a batch, each metric name is unique.
+		sameBatchDates:       false, // Each batch has a unique date.
+		sameRowDates:         true,  // Within a batch, the date is the same.
+	})
+
+	addRows := func(b *testing.B, disablePerDayIndex bool, batches [][]MetricRow) {
+		b.Helper()
+
+		var (
+			rowsAddedTotal int
+			dataSize       int64
+			indexSize      int64
+		)
+
+		path := b.Name()
+		for range b.N {
+			s := MustOpenStorage(path, OpenOptions{
+				DisablePerDayIndex: disablePerDayIndex,
+			})
+			s.AddRows(slices.Concat(batches...), defaultPrecisionBits)
+			s.DebugFlush()
+			if err := s.ForceMergePartitions(""); err != nil {
+				b.Fatalf("ForceMergePartitions() failed unexpectedly: %v", err)
+			}
+
+			// Reopen storage to ensure that index has been written to disk.
+			s.MustClose()
+			s = MustOpenStorage(path, OpenOptions{
+				DisablePerDayIndex: disablePerDayIndex,
+			})
+
+			rowsAddedTotal = numBatches * numRowsPerBatch
+			dataSize = benchmarkDirSize(path + "/data")
+			indexSize = benchmarkDirSize(path + "/indexdb")
+
+			s.MustClose()
+			vmfs.MustRemoveAll(path)
+		}
+
+		b.ReportMetric(float64(rowsAddedTotal)/float64(b.Elapsed().Seconds()), "rows/s")
+		b.ReportMetric(float64(dataSize)/(1024*1024), "data-MiB")
+		b.ReportMetric(float64(indexSize)/(1024*1024), "indexdb-MiB")
+	}
+
+	b.Run("HighChurnRate/perDayIndexes", func(b *testing.B) {
+		addRows(b, false, highChurnRateData)
+	})
+
+	b.Run("HighChurnRate/noPerDayIndexes", func(b *testing.B) {
+		addRows(b, true, highChurnRateData)
+	})
+
+	b.Run("NoChurnRate/perDayIndexes", func(b *testing.B) {
+		addRows(b, false, noChurnRateData)
+	})
+
+	b.Run("NoChurnRate/noPerDayIndexes", func(b *testing.B) {
+		addRows(b, true, noChurnRateData)
+	})
+}
+
+// benchmarkDirSize calculates the size of a directory.
+func benchmarkDirSize(path string) int64 {
+	var size int64
+	err := fs.WalkDir(os.DirFS(path), ".", func(_ string, d fs.DirEntry, err error) error {
+		if err != nil {
+			panic(err)
+		}
+		if !d.IsDir() {
+			info, err := d.Info()
+			if err != nil {
+				panic(err)
+			}
+			size += info.Size()
+		}
+		return nil
+	})
+	if err != nil {
+		panic(err)
+	}
+	return size
+}
diff --git a/lib/storage/time_test.go b/lib/storage/time_test.go
index 27dcfaf5f..9a1a81bdd 100644
--- a/lib/storage/time_test.go
+++ b/lib/storage/time_test.go
@@ -52,3 +52,82 @@ func testTimeRangeFromPartition(t *testing.T, initialTime time.Time) {
 		t.Fatalf("unexpected nextY, nextM; got %d, %d; want %d, %d+1;\nnextTime=%s\nmaxTime=%s", nextY, nextM, maxY, maxM, nextTime, maxTime)
 	}
 }
+
+func TestTimeRangeDateRange(t *testing.T) {
+	f := func(tr TimeRange, wantMinDate, wantMaxDate uint64) {
+		t.Helper()
+
+		gotMinDate, gotMaxDate := tr.DateRange()
+		if gotMinDate != wantMinDate {
+			t.Errorf("unexpected min date: got %d, want %d", gotMinDate, wantMinDate)
+		}
+		if gotMaxDate != wantMaxDate {
+			t.Errorf("unexpected max date: got %d, want %d", gotMaxDate, wantMaxDate)
+		}
+	}
+
+	var tr TimeRange
+
+	// MinTimestamp is less than MaxTimestamp, the timestamps belong to the
+	// different days. Min date must be less than the max date.
+	tr = TimeRange{1*msecPerDay + 123, 2*msecPerDay + 456}
+	f(tr, 1, 2)
+
+	// MinTimestamp is less than MaxTimestamp and both timestamps belong to the
+	// same day. Max date must be the same as min date.
+	tr = TimeRange{1*msecPerDay + 123, 1*msecPerDay + 456}
+	f(tr, 1, 1)
+
+	// MinTimestamp equals to MaxTimestamp. Max date must be the same as min
+	// date.
+	tr = TimeRange{1*msecPerDay + 123, 1*msecPerDay + 123}
+	f(tr, 1, 1)
+
+	// MinTimestamp is the first millisecond of the day and equals to
+	// MaxTimestamp. Min and max dates must be the same.
+	tr = TimeRange{1 * msecPerDay, 1 * msecPerDay}
+	f(tr, 1, 1)
+
+	// MinTimestamp is greater than MaxTimestamp MaxTimestamp. Max date must be
+	// the same as min date.
+	tr = TimeRange{2*msecPerDay + 654, 1*msecPerDay + 321}
+	f(tr, 2, 2)
+}
+
+func TestDateToString(t *testing.T) {
+	f := func(date uint64, want string) {
+		t.Helper()
+
+		if got := dateToString(date); got != want {
+			t.Errorf("dateToString(%d) unexpected return value: got %q, want %q", date, got, want)
+		}
+	}
+
+	f(globalIndexDate, "[entire retention period]")
+	f(1, "1970-01-02")
+	f(10, "1970-01-11")
+}
+
+func TestTimeRangeString(t *testing.T) {
+	f := func(tr TimeRange, want string) {
+		t.Helper()
+
+		if got := tr.String(); got != want {
+			t.Errorf("TimeRange.String() unexpected return value: got %q, want %q", got, want)
+		}
+	}
+
+	f(globalIndexTimeRange, "[entire retention period]")
+	f(TimeRange{
+		MinTimestamp: 0,
+		MaxTimestamp: 1,
+	}, "[1970-01-01T00:00:00Z..1970-01-01T00:00:00.001Z]")
+	f(TimeRange{
+		MinTimestamp: 1,
+		MaxTimestamp: 2,
+	}, "[1970-01-01T00:00:00.001Z..1970-01-01T00:00:00.002Z]")
+	f(TimeRange{
+		MinTimestamp: time.Date(2024, 9, 6, 0, 0, 0, 000, time.UTC).UnixMilli(),
+		MaxTimestamp: time.Date(2024, 9, 7, 0, 0, 0, 000, time.UTC).UnixMilli() - 1,
+	}, "[2024-09-06T00:00:00Z..2024-09-06T23:59:59.999Z]")
+}