Repository: VictoriaMetrics/VictoriaMetrics
Pull request #7167 (2024-10-02 12:00:07)

vmui/logs: fix display of hits chart
### Describe Your Changes

Fixed the display of hits chart in VictoriaLogs.
See #7133

### Checklist

The following checks are **mandatory**:

- [ ] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).



Patch:
diff --git a/app/vmui/packages/vmui/src/pages/ExploreLogs/ExploreLogsBarChart/ExploreLogsBarChart.tsx b/app/vmui/packages/vmui/src/pages/ExploreLogs/ExploreLogsBarChart/ExploreLogsBarChart.tsx
index 506cf418d..326e9febd 100644
--- a/app/vmui/packages/vmui/src/pages/ExploreLogs/ExploreLogsBarChart/ExploreLogsBarChart.tsx
+++ b/app/vmui/packages/vmui/src/pages/ExploreLogs/ExploreLogsBarChart/ExploreLogsBarChart.tsx
@@ -1,4 +1,4 @@
-import React, { FC, useMemo } from "preact/compat";
+import React, { FC, useCallback, useMemo } from "preact/compat";
 import "./style.scss";
 import useDeviceDetect from "../../../hooks/useDeviceDetect";
 import classNames from "classnames";
@@ -10,6 +10,7 @@ import BarHitsChart from "../../../components/Chart/BarHitsChart/BarHitsChart";
 import Alert from "../../../components/Main/Alert/Alert";
 import { TimeParams } from "../../../types";
 import LineLoader from "../../../components/Main/LineLoader/LineLoader";
+import { getHitsTimeParams } from "../../../utils/logs";
 
 interface Props {
   query: string;
@@ -24,26 +25,43 @@ const ExploreLogsBarChart: FC<Props> = ({ logHits, period, error, isLoading, onA
   const { isMobile } = useDeviceDetect();
   const timeDispatch = useTimeDispatch();
 
-  const getXAxis = (timestamps: string[]): number[] => {
-    return (timestamps.map(t => t ? dayjs(t).unix() : null)
-      .filter(Boolean) as number[])
-      .sort((a, b) => a - b);
-  };
-
-  const getYAxes = (logHits: LogHits[], timestamps: string[]) => {
+  const getYAxes = (logHits: LogHits[], timestamps: number[]) => {
     return logHits.map(hits => {
-      return timestamps.map(t => {
-        const index = hits.timestamps.findIndex(ts => ts === t);
-        return index === -1 ? null : hits.values[index] || null;
+      const timestampValueMap = new Map();
+      hits.timestamps.forEach((ts, idx) => {
+        const unixTime = dayjs(ts).unix();
+        timestampValueMap.set(unixTime, hits.values[idx] || null);
       });
+
+      return timestamps.map(t => timestampValueMap.get(t) || null);
     });
   };
 
+  const generateTimestamps = useCallback((date: dayjs.Dayjs) => {
+    const result: number[] = [];
+    const { start, end, step } = getHitsTimeParams(period);
+    const stepsToFirstTimestamp = Math.ceil(start.diff(date, "milliseconds") / step);
+    let firstTimestamp = date.add(stepsToFirstTimestamp * step, "milliseconds");
+
+    // If the first timestamp is before 'start', set it to 'start'
+    if (firstTimestamp.isBefore(start)) {
+      firstTimestamp = start.clone();
+    }
+
+    // Calculate the total number of steps from 'firstTimestamp' to 'end'
+    const totalSteps = Math.floor(end.diff(firstTimestamp, "milliseconds") / step);
+
+    for (let i = 0; i <= totalSteps; i++) {
+      result.push(firstTimestamp.add(i * step, "milliseconds").unix());
+    }
+
+    return result;
+  }, [period]);
+
   const data = useMemo(() => {
     if (!logHits.length) return [[], []] as AlignedData;
-    const timestamps = Array.from(new Set(logHits.map(l => l.timestamps).flat()));
-    const xAxis = getXAxis(timestamps);
-    const yAxes = getYAxes(logHits, timestamps);
+    const xAxis = generateTimestamps(dayjs(logHits[0].timestamps[0]));
+    const yAxes = getYAxes(logHits, xAxis);
     return [xAxis, ...yAxes] as AlignedData;
   }, [logHits]);
 
diff --git a/app/vmui/packages/vmui/src/pages/ExploreLogs/hooks/useFetchLogHits.ts b/app/vmui/packages/vmui/src/pages/ExploreLogs/hooks/useFetchLogHits.ts
index af784db47..c46d48c76 100644
--- a/app/vmui/packages/vmui/src/pages/ExploreLogs/hooks/useFetchLogHits.ts
+++ b/app/vmui/packages/vmui/src/pages/ExploreLogs/hooks/useFetchLogHits.ts
@@ -2,9 +2,8 @@ import { useCallback, useMemo, useRef, useState } from "preact/compat";
 import { getLogHitsUrl } from "../../../api/logs";
 import { ErrorTypes, TimeParams } from "../../../types";
 import { LogHits } from "../../../api/types";
-import dayjs from "dayjs";
-import { LOGS_BARS_VIEW } from "../../../constants/logs";
 import { useSearchParams } from "react-router-dom";
+import { getHitsTimeParams } from "../../../utils/logs";
 
 export const useFetchLogHits = (server: string, query: string) => {
   const [searchParams] = useSearchParams();
@@ -17,10 +16,7 @@ export const useFetchLogHits = (server: string, query: string) => {
   const url = useMemo(() => getLogHitsUrl(server), [server]);
 
   const getOptions = (query: string, period: TimeParams, signal: AbortSignal) => {
-    const start = dayjs(period.start * 1000);
-    const end = dayjs(period.end * 1000);
-    const totalSeconds = end.diff(start, "milliseconds");
-    const step = Math.ceil(totalSeconds / LOGS_BARS_VIEW) || 1;
+    const { start, end, step } = getHitsTimeParams(period);
 
     return {
       signal,
diff --git a/app/vmui/packages/vmui/src/utils/logs.ts b/app/vmui/packages/vmui/src/utils/logs.ts
index 13e3d246c..bcf970837 100644
--- a/app/vmui/packages/vmui/src/utils/logs.ts
+++ b/app/vmui/packages/vmui/src/utils/logs.ts
@@ -1,4 +1,16 @@
+import { TimeParams } from "../types";
+import dayjs from "dayjs";
+import { LOGS_BARS_VIEW } from "../constants/logs";
+
 export const getStreamPairs = (value: string): string[] => {
   const pairs = /^{.+}$/.test(value) ? value.slice(1, -1).split(",") : [value];
   return pairs.filter(Boolean);
 };
+
+export const getHitsTimeParams = (period: TimeParams) => {
+  const start = dayjs(period.start * 1000);
+  const end = dayjs(period.end * 1000);
+  const totalSeconds = end.diff(start, "milliseconds");
+  const step = Math.ceil(totalSeconds / LOGS_BARS_VIEW) || 1;
+  return { start, end, step };
+};
diff --git a/deployment/docker/vmanomaly/vmanomaly-integration/docker-compose.yml b/deployment/docker/vmanomaly/vmanomaly-integration/docker-compose.yml
index 84d09b4ae..b90047a83 100644
--- a/deployment/docker/vmanomaly/vmanomaly-integration/docker-compose.yml
+++ b/deployment/docker/vmanomaly/vmanomaly-integration/docker-compose.yml
@@ -73,7 +73,7 @@ services:
     restart: always
   vmanomaly:
     container_name: vmanomaly
-    image: victoriametrics/vmanomaly:v1.16.3
+    image: victoriametrics/vmanomaly:v1.17.0
     depends_on:
       - "victoriametrics"
     ports:
diff --git a/docs/Troubleshooting.md b/docs/Troubleshooting.md
index 9c71b30c2..1dcffe051 100644
--- a/docs/Troubleshooting.md
+++ b/docs/Troubleshooting.md
@@ -129,9 +129,9 @@ If you see unexpected or unreliable query results from VictoriaMetrics, then try
    on the given `[start..end]` time range and check whether they are expected:
 
    ```sh
-   single-node: curl http://victoriametrics:8428/api/v1/export -d 'match[]=http_requests_total' -d 'start=...' -d 'end=...'
+   single-node: curl http://victoriametrics:8428/api/v1/export -d 'match[]=http_requests_total' -d 'start=...' -d 'end=...' -d 'reduce_mem_usage=1'
    
-   cluster: curl http://<vmselect>:8481/select/<tenantID>/prometheus/api/v1/export -d 'match[]=http_requests_total' -d 'start=...' -d 'end=...'
+   cluster: curl http://<vmselect>:8481/select/<tenantID>/prometheus/api/v1/export -d 'match[]=http_requests_total' -d 'start=...' -d 'end=...' -d 'reduce_mem_usage=1'
    ```
    Note that responses returned from [/api/v1/query](https://docs.victoriametrics.com/keyconcepts/#instant-query)
    and from [/api/v1/query_range](https://docs.victoriametrics.com/keyconcepts/#range-query) contain **evaluated** data
diff --git a/docs/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md
index a16a6056b..f9a40d7ca 100644
--- a/docs/VictoriaLogs/CHANGELOG.md
+++ b/docs/VictoriaLogs/CHANGELOG.md
@@ -15,7 +15,12 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
 
 ## tip
 
-* FEATURE: add basic [alerting rules](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/alerts-vlogs.yml) for VictoriaLogs process. See details at [monitoring docs](https://docs.victoriametrics.com/victorialogs/index.html#monitoring).
+* FEATURE: add basic [alerting rules](https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/deployment/docker/alerts-vlogs.yml) for VictoriaLogs process. See details at [monitoring docs](https://docs.victoriametrics.com/victorialogs/#monitoring).
+* FEATURE: improve [`stats` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#stats-pipe) performance on systems with many CPU cores when `by(...)` fields contain big number of unique values. For example, `_time:1d | stats by (user_id) count() x` should be executed much faster when `user_id` field contains millions of unique values.
+* FEATURE: improve performance for [`top`](https://docs.victoriametrics.com/victorialogs/logsql/#top-pipe), [`uniq`](https://docs.victoriametrics.com/victorialogs/logsql/#uniq-pipe) and [`field_values`](https://docs.victoriametrics.com/victorialogs/logsql/#field_values-pipe) pipes on systems with many CPU cores when it is applied to [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model) with big number of unique values. For example, `_time:1d | top 5 (user_id)` should be executed much faster when `user_id` field contains millions of unique values.
+* FEATURE: improve performance for [`field_names` pipe](https://docs.victoriametrics.com/victorialogs/logsql/#field_names-pipe) when it is applied to logs with hundreds of [log fields](https://docs.victoriametrics.com/victorialogs/keyconcepts/#data-model).
+
+* BUGFIX: [web UI](https://docs.victoriametrics.com/victorialogs/querying/#web-ui): fix display of hits chart. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7133).
 
 ## [v0.36.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.36.0-victorialogs)
 
diff --git a/docs/anomaly-detection/CHANGELOG.md b/docs/anomaly-detection/CHANGELOG.md
index f6e4d60dc..338519591 100644
--- a/docs/anomaly-detection/CHANGELOG.md
+++ b/docs/anomaly-detection/CHANGELOG.md
@@ -11,6 +11,29 @@ aliases:
 ---
 Please find the changelog for VictoriaMetrics Anomaly Detection below.
 
+## v1.17.0
+Released: 2024-10-17
+
+- FEATURE: Added `max_points_per_query` (global and [query-specific](https://docs.victoriametrics.com/anomaly-detection/components/reader/#per-query-parameters)) [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/reader/#vm-reader) arg to control query chunking. This overrides how `search.maxPointsPerTimeseries` flag (introduced in [v1.14.1](#v1141)) is used in `vmanomaly` for splitting long `fit_window` queries into smaller sub-intervals. This helps users avoid hitting the `search.maxQueryDuration` limit for individual queries by distributing initial query across multiple subquery requests with minimal overhead.
+
+- IMPROVEMENT: Enhanced the [self-monitoring](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#metrics-generated-by-vmanomaly) metrics for consistency across the components. Key changes include:
+  - Converted several [self-monitoring](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#metrics-generated-by-vmanomaly) metrics from `Summary` to `Histogram` to enable quantile calculation. This addresses the limitation of the `prometheus_client`'s [Summary](https://prometheus.github.io/client_python/instrumenting/summary/) implementation, which does not support quantiles. The change ensures metrics are more informative for performance analysis. Affected metrics are:
+    - `vmanomaly_reader_request_duration_seconds` ([VmReader](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#reader-behaviour-metrics))
+    - `vmanomaly_reader_response_parsing_seconds` ([VmReader](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#reader-behaviour-metrics))
+    - `vmanomaly_writer_request_duration_seconds` ([VmWriter](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#writer-behaviour-metrics))
+    - `vmanomaly_writer_request_serialize_seconds` ([VmWriter](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#writer-behaviour-metrics))
+  - Added a `query_key` label to the `vmanomaly_reader_response_parsing_seconds` [metric](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#reader-behaviour-metrics) to provide finer granularity in tracking the performance of individual queries. This metric has also been switched from `Summary` to `Histogram` to align with the other metrics and support quantile calculations.
+  - Added `preset` and `scheduler_alias` keys to [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#reader-behaviour-metrics) and [VmWriter](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#writer-behaviour-metrics) metrics for consistency in multi-[scheduler](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/) setups.
+  - Renamed [Counters](https://prometheus.io/docs/concepts/metric_types/#counter) `vmanomaly_reader_response_count` to `vmanomaly_reader_responses` and `vmanomaly_writer_response_count` to `vmanomaly_writer_responses`.
+  - Updated [docs](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#metrics-generated-by-vmanomaly) for better clarity.
+
+- IMPROVEMENT: Accelerated performance of model fitting stages on multicore systems.
+- IMPROVEMENT: Optimized query handling in multi-[scheduler](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/) setups by filtering [queries](https://docs.victoriametrics.com/anomaly-detection/components/models/#queries) for each scheduler based on model requirements. This reduces unnecessary data fetching from VictoriaMetrics, ensuring only relevant queries are processed by the [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/reader#vm-reader), leading to better performance and efficiency of configs with multiple active schedulers.
+
+- IMPROVEMENT: Implemented automatic cleanup of files in subdirectories within `/tmp` ([generated by the Stan backend](https://mc-stan.org/cmdstanpy/users-guide/outputs.html) when utilizing [Prophet](https://docs.victoriametrics.com/anomaly-detection/components/models/#prophet) models) after each `fit` operation. This prevents the accumulation of unused data over time in `/tmp`, addressing a potential issue where these files would only be deleted upon termination of the current Python session or service, leading to uncontrolled disk growth.
+
+- FIX: Re-enable the `vmanomaly_reader_response_count` (now called `vmanomaly_reader_responses`) self-monitoring [metric](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#reader-behaviour-metrics) for the [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/reader/#vm-reader), which was unintentionally disabled in previous releases and now updates correctly as intended.
+
 ## v1.16.3
 Released: 2024-10-08
 - IMPROVEMENT: Added `tls_cert_file` and `tls_key_file` arguments to support mTLS (mutual TLS) in `vmanomaly` components. This enhancement applies to the following components: [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/reader/#vm-reader), [VmWriter](https://docs.victoriametrics.com/anomaly-detection/components/writer/#vm-writer), and [Monitoring/Push](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#push-config-parameters). You can also use these arguments in conjunction with `verify_tls` when it is set as a path to a custom CA certificate file.
diff --git a/docs/anomaly-detection/FAQ.md b/docs/anomaly-detection/FAQ.md
index 773ead709..24b11806c 100644
--- a/docs/anomaly-detection/FAQ.md
+++ b/docs/anomaly-detection/FAQ.md
@@ -132,7 +132,7 @@ services:
   # ...
   vmanomaly:
     container_name: vmanomaly
-    image: victoriametrics/vmanomaly:v1.16.3
+    image: victoriametrics/vmanomaly:v1.17.0
     # ...
     ports:
       - "8490:8490"
@@ -225,6 +225,76 @@ P.s. `infer` data volume will remain the same for both models, so it does not af
 - Old: 8064 hours/week (fit) + 168 hours/week (infer)
 - New:    4 hours/week (fit) + 168 hours/week (infer)
 
+## Handling large queries in vmanomaly
+
+
+If you're dealing with a large query in the `queries` argument of [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/reader/#vm-reader) (especially when running [within a scheduler using a long](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/?highlight=fit_window#periodic-scheduler) `fit_window`), you may encounter issues such as query timeouts (due to the `search.maxQueryDuration` server limit) or rejections (if the `search.maxPointsPerTimeseries` server limit is exceeded). 
+
+We recommend upgrading to [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170), which introduced the `max_points_per_query` argument (both global and [query-specific](https://docs.victoriametrics.com/anomaly-detection/components/reader/#per-query-parameters)) for the [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/reader/#vm-reader). This argument overrides how `search.maxPointsPerTimeseries` flag handling (introduced in [v1.14.1](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1141)) is used in `vmanomaly` for splitting long `fit_window` queries into smaller sub-intervals. This helps users avoid hitting the `search.maxQueryDuration` limit for individual queries by distributing initial query across multiple subquery requests with minimal overhead.
+
+By splitting long `fit_window` queries into smaller sub-intervals, this helps avoid hitting the `search.maxQueryDuration` limit, distributing the load across multiple subquery requests with minimal overhead. To resolve the issue, reduce `max_points_per_query` to a value lower than `search.maxPointsPerTimeseries` until the problem is gone:
+
+```yaml
+reader:
+  # other reader args
+  max_points_per_query: 10000  # reader-level constraint
+  queries:
+    sum_alerts:
+      expr: 'sum(ALERTS{alertstate=~'(pending|firing)'}) by (alertstate)'
+      max_points_per_query: 5000  # query-level override
+models:
+    prophet:
+      # other model args
+      queries: [
+        'sum_alerts',
+      ]
+# other config sections
+```
+
+### Alternative workaround for older versions
+
+If upgrading is not an option, you can partially address the issue by splitting your large query into smaller ones using appropriate label filters:
+
+For example, such query
+
+```yaml
+reader:
+  # other reader args
+  queries:
+    sum_alerts:
+      expr: 'sum(ALERTS{alertstate=~'(pending|firing)'}) by (alertstate)'
+models:
+    prophet:
+      # other model args
+      queries: [
+        'sum_alerts',
+      ]
+# other config sections
+```
+
+can be modified to:
+
+```yaml
+reader:
+  # other reader args
+  queries:
+    sum_alerts_pending:
+      expr: 'sum(ALERTS{alertstate='pending'}) by ()'
+    sum_alerts_firing:
+      expr: 'sum(ALERTS{alertstate='firing'}) by ()'
+models:
+    prophet:
+      # other model args
+      queries: [
+        'sum_alerts_pending',
+        'sum_alerts_firing',
+      ]
+# other config sections
+```
+
+Please note that this approach may not fully resolve the issue if subqueries are not evenly distributed in terms of returned timeseries. Additionally, this workaround is not suitable for queries used in [multivariate models](https://docs.victoriametrics.com/anomaly-detection/components/models#multivariate-models) (especially when using the [groupby](https://docs.victoriametrics.com/anomaly-detection/components/models/#group-by) argument).
+
+
 ## Scaling vmanomaly
 > **Note:** As of latest release we do not support cluster or auto-scaled version yet (though, it's in our roadmap for - better backends, more parallelization, etc.), so proposed workarounds should be addressed *manually*.
 
diff --git a/docs/anomaly-detection/Overview.md b/docs/anomaly-detection/Overview.md
index 0db99ecfe..4c7856cec 100644
--- a/docs/anomaly-detection/Overview.md
+++ b/docs/anomaly-detection/Overview.md
@@ -229,7 +229,7 @@ This will expose metrics at `http://0.0.0.0:8080/metrics` page.
 To use *vmanomaly* you need to pull docker image:
 
 ```sh
-docker pull victoriametrics/vmanomaly:v1.16.3
+docker pull victoriametrics/vmanomaly:v1.17.0
 ```
 
 > Note: please check what is latest release in [CHANGELOG](https://docs.victoriametrics.com/anomaly-detection/changelog/)
@@ -239,7 +239,7 @@ docker pull victoriametrics/vmanomaly:v1.16.3
 You can put a tag on it for your convenience:
 
 ```sh
-docker image tag victoriametrics/vmanomaly:v1.16.3 vmanomaly
+docker image tag victoriametrics/vmanomaly:v1.17.0 vmanomaly
 ```
 Here is an example of how to run *vmanomaly* docker container with [license file](#licensing):
 
diff --git a/docs/anomaly-detection/components/models.md b/docs/anomaly-detection/components/models.md
index 4ebd669c1..56ae1b4cc 100644
--- a/docs/anomaly-detection/components/models.md
+++ b/docs/anomaly-detection/components/models.md
@@ -962,7 +962,7 @@ monitoring:
 Let's pull the docker image for `vmanomaly`:
 
 ```sh
-docker pull victoriametrics/vmanomaly:v1.16.3
+docker pull victoriametrics/vmanomaly:v1.17.0
 ```
 
 Now we can run the docker container putting as volumes both config and model file:
@@ -976,7 +976,7 @@ docker run -it \
 -v $(PWD)/license:/license \
 -v $(PWD)/custom_model.py:/vmanomaly/model/custom.py \
 -v $(PWD)/custom.yaml:/config.yaml \
-victoriametrics/vmanomaly:v1.16.3 /config.yaml \
+victoriametrics/vmanomaly:v1.17.0 /config.yaml \
 --licenseFile=/license
 ```
 
diff --git a/docs/anomaly-detection/components/monitoring.md b/docs/anomaly-detection/components/monitoring.md
index a2563b350..235e38f0d 100644
--- a/docs/anomaly-detection/components/monitoring.md
+++ b/docs/anomaly-detection/components/monitoring.md
@@ -11,6 +11,16 @@ aliases:
 ---
 There are 2 models to monitor VictoriaMetrics Anomaly Detection behavior - [push](https://docs.victoriametrics.com/keyconcepts/#push-model) and [pull](https://docs.victoriametrics.com/keyconcepts/#pull-model). Parameters for each of them should be specified in the config file, `monitoring` section.
 
+> **Note**: there was an enhancement of [self-monitoring](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#metrics-generated-by-vmanomaly) metrics for consistency across the components ([v.1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170)). Documentation was updated accordingly. Key changes included:
+- Converting several [self-monitoring](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#metrics-generated-by-vmanomaly) metrics from `Summary` to `Histogram` to enable quantile calculation. This addresses the limitation of the `prometheus_client`'s [Summary](https://prometheus.github.io/client_python/instrumenting/summary/) implementation, which does not support quantiles. The change ensures metrics are more informative for performance analysis. Affected metrics are:
+    - `vmanomaly_reader_request_duration_seconds` ([VmReader](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#reader-behaviour-metrics))
+    - `vmanomaly_reader_response_parsing_seconds` ([VmReader](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#reader-behaviour-metrics))
+    - `vmanomaly_writer_request_duration_seconds` ([VmWriter](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#writer-behaviour-metrics))
+    - `vmanomaly_writer_request_serialize_seconds` ([VmWriter](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#writer-behaviour-metrics))
+- Adding a `query_key` label to the `vmanomaly_reader_response_parsing_seconds` [metric](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#reader-behaviour-metrics) to provide finer granularity in tracking the performance of individual queries. This metric has also been switched from `Summary` to `Histogram` to align with the other metrics and support quantile calculations.
+- Adding `preset` and `scheduler_alias` keys to [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#reader-behaviour-metrics) and [VmWriter](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#writer-behaviour-metrics) metrics for consistency in multi-[scheduler](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/) setups.
+- Renaming [Counters](https://prometheus.io/docs/concepts/metric_types/#counter) `vmanomaly_reader_response_count` to `vmanomaly_reader_responses` and `vmanomaly_writer_response_count` to `vmanomaly_writer_responses`.
+
 ## Pull Model Config parameters
 
 <table class="params">
@@ -241,6 +251,8 @@ Label names [description](#labelnames)
 
 > **Note**: There is a new label key `model_alias` introduced in multi-model support [v1.10.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1100). This label key adjustment was made to preserve unique label set production during writing produced metrics back to VictoriaMetrics.
 
+> **Note**: as a part of [self-monitoring](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#metrics-generated-by-vmanomaly) metrics enchancement ([v.1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170)), new metrics, like `vmanomaly_model_run_errors`, was added. Some of them changed the type (`Summary` -> `Histogram`), like `vmanomaly_model_run_duration_seconds`.
+
 <table class="params">
     <thead>
         <tr>
@@ -256,11 +268,11 @@ Label names [description](#labelnames)
 
 `vmanomaly_model_runs`
             </td>
-            <td>Counter</td>
-            <td>How many times models ran (per model)</td>
+            <td>`Counter`</td>
+            <td>How many successful `stage` (`fit`, `infer`, `fit_infer`) runs occurred for models of class `model_alias` based on results from the `query_key` query, within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`stage, query_key, model_alias, scheduler_alias, preset`
+`stage`, `query_key`, `model_alias`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -268,11 +280,11 @@ Label names [description](#labelnames)
 
 `vmanomaly_model_run_duration_seconds`
             </td>
-            <td>Summary</td>
-            <td>How much time (in seconds) model invocations took</td>
+            <td>`Histogram` (was `Summary` prior to [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170)) </td>
+            <td>The total time (in seconds) taken by model invocations during the `stage` (`fit`, `infer`, `fit_infer`), based on the results of the `query_key` query, for models of class `model_alias`, within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`stage, query_key, model_alias, scheduler_alias, preset`
+`stage`, `query_key`, `model_alias`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -280,11 +292,11 @@ Label names [description](#labelnames)
 
 `vmanomaly_model_datapoints_accepted`
             </td>
-            <td>Counter</td>
-            <td>How many datapoints did models accept</td>
+            <td>`Counter`</td>
+            <td>The number of datapoints accepted (excluding NaN or Inf values) by models of class `model_alias` from the results of the `query_key` query during the `stage` (`infer`, `fit_infer`), within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`stage, query_key, model_alias, scheduler_alias, preset`
+`stage`, `query_key`, `model_alias`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -292,11 +304,11 @@ Label names [description](#labelnames)
 
 `vmanomaly_model_datapoints_produced`
             </td>
-            <td>Counter</td>
-            <td>How many datapoints were generated by models</td>
+            <td>`Counter`</td>
+            <td>The number of datapoints generated by models of class `model_alias` during the `stage` (`infer`, `fit_infer`) based on results from the `query_key` query, within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`stage, query_key, model_alias, scheduler_alias, preset`
+`stage`, `query_key`, `model_alias`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -304,11 +316,11 @@ Label names [description](#labelnames)
 
 `vmanomaly_models_active`
             </td>
-            <td>Gauge</td>
-            <td>How many models are currently inferring</td>
+            <td>`Gauge`</td>
+            <td>The number of model instances of class `model_alias` currently available for inference for the `query_key` query, within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`query_key, model_alias, scheduler_alias, preset`
+`query_key`, `model_alias`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -316,11 +328,22 @@ Label names [description](#labelnames)
 
 `vmanomaly_model_runs_skipped`
             </td>
-            <td>Counter</td>
-            <td>How many times a run was skipped (per model)</td>
+            <td>`Counter`</td>
+            <td>The number of times model runs (of class `model_alias`) were skipped in expected situations (e.g., no data for fitting/inference, or no new data to infer on) during the `stage` (`fit`, `infer`, `fit_infer`), based on results from the `query_key` query, within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`stage, query_key, model_alias, scheduler_alias, preset`
+`stage`, `query_key`, `model_alias`, `scheduler_alias`, `preset`
+            </td>
+        </tr>
+        <tr>
+            <td>
+
+`vmanomaly_model_run_errors`
+            </td>
+            <td>`Counter`</td>
+            <td>The number of times model runs (of class `model_alias`) failed due to internal service errors during the `stage` (`fit`, `infer`, `fit_infer`), based on results from the `query_key` query, within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
+            <td>
+`stage`, `query_key`, `model_alias`, `scheduler_alias`, `preset`
             </td>
         </tr>
     </tbody>
@@ -329,6 +352,8 @@ Label names [description](#labelnames)
 ### Writer Behaviour Metrics
 Label names [description](#labelnames)
 
+> **Note**: additional labels (`scheduler_alias`, `preset`) were added to writer and reader metrics in [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170) to improve consistency across the components. Also, metrics `vmanomaly_writer_request_duration_seconds` and `vmanomaly_writer_request_serialize_seconds` changed their type to `Histogram` (was `Summary` prior to [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170)).
+
 <table class="params">
     <thead>
         <tr>
@@ -344,23 +369,25 @@ Label names [description](#labelnames)
 
 `vmanomaly_writer_request_duration_seconds`
             </td>
-            <td>Summary</td>
-            <td>How much time (in seconds) did requests to VictoriaMetrics take</td>
+            <td>`Histogram` (was `Summary` prior to [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170))</td>
+            <td>The total time (in seconds) taken by write requests to VictoriaMetrics `url` for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.
+</td>
             <td>
 
-`url, query_key`
+`url`, `query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
             <td>
 
-`vmanomaly_writer_response_count`
+`vmanomaly_writer_responses` (named `vmanomaly_reader_response_count` prior to [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170))
             </td>
-            <td>Counter</td>
-            <td>Response code counts we got from VictoriaMetrics</td>
+            <td>`Counter`</td>
+            <td>The count of response codes received from VictoriaMetrics `url` for the `query_key` query, categorized by `code`, within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.
+</td>
             <td>
 
-`url, query_key, code`
+`url`, `code`, `query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -368,11 +395,11 @@ Label names [description](#labelnames)
 
 `vmanomaly_writer_sent_bytes`
             </td>
-            <td>Counter</td>
-            <td>How much bytes were sent to VictoriaMetrics</td>
+            <td>`Counter`</td>
+            <td>The total number of bytes sent to VictoriaMetrics `url` for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`url, query_key`
+`url`, `query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -380,11 +407,11 @@ Label names [description](#labelnames)
 
 `vmanomaly_writer_request_serialize_seconds`
             </td>
-            <td>Summary</td>
-            <td>How much time (in seconds) did serializing take</td>
+            <td>`Histogram` (was `Summary` prior to [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170))</td>
+            <td>The total time (in seconds) taken for serializing data for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`query_key`
+`query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -392,23 +419,21 @@ Label names [description](#labelnames)
 
 `vmanomaly_writer_datapoints_sent`
             </td>
-            <td>Counter</td>
-            <td>How many datapoints were sent to VictoriaMetrics</td>
+            <td>`Counter`</td>
+            <td>The total number of datapoints sent to VictoriaMetrics for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`query_key`
+`query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
             <td>
-
 `vmanomaly_writer_timeseries_sent`
             </td>
-            <td>Counter</td>
-            <td>How many timeseries were sent to VictoriaMetrics</td>
+            <td>`Counter`</td>
+            <td>The total number of timeseries sent to VictoriaMetrics for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
-
-`query_key`
+`query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
     </tbody>
@@ -432,23 +457,23 @@ Label names [description](#labelnames)
 
 `vmanomaly_reader_request_duration_seconds`
             </td>
-            <td>Summary</td>
-            <td>How much time (in seconds) did queries to VictoriaMetrics take</td>
+            <td>`Histogram` (was `Summary` prior to [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170))</td>
+            <td>The total time (in seconds) taken by queries to VictoriaMetrics `url` for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`url, query_key`
+`url`, `query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
             <td>
 
-`vmanomaly_reader_response_count`
+`vmanomaly_reader_responses` (named `vmanomaly_reader_response_count` prior to [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170))
             </td>
-            <td>Counter</td>
-            <td>Response code counts we got from VictoriaMetrics</td>
+            <td>`Counter`</td>
+            <td>The count of responses received from VictoriaMetrics `url` for the `query_key` query, categorized by `code`, within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`url, query_key, code`
+`url`, `query_key`, `code`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -456,11 +481,11 @@ Label names [description](#labelnames)
 
 `vmanomaly_reader_received_bytes`
             </td>
-            <td>Counter</td>
-            <td>How much bytes were received in responses</td>
+            <td>`Counter`</td>
+            <td>The total number of bytes received in responses for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`query_key`
+`query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -468,23 +493,23 @@ Label names [description](#labelnames)
 
 `vmanomaly_reader_response_parsing_seconds`
             </td>
-            <td>Summary</td>
-            <td>How much time (in seconds) did parsing take for each step</td>
+            <td>`Histogram` (was `Summary` prior to [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170))</td>
+            <td>The total time (in seconds) taken for data parsing at each `step` (json, dataframe) for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`step`
+`step`, `query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
             <td>
 
 `vmanomaly_reader_timeseries_received`
-</td>
-            <td>Counter</td>
-            <td>How many timeseries were received from VictoriaMetrics</td>
+            </td>
+            <td>`Counter`</td>
+            <td>The total number of timeseries received from VictoriaMetrics for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`query_key`
+`query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
         <tr>
@@ -492,11 +517,11 @@ Label names [description](#labelnames)
 
 `vmanomaly_reader_datapoints_received`
             </td>
-            <td>Counter</td>
-            <td>How many rows were received from VictoriaMetrics</td>
+            <td>`Counter`</td>
+            <td>The total number of datapoints received from VictoriaMetrics for the `query_key` query within the specified scheduler `scheduler_alias`, in the `vmanomaly` service running in `preset` mode.</td>
             <td>
 
-`query_key`
+`query_key`, `scheduler_alias`, `preset`
             </td>
         </tr>
     </tbody>
@@ -508,7 +533,7 @@ Label names [description](#labelnames)
 * `query_key` - query alias from [`reader`](https://docs.victoriametrics.com/anomaly-detection/components/reader/) config section.
 * `model_alias` - model alias from [`models`](https://docs.victoriametrics.com/anomaly-detection/components/models/) config section. **Introduced in [v1.10.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1100).**
 * `scheduler_alias` - scheduler alias from [`schedulers`](https://docs.victoriametrics.com/anomaly-detection/components/scheduler) config section. **Introduced in [v1.11.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1110).**
-* `preset` - preset alias for forthcoming `preset` section compatibility. **Introduced in [v1.12.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1120).**
+* `preset` - preset alias for [`preset`](https://docs.victoriametrics.com/anomaly-detection/presets/) mode of `vmanomaly`. **Introduced in [v1.12.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1120).**
 * `url` - writer or reader url endpoint.
 * `code` - response status code or `connection_error`, `timeout`.
 * `step` - json or dataframe reading step.
diff --git a/docs/anomaly-detection/components/reader.md b/docs/anomaly-detection/components/reader.md
index 8ef3b16f3..6288efd81 100644
--- a/docs/anomaly-detection/components/reader.md
+++ b/docs/anomaly-detection/components/reader.md
@@ -67,18 +67,22 @@ Starting from [v1.13.0](https://docs.victoriametrics.com/anomaly-detection/chang
   - **High anomaly scores** (>1) when the *data falls outside the expected range*, indicating a data constraint violation.
   - **Lowest anomaly scores** (=0) when the *model's predictions (`yhat`) fall outside the expected range*, meaning uncertain predictions.
 
+- `max_points_per_query` (int): Introduced in [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170), optional arg overrides how `search.maxPointsPerTimeseries` flag (available since [v1.14.1](#v1141)) impacts `vmanomaly` on splitting long `fit_window` [queries](https://docs.victoriametrics.com/anomaly-detection/components/reader/?highlight=queries#vm-reader) into smaller sub-intervals. This helps users avoid hitting the `search.maxQueryDuration` limit for individual queries by distributing initial query across multiple subquery requests with minimal overhead. Set less than `search.maxPointsPerTimeseries` if hitting `maxQueryDuration` limits. If set on a query-level, it overrides the global `max_points_per_query` (reader-level).
+
 
 ### Per-query config example
 ```yaml
 reader:
   class: 'vm'
   sampling_period: '1m'
+  max_points_per_query: 10000
   # other reader params ...
   queries:
     ingestion_rate:
       expr: 'sum(rate(vm_rows_inserted_total[5m])) by (type) > 0'
       step: '2m'  # overrides global `sampling_period` of 1m
       data_range: [10, 'inf']  # meaning only positive values > 10 are expected, i.e. a value `y` < 10 will trigger anomaly score > 1
+      max_points_per_query: 5000 # overrides reader-level value of 10000 for `ingestion_rate` query
 ```
 
 ### Config parameters
@@ -293,6 +297,17 @@ If True, then query will be performed from the last seen timestamp for a given s
 Introduced in [v1.15.1](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1151), it allows overriding the default `-search.latencyOffset` [flag of VictoriaMetrics](https://docs.victoriametrics.com/?highlight=search.latencyOffset#list-of-command-line-flags) (30s). The default value is set to 1ms, which should help in cases where `sampling_frequency` is low (10-60s) and `sampling_frequency` equals `infer_every` in the [PeriodicScheduler](https://docs.victoriametrics.com/anomaly-detection/components/scheduler/?highlight=infer_every#periodic-scheduler). This prevents users from receiving `service - WARNING - [Scheduler [scheduler_alias]] No data available for inference.` warnings in logs and allows for consecutive `infer` calls without gaps. To restore the old behavior, set it equal to your `-search.latencyOffset` [flag value]((https://docs.victoriametrics.com/?highlight=search.latencyOffset#list-of-command-line-flags)).
             </td>
         </tr>
+        <tr>
+            <td>
+`max_points_per_query`
+            </td>
+            <td>
+`10000`
+            </td>
+            <td>
+Introduced in [v1.17.0](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1170), optional arg overrides how `search.maxPointsPerTimeseries` flag (available since [v1.14.1](#v1141)) impacts `vmanomaly` on splitting long `fit_window` [queries](https://docs.victoriametrics.com/anomaly-detection/components/reader/?highlight=queries#vm-reader) into smaller sub-intervals. This helps users avoid hitting the `search.maxQueryDuration` limit for individual queries by distributing initial query across multiple subquery requests with minimal overhead. Set less than `search.maxPointsPerTimeseries` if hitting `maxQueryDuration` limits. You can also set it on [per-query](#per-query-parameters) basis to override this global one.
+            </td>
+        </tr>
     </tbody>
 </table>
 
@@ -313,16 +328,6 @@ reader:
   latency_offset: '1ms'
 ```
 
-<!-- ### mTLS protection
-
-Starting from [v1.16.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1163), `vmanomaly` supports [mTLS](https://en.wikipedia.org/wiki/Mutual_authentication) requests in its components, like [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/reader/#vm-reader), [VmWriter](https://docs.victoriametrics.com/anomaly-detection/components/writer/#vm-writer), and [Monitoring/Push](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#push-config-parameters) to query from and write to [VictoriaMetrics Enterprise, configured in the same mode](https://docs.victoriametrics.com/#mtls-protection).
-
-Please see the description of next arguments in a [config](#config-parameters):
-- `verify_tls` (if string, acts similar to `-mtlsCAFile` command line arg of VictoriaMetrics).
-- `tls_cert_file` (if given, acts similar to `-tlsCertFile` command line arg of VictoriaMetrics).
-- `tls_key_file` (if given, acts similar to `-tlsKeyFile` command line arg of VictoriaMetrics). -->
-
-
 ### mTLS protection
 
 As of [v1.16.3](https://docs.victoriametrics.com/anomaly-detection/changelog/#v1163), `vmanomaly` supports [mutual TLS (mTLS)](https://en.wikipedia.org/wiki/Mutual_authentication) for secure communication across its components, including [VmReader](https://docs.victoriametrics.com/anomaly-detection/components/reader/#vm-reader), [VmWriter](https://docs.victoriametrics.com/anomaly-detection/components/writer/#vm-writer), and [Monitoring/Push](https://docs.victoriametrics.com/anomaly-detection/components/monitoring/#push-config-parameters). This allows for mutual authentication between the client and server when querying or writing data to [VictoriaMetrics Enterprise, configured for mTLS](https://docs.victoriametrics.com/#mtls-protection).
diff --git a/docs/anomaly-detection/guides/guide-vmanomaly-vmalert/README.md b/docs/anomaly-detection/guides/guide-vmanomaly-vmalert/README.md
index 5df96ed2a..b9453434f 100644
--- a/docs/anomaly-detection/guides/guide-vmanomaly-vmalert/README.md
+++ b/docs/anomaly-detection/guides/guide-vmanomaly-vmalert/README.md
@@ -385,7 +385,7 @@ services:
     restart: always
   vmanomaly:
     container_name: vmanomaly
-    image: victoriametrics/vmanomaly:v1.16.3
+    image: victoriametrics/vmanomaly:v1.17.0
     depends_on:
       - "victoriametrics"
     ports:
diff --git a/docs/changelog/CHANGELOG.md b/docs/changelog/CHANGELOG.md
index e038d2a16..9ea85db95 100644
--- a/docs/changelog/CHANGELOG.md
+++ b/docs/changelog/CHANGELOG.md
@@ -33,6 +33,7 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 * BUGFIX: [vmalert-tool](https://docs.victoriametrics.com/vmalert-tool/): allow specifying empty labels list `labels: '{}'` in the same way as promtool does. This improves compatibility between these tools.
 * BUGFIX: [vmagent](https://docs.victoriametrics.com/vmagent/): support `m` unit for `minutes` duration in command-line flag `-streamAggr.dedupInterval`. Previously unit `m` wasn't supported correctly.
 * BUGFIX: [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): properly apply replication factor when storage node groups are used and replication factor is configured via global value such as `-replicationFactor=2`. Previously, global replication factor was ignored for storage node groups. See [these docs](https://docs.victoriametrics.com/cluster-victoriametrics/#vmstorage-groups-at-vmselect) for more information about storage groups configuration.
+* BUGFIX: `vmselect` in [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/): properly process response in [multi-level cluster setup](https://docs.victoriametrics.com/cluster-victoriametrics/#multi-level-cluster-setup). Before, vmselect could return no data in multi-level setup. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7270) for details. The issue was introduced in [v1.104.0](https://docs.victoriametrics.com/changelog/#v11040).
 
 ## [v1.104.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.104.0)
 
diff --git a/docs/vmagent.md b/docs/vmagent.md
index 6ced3b060..15334a0ae 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -502,15 +502,15 @@ and attaches `instance`, `job` and other target-specific labels to these metrics
   scrape_response_size_bytes > 10MiB
   ```
 
-* `scrape_samples_scraped` - the number of samples (aka metrics) parsed per each scrape. This allows detecting targets,
-  which expose too many metrics. For example, the following [MetricsQL query](https://docs.victoriametrics.com/metricsql/)
+* `scrape_samples_scraped` - the number of [samples](https://docs.victoriametrics.com/keyconcepts/#raw-samples) parsed per each scrape. This allows detecting targets,
+  which expose too many [series](https://docs.victoriametrics.com/keyconcepts/#time-series). For example, the following [MetricsQL query](https://docs.victoriametrics.com/metricsql/)
   returns targets, which expose more than 10000 metrics:
 
   ```metricsql
   scrape_samples_scraped > 10000
   ```
 
-* `scrape_samples_limit` - the configured limit on the number of metrics the given target can expose.
+* `scrape_samples_limit` - the configured limit on the number of [samples](https://docs.victoriametrics.com/keyconcepts/#raw-samples) the given target can expose.
   The limit can be set via `sample_limit` option at [scrape_configs](https://docs.victoriametrics.com/sd_configs/#scrape_configs).
   This metric is exposed only if the `sample_limit` is set. This allows detecting targets,
   which expose too many metrics compared to the configured `sample_limit`. For example, the following query
@@ -520,9 +520,9 @@ and attaches `instance`, `job` and other target-specific labels to these metrics
   scrape_samples_scraped / scrape_samples_limit > 0.8
   ```
 
-* `scrape_samples_post_metric_relabeling` - the number of samples (aka metrics) left after applying metric-level relabeling
+* `scrape_samples_post_metric_relabeling` - the number of [samples](https://docs.victoriametrics.com/keyconcepts/#raw-samples) left after applying metric-level relabeling
   from `metric_relabel_configs` section (see [relabeling docs](#relabeling) for more details).
-  This allows detecting targets with too many metrics after the relabeling.
+  This allows detecting targets with too many [series](https://docs.victoriametrics.com/keyconcepts/#time-series) after the relabeling.
   For example, the following [MetricsQL query](https://docs.victoriametrics.com/metricsql/) returns targets
   with more than 10000 metrics after the relabeling:
 
@@ -530,7 +530,7 @@ and attaches `instance`, `job` and other target-specific labels to these metrics
   scrape_samples_post_metric_relabeling > 10000
   ```
 
-* `scrape_series_added` - **an approximate** number of new series the given target generates during the current scrape.
+* `scrape_series_added` - **an approximate** number of new [series](https://docs.victoriametrics.com/keyconcepts/#time-series) the given target generates during the current scrape.
   This metric allows detecting targets (identified by `instance` label),
   which lead to [high churn rate](https://docs.victoriametrics.com/faq/#what-is-high-churn-rate).
   For example, the following [MetricsQL query](https://docs.victoriametrics.com/metricsql/) returns targets,
@@ -543,10 +543,10 @@ and attaches `instance`, `job` and other target-specific labels to these metrics
   `vmagent` sets `scrape_series_added` to zero when it runs with `-promscrape.noStaleMarkers` command-line flag
   or when it scrapes target with `no_stale_markers: true` option, e.g. when [staleness markers](#prometheus-staleness-markers) are disabled.
 
-* `scrape_series_limit` - the limit on the number of unique time series the given target can expose according to [these docs](#cardinality-limiter).
+* `scrape_series_limit` - the limit on the number of unique [series](https://docs.victoriametrics.com/keyconcepts/#time-series) the given target can expose according to [these docs](#cardinality-limiter).
   This metric is exposed only if the series limit is set.
 
-* `scrape_series_current` - the number of unique series the given target exposed so far.
+* `scrape_series_current` - the number of unique [series](https://docs.victoriametrics.com/keyconcepts/#time-series) the given target exposed so far.
   This metric is exposed only if the series limit is set according to [these docs](#cardinality-limiter).
   This metric allows alerting when the number of exposed series by the given target reaches the limit.
   For example, the following query would alert when the target exposes more than 90% of unique series compared to the configured limit.
@@ -556,7 +556,7 @@ and attaches `instance`, `job` and other target-specific labels to these metrics
   ```
 
 * `scrape_series_limit_samples_dropped` - exposes the number of dropped samples during the scrape because of the exceeded limit
-  on the number of unique series. This metric is exposed only if the series limit is set according to [these docs](#cardinality-limiter).
+  on the number of unique [series](https://docs.victoriametrics.com/keyconcepts/#time-series). This metric is exposed only if the series limit is set according to [these docs](#cardinality-limiter).
   This metric allows alerting when scraped samples are dropped because of the exceeded limit.
   For example, the following query alerts when at least a single sample is dropped because of the exceeded limit during the last hour:
 
diff --git a/lib/logstorage/block_result.go b/lib/logstorage/block_result.go
index 8eb2b1c72..76459953a 100644
--- a/lib/logstorage/block_result.go
+++ b/lib/logstorage/block_result.go
@@ -316,8 +316,8 @@ func (br *blockResult) initAllColumns() {
 	}
 
 	// Add other const columns
-	ccs := br.bs.getConstColumns()
-	for _, cc := range ccs {
+	csh := br.bs.getColumnsHeader()
+	for _, cc := range csh.constColumns {
 		if cc.Name == "" {
 			continue
 		}
@@ -327,7 +327,7 @@ func (br *blockResult) initAllColumns() {
 	}
 
 	// Add other non-const columns
-	chs := br.bs.getColumnHeaders()
+	chs := csh.columnHeaders
 	for i := range chs {
 		ch := &chs[i]
 		if ch.name == "" {
diff --git a/lib/logstorage/block_search.go b/lib/logstorage/block_search.go
index cd7804938..c932cbc8d 100644
--- a/lib/logstorage/block_search.go
+++ b/lib/logstorage/block_search.go
@@ -139,7 +139,7 @@ type blockSearch struct {
 
 	// cshCache is the columnsHeader associated with the given block.
 	//
-	// It is initialized lazily by calling getColumnsHeaderV0().
+	// It is initialized lazily by calling getColumnsHeader().
 	cshCache *columnsHeader
 
 	// seenStreams contains seen streamIDs for the recent searches.
@@ -240,7 +240,7 @@ func (bs *blockSearch) getConstColumnValue(name string) string {
 	}
 
 	if bs.partFormatVersion() < 1 {
-		csh := bs.getColumnsHeaderV0()
+		csh := bs.getColumnsHeader()
 		for _, cc := range csh.constColumns {
 			if cc.Name == name {
 				return cc.Value
@@ -288,7 +288,7 @@ func (bs *blockSearch) getColumnHeader(name string) *columnHeader {
 	}
 
 	if bs.partFormatVersion() < 1 {
-		csh := bs.getColumnsHeaderV0()
+		csh := bs.getColumnsHeader()
 		chs := csh.columnHeaders
 		for i := range chs {
 			ch := &chs[i]
@@ -337,48 +337,6 @@ func (bs *blockSearch) getColumnNameID(name string) (uint64, bool) {
 	return id, ok
 }
 
-func (bs *blockSearch) getColumnNameByID(id uint64) (string, bool) {
-	columnNames := bs.bsw.p.columnNames
-	if id >= uint64(len(columnNames)) {
-		return "", false
-	}
-	return columnNames[id], true
-}
-
-func (bs *blockSearch) getConstColumns() []Field {
-	if bs.partFormatVersion() < 1 {
-		csh := bs.getColumnsHeaderV0()
-		return csh.constColumns
-	}
-
-	chsIndex := bs.getColumnsHeaderIndex()
-	for _, cr := range chsIndex.constColumnsRefs {
-		columnName, ok := bs.getColumnNameByID(cr.columnNameID)
-		if !ok {
-			logger.Panicf("FATAL: %s: missing column name for id=%d", bs.bsw.p.path, cr.columnNameID)
-		}
-		_ = bs.getConstColumnValue(columnName)
-	}
-	return bs.ccsCache
-}
-
-func (bs *blockSearch) getColumnHeaders() []columnHeader {
-	if bs.partFormatVersion() < 1 {
-		csh := bs.getColumnsHeaderV0()
-		return csh.columnHeaders
-	}
-
-	chsIndex := bs.getColumnsHeaderIndex()
-	for _, cr := range chsIndex.columnHeadersRefs {
-		columnName, ok := bs.getColumnNameByID(cr.columnNameID)
-		if !ok {
-			logger.Panicf("FATAL: %s: missing column name for id=%d", bs.bsw.p.path, cr.columnNameID)
-		}
-		_ = bs.getColumnHeader(columnName)
-	}
-	return bs.chsCache
-}
-
 func (bs *blockSearch) getColumnsHeaderIndex() *columnsHeaderIndex {
 	if bs.partFormatVersion() < 1 {
 		logger.Panicf("BUG: getColumnsHeaderIndex() can be called only for part encoding v1+, while it has been called for v%d", bs.partFormatVersion())
@@ -395,18 +353,23 @@ func (bs *blockSearch) getColumnsHeaderIndex() *columnsHeaderIndex {
 	return bs.cshIndexCache
 }
 
-func (bs *blockSearch) getColumnsHeaderV0() *columnsHeader {
-	if bs.partFormatVersion() >= 1 {
-		logger.Panicf("BUG: getColumnsHeaderV0() can be called only for part encoding v0, while it has been called for v%d", bs.partFormatVersion())
-	}
-
+func (bs *blockSearch) getColumnsHeader() *columnsHeader {
 	if bs.cshCache == nil {
 		b := bs.getColumnsHeaderBlock()
 
-		bs.cshCache = getColumnsHeader()
-		if err := bs.cshCache.unmarshalNoArena(b, 0); err != nil {
+		csh := getColumnsHeader()
+		partFormatVersion := bs.partFormatVersion()
+		if err := csh.unmarshalNoArena(b, partFormatVersion); err != nil {
 			logger.Panicf("FATAL: %s: cannot unmarshal columns header: %s", bs.bsw.p.path, err)
 		}
+		if partFormatVersion >= 1 {
+			cshIndex := bs.getColumnsHeaderIndex()
+			if err := csh.setColumnNames(cshIndex, bs.bsw.p.columnNames); err != nil {
+				logger.Panicf("FATAL: %s: %s", bs.bsw.p.path, err)
+			}
+		}
+
+		bs.cshCache = csh
 	}
 	return bs.cshCache
 }
diff --git a/lib/logstorage/cache.go b/lib/logstorage/cache.go
new file mode 100644
index 000000000..1fde7dd1e
--- /dev/null
+++ b/lib/logstorage/cache.go
@@ -0,0 +1,83 @@
+package logstorage
+
+import (
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
+)
+
+type cache struct {
+	curr atomic.Pointer[sync.Map]
+	prev atomic.Pointer[sync.Map]
+
+	stopCh chan struct{}
+	wg     sync.WaitGroup
+}
+
+func newCache() *cache {
+	var c cache
+	c.curr.Store(&sync.Map{})
+	c.prev.Store(&sync.Map{})
+
+	c.stopCh = make(chan struct{})
+	c.wg.Add(1)
+	go func() {
+		defer c.wg.Done()
+		c.runCleaner()
+	}()
+	return &c
+}
+
+func (c *cache) MustStop() {
+	close(c.stopCh)
+	c.wg.Wait()
+}
+
+func (c *cache) runCleaner() {
+	d := timeutil.AddJitterToDuration(3 * time.Minute)
+	t := time.NewTicker(d)
+	defer t.Stop()
+	for {
+		select {
+		case <-t.C:
+			c.clean()
+		case <-c.stopCh:
+			return
+		}
+	}
+}
+
+func (c *cache) clean() {
+	curr := c.curr.Load()
+	c.prev.Store(curr)
+	c.curr.Store(&sync.Map{})
+}
+
+func (c *cache) Get(k []byte) (any, bool) {
+	kStr := bytesutil.ToUnsafeString(k)
+
+	curr := c.curr.Load()
+	v, ok := curr.Load(kStr)
+	if ok {
+		return v, true
+	}
+
+	prev := c.prev.Load()
+	v, ok = prev.Load(kStr)
+	if ok {
+		kStr = strings.Clone(kStr)
+		curr.Store(kStr, v)
+		return v, true
+	}
+	return nil, false
+}
+
+func (c *cache) Set(k []byte, v any) {
+	kStr := string(k)
+	curr := c.curr.Load()
+	curr.Store(kStr, v)
+}
diff --git a/lib/logstorage/filenames.go b/lib/logstorage/filenames.go
index 909286ba2..ff3458e97 100644
--- a/lib/logstorage/filenames.go
+++ b/lib/logstorage/filenames.go
@@ -17,10 +17,7 @@ const (
 	metadataFilename = "metadata.json"
 	partsFilename    = "parts.json"
 
-	streamIDCacheFilename = "stream_id.bin"
-
 	indexdbDirname    = "indexdb"
 	datadbDirname     = "datadb"
-	cacheDirname      = "cache"
 	partitionsDirname = "partitions"
 )
diff --git a/lib/logstorage/indexdb.go b/lib/logstorage/indexdb.go
index fd90ac108..811d2b536 100644
--- a/lib/logstorage/indexdb.go
+++ b/lib/logstorage/indexdb.go
@@ -500,13 +500,14 @@ func (idb *indexdb) marshalStreamFilterCacheKey(dst []byte, tenantIDs []TenantID
 func (idb *indexdb) loadStreamIDsFromCache(tenantIDs []TenantID, sf *StreamFilter) ([]streamID, bool) {
 	bb := bbPool.Get()
 	bb.B = idb.marshalStreamFilterCacheKey(bb.B[:0], tenantIDs, sf)
-	data := idb.s.filterStreamCache.GetBig(nil, bb.B)
+	v, ok := idb.s.filterStreamCache.Get(bb.B)
 	bbPool.Put(bb)
-	if len(data) == 0 {
+	if !ok {
 		// Cache miss
 		return nil, false
 	}
 	// Cache hit - unpack streamIDs from data.
+	data := *(v.(*[]byte))
 	n, nSize := encoding.UnmarshalVarUint64(data)
 	if nSize <= 0 {
 		logger.Panicf("BUG: unexpected error when unmarshaling the number of streamIDs from cache")
@@ -537,7 +538,7 @@ func (idb *indexdb) storeStreamIDsToCache(tenantIDs []TenantID, sf *StreamFilter
 	// Store marshaled streamIDs to cache.
 	bb := bbPool.Get()
 	bb.B = idb.marshalStreamFilterCacheKey(bb.B[:0], tenantIDs, sf)
-	idb.s.filterStreamCache.SetBig(bb.B, b)
+	idb.s.filterStreamCache.Set(bb.B, &b)
 	bbPool.Put(bb)
 }
 
diff --git a/lib/logstorage/partition.go b/lib/logstorage/partition.go
index 439ca002f..cf3fa9b23 100644
--- a/lib/logstorage/partition.go
+++ b/lib/logstorage/partition.go
@@ -1,7 +1,6 @@
 package logstorage
 
 import (
-	"bytes"
 	"path/filepath"
 	"sort"
 
@@ -160,20 +159,18 @@ func (pt *partition) logIngestedRows(lr *LogRows) {
 }
 
 func (pt *partition) hasStreamIDInCache(sid *streamID) bool {
-	var result [1]byte
-
 	bb := bbPool.Get()
 	bb.B = pt.marshalStreamIDCacheKey(bb.B, sid)
-	value := pt.s.streamIDCache.Get(result[:0], bb.B)
+	_, ok := pt.s.streamIDCache.Get(bb.B)
 	bbPool.Put(bb)
 
-	return bytes.Equal(value, okValue)
+	return ok
 }
 
 func (pt *partition) putStreamIDToCache(sid *streamID) {
 	bb := bbPool.Get()
 	bb.B = pt.marshalStreamIDCacheKey(bb.B, sid)
-	pt.s.streamIDCache.Set(bb.B, okValue)
+	pt.s.streamIDCache.Set(bb.B, nil)
 	bbPool.Put(bb)
 }
 
@@ -183,8 +180,6 @@ func (pt *partition) marshalStreamIDCacheKey(dst []byte, sid *streamID) []byte {
 	return dst
 }
 
-var okValue = []byte("1")
-
 // debugFlush makes sure that all the recently ingested data data becomes searchable
 func (pt *partition) debugFlush() {
 	pt.ddb.debugFlush()
diff --git a/lib/logstorage/pipe_field_names.go b/lib/logstorage/pipe_field_names.go
index 1cc1b79cf..8d6e393fb 100644
--- a/lib/logstorage/pipe_field_names.go
+++ b/lib/logstorage/pipe_field_names.go
@@ -103,8 +103,8 @@ func (pfp *pipeFieldNamesProcessor) writeBlock(workerID uint, br *blockResult) {
 
 	cs := br.getColumns()
 	for _, c := range cs {
-		pHits, ok := m[c.name]
-		if !ok {
+		pHits := m[c.name]
+		if pHits == nil {
 			nameCopy := strings.Clone(c.name)
 			hits := uint64(0)
 			pHits = &hits
@@ -128,8 +128,8 @@ func (pfp *pipeFieldNamesProcessor) flush() error {
 	shards = shards[1:]
 	for i := range shards {
 		for name, pHitsSrc := range shards[i].getM() {
-			pHits, ok := m[name]
-			if !ok {
+			pHits := m[name]
+			if pHits == nil {
 				m[name] = pHitsSrc
 			} else {
 				*pHits += *pHitsSrc
diff --git a/lib/logstorage/pipe_stats.go b/lib/logstorage/pipe_stats.go
index 0f44f2403..7cf723a1b 100644
--- a/lib/logstorage/pipe_stats.go
+++ b/lib/logstorage/pipe_stats.go
@@ -3,9 +3,12 @@ package logstorage
 import (
 	"fmt"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"unsafe"
 
+	"github.com/cespare/xxhash/v2"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
@@ -425,40 +428,38 @@ func (psp *pipeStatsProcessor) flush() error {
 		return fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", psp.ps.String(), psp.maxStateSize/(1<<20))
 	}
 
-	// Merge states across shards
-	shards := psp.shards
-	shardMain := &shards[0]
-	shardMain.init()
-	m := shardMain.m
-	shards = shards[1:]
-	for i := range shards {
-		shard := &shards[i]
-		for key, psg := range shard.m {
-			// shard.m may be quite big, so this loop can take a lot of time and CPU.
-			// Stop processing data as soon as stopCh is closed without wasting additional CPU time.
-			if needStop(psp.stopCh) {
-				return nil
-			}
-
-			spgBase := m[key]
-			if spgBase == nil {
-				m[key] = psg
-			} else {
-				for i, sfp := range spgBase.sfps {
-					sfp.mergeState(psg.sfps[i])
-				}
-			}
-		}
+	// Merge states across shards in parallel
+	ms, err := psp.mergeShardsParallel()
+	if err != nil {
+		return err
+	}
+	if needStop(psp.stopCh) {
+		return nil
 	}
 
-	// Write per-group states to ppNext
-	byFields := psp.ps.byFields
-	if len(byFields) == 0 && len(m) == 0 {
+	if len(psp.ps.byFields) == 0 && len(ms) == 0 {
 		// Special case - zero matching rows.
-		_ = shardMain.getPipeStatsGroup(nil)
-		m = shardMain.m
+		psp.shards[0].init()
+		_ = psp.shards[0].getPipeStatsGroup(nil)
+		ms = append(ms, psp.shards[0].m)
 	}
 
+	// Write the calculated stats in parallel to the next pipe.
+	var wg sync.WaitGroup
+	for i, m := range ms {
+		wg.Add(1)
+		go func(workerID uint) {
+			defer wg.Done()
+			psp.writeShardData(workerID, m)
+		}(uint(i))
+	}
+	wg.Wait()
+
+	return nil
+}
+
+func (psp *pipeStatsProcessor) writeShardData(workerID uint, m map[string]*pipeStatsGroup) {
+	byFields := psp.ps.byFields
 	rcs := make([]resultColumn, 0, len(byFields)+len(psp.ps.funcs))
 	for _, bf := range byFields {
 		rcs = appendResultColumnWithName(rcs, bf.name)
@@ -475,7 +476,7 @@ func (psp *pipeStatsProcessor) flush() error {
 		// m may be quite big, so this loop can take a lot of time and CPU.
 		// Stop processing data as soon as stopCh is closed without wasting additional CPU time.
 		if needStop(psp.stopCh) {
-			return nil
+			return
 		}
 
 		// Unmarshal values for byFields from key.
@@ -511,7 +512,7 @@ func (psp *pipeStatsProcessor) flush() error {
 		if valuesLen >= 1_000_000 {
 			br.setResultColumns(rcs, rowsCount)
 			rowsCount = 0
-			psp.ppNext.writeBlock(0, &br)
+			psp.ppNext.writeBlock(workerID, &br)
 			br.reset()
 			for i := range rcs {
 				rcs[i].resetValues()
@@ -521,9 +522,134 @@ func (psp *pipeStatsProcessor) flush() error {
 	}
 
 	br.setResultColumns(rcs, rowsCount)
-	psp.ppNext.writeBlock(0, &br)
+	psp.ppNext.writeBlock(workerID, &br)
+}
 
-	return nil
+func (psp *pipeStatsProcessor) mergeShardsParallel() ([]map[string]*pipeStatsGroup, error) {
+	shards := psp.shards
+	shardsLen := len(shards)
+
+	if shardsLen == 1 {
+		var ms []map[string]*pipeStatsGroup
+		shards[0].init()
+		if len(shards[0].m) > 0 {
+			ms = append(ms, shards[0].m)
+		}
+		return ms, nil
+	}
+
+	var wg sync.WaitGroup
+	perShardMaps := make([][]map[string]*pipeStatsGroup, shardsLen)
+	for i := range shards {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+
+			shardMaps := make([]map[string]*pipeStatsGroup, shardsLen)
+			for i := range shardMaps {
+				shardMaps[i] = make(map[string]*pipeStatsGroup)
+			}
+
+			shards[idx].init()
+
+			n := int64(0)
+			nTotal := int64(0)
+			for k, psg := range shards[idx].m {
+				if needStop(psp.stopCh) {
+					return
+				}
+				h := xxhash.Sum64(bytesutil.ToUnsafeBytes(k))
+				m := shardMaps[h%uint64(len(shardMaps))]
+				n += updatePipeStatsMap(m, k, psg)
+				if n > stateSizeBudgetChunk {
+					if nRemaining := psp.stateSizeBudget.Add(-n); nRemaining < 0 {
+						return
+					}
+					nTotal += n
+					n = 0
+				}
+			}
+			nTotal += n
+			psp.stateSizeBudget.Add(-n)
+
+			perShardMaps[idx] = shardMaps
+
+			// Clean the original map and return its state size budget back.
+			shards[idx].m = nil
+			psp.stateSizeBudget.Add(nTotal)
+		}(i)
+	}
+	wg.Wait()
+	if needStop(psp.stopCh) {
+		return nil, nil
+	}
+	if n := psp.stateSizeBudget.Load(); n < 0 {
+		return nil, fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", psp.ps.String(), psp.maxStateSize/(1<<20))
+	}
+
+	// Merge per-shard entries into perShardMaps[0]
+	for i := range perShardMaps {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+
+			m := perShardMaps[0][idx]
+			for i := 1; i < len(perShardMaps); i++ {
+				n := int64(0)
+				nTotal := int64(0)
+				for k, psg := range perShardMaps[i][idx] {
+					if needStop(psp.stopCh) {
+						return
+					}
+					n += updatePipeStatsMap(m, k, psg)
+					if n > stateSizeBudgetChunk {
+						if nRemaining := psp.stateSizeBudget.Add(-n); nRemaining < 0 {
+							return
+						}
+						nTotal += n
+						n = 0
+					}
+				}
+				nTotal += n
+				psp.stateSizeBudget.Add(-n)
+
+				// Clean the original map and return its state size budget back.
+				perShardMaps[i][idx] = nil
+				psp.stateSizeBudget.Add(nTotal)
+			}
+		}(i)
+	}
+	wg.Wait()
+	if needStop(psp.stopCh) {
+		return nil, nil
+	}
+	if n := psp.stateSizeBudget.Load(); n < 0 {
+		return nil, fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", psp.ps.String(), psp.maxStateSize/(1<<20))
+	}
+
+	// Filter out maps without entries
+	ms := perShardMaps[0]
+	result := ms[:0]
+	for _, m := range ms {
+		if len(m) > 0 {
+			result = append(result, m)
+		}
+	}
+
+	return result, nil
+}
+
+func updatePipeStatsMap(m map[string]*pipeStatsGroup, k string, psgSrc *pipeStatsGroup) int64 {
+	psgDst := m[k]
+	if psgDst != nil {
+		for i, sfp := range psgDst.sfps {
+			sfp.mergeState(psgSrc.sfps[i])
+		}
+		return 0
+	}
+
+	m[k] = psgSrc
+	return int64(unsafe.Sizeof(k) + unsafe.Sizeof(psgSrc))
 }
 
 func parsePipeStats(lex *lexer, needStatsKeyword bool) (*pipeStats, error) {
diff --git a/lib/logstorage/pipe_top.go b/lib/logstorage/pipe_top.go
index 1b8677934..df1f59a3c 100644
--- a/lib/logstorage/pipe_top.go
+++ b/lib/logstorage/pipe_top.go
@@ -1,13 +1,17 @@
 package logstorage
 
 import (
+	"container/heap"
 	"fmt"
 	"slices"
 	"sort"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"unsafe"
 
+	"github.com/cespare/xxhash/v2"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
@@ -197,8 +201,8 @@ func (shard *pipeTopProcessorShard) writeBlock(br *blockResult) {
 
 func (shard *pipeTopProcessorShard) updateState(v string, hits uint64) {
 	m := shard.getM()
-	pHits, ok := m[v]
-	if !ok {
+	pHits := m[v]
+	if pHits == nil {
 		vCopy := strings.Clone(v)
 		hits := uint64(0)
 		pHits = &hits
@@ -244,42 +248,13 @@ func (ptp *pipeTopProcessor) flush() error {
 		return fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", ptp.pt.String(), ptp.maxStateSize/(1<<20))
 	}
 
-	// merge state across shards
-	shards := ptp.shards
-	m := shards[0].getM()
-	shards = shards[1:]
-	for i := range shards {
-		if needStop(ptp.stopCh) {
-			return nil
-		}
-
-		for k, pHitsSrc := range shards[i].getM() {
-			pHits, ok := m[k]
-			if !ok {
-				m[k] = pHitsSrc
-			} else {
-				*pHits += *pHitsSrc
-			}
-		}
-	}
-
-	// select top entries with the biggest number of hits
-	entries := make([]pipeTopEntry, 0, len(m))
-	for k, pHits := range m {
-		entries = append(entries, pipeTopEntry{
-			k:    k,
-			hits: *pHits,
-		})
+	// merge state across shards in parallel
+	entries, err := ptp.mergeShardsParallel()
+	if err != nil {
+		return err
 	}
-	sort.Slice(entries, func(i, j int) bool {
-		a, b := &entries[i], &entries[j]
-		if a.hits == b.hits {
-			return a.k < b.k
-		}
-		return a.hits > b.hits
-	})
-	if uint64(len(entries)) > ptp.pt.limit {
-		entries = entries[:ptp.pt.limit]
+	if needStop(ptp.stopCh) {
+		return nil
 	}
 
 	// write result
@@ -373,11 +348,208 @@ func (ptp *pipeTopProcessor) flush() error {
 	return nil
 }
 
+func (ptp *pipeTopProcessor) mergeShardsParallel() ([]*pipeTopEntry, error) {
+	limit := ptp.pt.limit
+	if limit == 0 {
+		return nil, nil
+	}
+
+	shards := ptp.shards
+	shardsLen := len(shards)
+	if shardsLen == 1 {
+		entries := getTopEntries(shards[0].getM(), limit, ptp.stopCh)
+		return entries, nil
+	}
+
+	var wg sync.WaitGroup
+	perShardMaps := make([][]map[string]*uint64, shardsLen)
+	for i := range shards {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+
+			shardMaps := make([]map[string]*uint64, shardsLen)
+			for i := range shardMaps {
+				shardMaps[i] = make(map[string]*uint64)
+			}
+
+			n := int64(0)
+			nTotal := int64(0)
+			for k, pHits := range shards[idx].getM() {
+				if needStop(ptp.stopCh) {
+					return
+				}
+				h := xxhash.Sum64(bytesutil.ToUnsafeBytes(k))
+				m := shardMaps[h%uint64(len(shardMaps))]
+				n += updatePipeTopMap(m, k, pHits)
+				if n > stateSizeBudgetChunk {
+					if nRemaining := ptp.stateSizeBudget.Add(-n); nRemaining < 0 {
+						return
+					}
+					nTotal += n
+					n = 0
+				}
+			}
+			nTotal += n
+			ptp.stateSizeBudget.Add(-n)
+
+			perShardMaps[idx] = shardMaps
+
+			// Clean the original map and return its state size budget back.
+			shards[idx].m = nil
+			ptp.stateSizeBudget.Add(nTotal)
+		}(i)
+	}
+	wg.Wait()
+	if needStop(ptp.stopCh) {
+		return nil, nil
+	}
+	if n := ptp.stateSizeBudget.Load(); n < 0 {
+		return nil, fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", ptp.pt.String(), ptp.maxStateSize/(1<<20))
+	}
+
+	// Obtain topN entries per each shard
+	entriess := make([][]*pipeTopEntry, shardsLen)
+	for i := range entriess {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+
+			m := perShardMaps[0][idx]
+			for i := 1; i < len(perShardMaps); i++ {
+				n := int64(0)
+				nTotal := int64(0)
+				for k, pHits := range perShardMaps[i][idx] {
+					if needStop(ptp.stopCh) {
+						return
+					}
+					n += updatePipeTopMap(m, k, pHits)
+					if n > stateSizeBudgetChunk {
+						if nRemaining := ptp.stateSizeBudget.Add(-n); nRemaining < 0 {
+							return
+						}
+						nTotal += n
+						n = 0
+					}
+				}
+				nTotal += n
+				ptp.stateSizeBudget.Add(-n)
+
+				// Clean the original map and return its state size budget back.
+				perShardMaps[i][idx] = nil
+				ptp.stateSizeBudget.Add(nTotal)
+			}
+			perShardMaps[0][idx] = nil
+
+			entriess[idx] = getTopEntries(m, ptp.pt.limit, ptp.stopCh)
+		}(i)
+	}
+	wg.Wait()
+	if needStop(ptp.stopCh) {
+		return nil, nil
+	}
+	if n := ptp.stateSizeBudget.Load(); n < 0 {
+		return nil, fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", ptp.pt.String(), ptp.maxStateSize/(1<<20))
+	}
+
+	// merge entriess
+	entries := entriess[0]
+	for _, es := range entriess[1:] {
+		entries = append(entries, es...)
+	}
+	sort.Slice(entries, func(i, j int) bool {
+		return entries[j].less(entries[i])
+	})
+	if uint64(len(entries)) > limit {
+		entries = entries[:limit]
+	}
+	return entries, nil
+}
+
+func getTopEntries(m map[string]*uint64, limit uint64, stopCh <-chan struct{}) []*pipeTopEntry {
+	if limit == 0 {
+		return nil
+	}
+
+	var eh topEntriesHeap
+	for k, pHits := range m {
+		if needStop(stopCh) {
+			return nil
+		}
+
+		e := pipeTopEntry{
+			k:    k,
+			hits: *pHits,
+		}
+		if uint64(len(eh)) < limit {
+			eCopy := e
+			heap.Push(&eh, &eCopy)
+			continue
+		}
+		if eh[0].less(&e) {
+			eCopy := e
+			eh[0] = &eCopy
+			heap.Fix(&eh, 0)
+		}
+	}
+
+	result := ([]*pipeTopEntry)(eh)
+	for len(eh) > 0 {
+		x := heap.Pop(&eh)
+		result[len(eh)] = x.(*pipeTopEntry)
+	}
+
+	return result
+}
+
+func updatePipeTopMap(m map[string]*uint64, k string, pHitsSrc *uint64) int64 {
+	pHitsDst := m[k]
+	if pHitsDst != nil {
+		*pHitsDst += *pHitsSrc
+		return 0
+	}
+
+	m[k] = pHitsSrc
+	return int64(unsafe.Sizeof(k) + unsafe.Sizeof(pHitsSrc))
+}
+
+type topEntriesHeap []*pipeTopEntry
+
+func (h *topEntriesHeap) Less(i, j int) bool {
+	a := *h
+	return a[i].less(a[j])
+}
+func (h *topEntriesHeap) Swap(i, j int) {
+	a := *h
+	a[i], a[j] = a[j], a[i]
+}
+func (h *topEntriesHeap) Len() int {
+	return len(*h)
+}
+func (h *topEntriesHeap) Push(v any) {
+	x := v.(*pipeTopEntry)
+	*h = append(*h, x)
+}
+func (h *topEntriesHeap) Pop() any {
+	a := *h
+	x := a[len(a)-1]
+	a[len(a)-1] = nil
+	*h = a[:len(a)-1]
+	return x
+}
+
 type pipeTopEntry struct {
 	k    string
 	hits uint64
 }
 
+func (e *pipeTopEntry) less(r *pipeTopEntry) bool {
+	if e.hits == r.hits {
+		return e.k > r.k
+	}
+	return e.hits < r.hits
+}
+
 type pipeTopWriteContext struct {
 	ptp *pipeTopProcessor
 	rcs []resultColumn
diff --git a/lib/logstorage/pipe_uniq.go b/lib/logstorage/pipe_uniq.go
index dd199d1cf..ddf043c58 100644
--- a/lib/logstorage/pipe_uniq.go
+++ b/lib/logstorage/pipe_uniq.go
@@ -4,9 +4,12 @@ import (
 	"fmt"
 	"slices"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"unsafe"
 
+	"github.com/cespare/xxhash/v2"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/encoding"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
@@ -214,8 +217,8 @@ func (shard *pipeUniqProcessorShard) writeBlock(br *blockResult) bool {
 
 func (shard *pipeUniqProcessorShard) updateState(v string, hits uint64) {
 	m := shard.getM()
-	pHits, ok := m[v]
-	if !ok {
+	pHits := m[v]
+	if pHits == nil {
 		vCopy := strings.Clone(v)
 		hits := uint64(0)
 		pHits = &hits
@@ -263,32 +266,64 @@ func (pup *pipeUniqProcessor) flush() error {
 		return fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", pup.pu.String(), pup.maxStateSize/(1<<20))
 	}
 
-	// merge state across shards
-	shards := pup.shards
-	m := shards[0].getM()
-	shards = shards[1:]
-	for i := range shards {
-		if needStop(pup.stopCh) {
-			return nil
-		}
+	// merge state across shards in parallel
+	ms, err := pup.mergeShardsParallel()
+	if err != nil {
+		return err
+	}
+	if needStop(pup.stopCh) {
+		return nil
+	}
 
-		for k, pHitsSrc := range shards[i].getM() {
-			pHits, ok := m[k]
-			if !ok {
-				m[k] = pHitsSrc
-			} else {
-				*pHits += *pHitsSrc
+	resetHits := false
+	if limit := pup.pu.limit; limit > 0 {
+		// Trim the number of entries according to the given limit
+		entriesLen := 0
+		result := ms[:0]
+		for _, m := range ms {
+			entriesLen += len(m)
+			if uint64(entriesLen) <= limit {
+				result = append(result, m)
+				continue
 			}
+
+			// There is little sense in returning partial hits when the limit on the number of unique entries is reached,
+			// since arbitrary number of unique entries and hits for these entries could be skipped.
+			// It is better to return zero hits instead of misleading hits results.
+			resetHits = true
+			for k := range m {
+				delete(m, k)
+				entriesLen--
+				if uint64(entriesLen) <= limit {
+					break
+				}
+			}
+			if len(m) > 0 {
+				result = append(result, m)
+			}
+			break
 		}
+		ms = result
 	}
 
-	// There is little sense in returning partial hits when the limit on the number of unique entries is reached.
-	// It is better from UX experience is to return zero hits instead.
-	resetHits := pup.pu.limit > 0 && uint64(len(m)) > pup.pu.limit
+	// Write the calculated stats in parallel to the next pipe.
+	var wg sync.WaitGroup
+	for i, m := range ms {
+		wg.Add(1)
+		go func(workerID uint) {
+			defer wg.Done()
+			pup.writeShardData(workerID, m, resetHits)
+		}(uint(i))
+	}
+	wg.Wait()
 
-	// write result
+	return nil
+}
+
+func (pup *pipeUniqProcessor) writeShardData(workerID uint, m map[string]*uint64, resetHits bool) {
 	wctx := &pipeUniqWriteContext{
-		pup: pup,
+		workerID: workerID,
+		pup:      pup,
 	}
 	byFields := pup.pu.byFields
 	var rowFields []Field
@@ -311,7 +346,7 @@ func (pup *pipeUniqProcessor) flush() error {
 	if len(byFields) == 0 {
 		for k, pHits := range m {
 			if needStop(pup.stopCh) {
-				return nil
+				return
 			}
 
 			rowFields = rowFields[:0]
@@ -341,7 +376,7 @@ func (pup *pipeUniqProcessor) flush() error {
 		fieldName := byFields[0]
 		for k, pHits := range m {
 			if needStop(pup.stopCh) {
-				return nil
+				return
 			}
 
 			rowFields = append(rowFields[:0], Field{
@@ -354,7 +389,7 @@ func (pup *pipeUniqProcessor) flush() error {
 	} else {
 		for k, pHits := range m {
 			if needStop(pup.stopCh) {
-				return nil
+				return
 			}
 
 			rowFields = rowFields[:0]
@@ -379,17 +414,135 @@ func (pup *pipeUniqProcessor) flush() error {
 	}
 
 	wctx.flush()
+}
 
-	return nil
+func (pup *pipeUniqProcessor) mergeShardsParallel() ([]map[string]*uint64, error) {
+	shards := pup.shards
+	shardsLen := len(shards)
+	if shardsLen == 1 {
+		m := shards[0].getM()
+		var ms []map[string]*uint64
+		if len(m) > 0 {
+			ms = append(ms, m)
+		}
+		return ms, nil
+	}
+
+	var wg sync.WaitGroup
+	perShardMaps := make([][]map[string]*uint64, shardsLen)
+	for i := range shards {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+
+			shardMaps := make([]map[string]*uint64, shardsLen)
+			for i := range shardMaps {
+				shardMaps[i] = make(map[string]*uint64)
+			}
+
+			n := int64(0)
+			nTotal := int64(0)
+			for k, pHits := range shards[idx].getM() {
+				if needStop(pup.stopCh) {
+					return
+				}
+				h := xxhash.Sum64(bytesutil.ToUnsafeBytes(k))
+				m := shardMaps[h%uint64(len(shardMaps))]
+				n += updatePipeUniqMap(m, k, pHits)
+				if n > stateSizeBudgetChunk {
+					if nRemaining := pup.stateSizeBudget.Add(-n); nRemaining < 0 {
+						return
+					}
+					nTotal += n
+					n = 0
+				}
+			}
+			nTotal += n
+			pup.stateSizeBudget.Add(-n)
+
+			perShardMaps[idx] = shardMaps
+
+			// Clean the original map and return its state size budget back.
+			shards[idx].m = nil
+			pup.stateSizeBudget.Add(nTotal)
+		}(i)
+	}
+	wg.Wait()
+	if needStop(pup.stopCh) {
+		return nil, nil
+	}
+	if n := pup.stateSizeBudget.Load(); n < 0 {
+		return nil, fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", pup.pu.String(), pup.maxStateSize/(1<<20))
+	}
+
+	// Merge per-shard entries into perShardMaps[0]
+	for i := range perShardMaps {
+		wg.Add(1)
+		go func(idx int) {
+			defer wg.Done()
+
+			m := perShardMaps[0][idx]
+			for i := 1; i < len(perShardMaps); i++ {
+				n := int64(0)
+				nTotal := int64(0)
+				for k, psg := range perShardMaps[i][idx] {
+					if needStop(pup.stopCh) {
+						return
+					}
+					n += updatePipeUniqMap(m, k, psg)
+					if n > stateSizeBudgetChunk {
+						if nRemaining := pup.stateSizeBudget.Add(-n); nRemaining < 0 {
+							return
+						}
+						nTotal += n
+						n = 0
+					}
+				}
+				nTotal += n
+				pup.stateSizeBudget.Add(-n)
+
+				// Clean the original map and return its state size budget back.
+				perShardMaps[i][idx] = nil
+				pup.stateSizeBudget.Add(nTotal)
+			}
+		}(i)
+	}
+	wg.Wait()
+	if needStop(pup.stopCh) {
+		return nil, nil
+	}
+	if n := pup.stateSizeBudget.Load(); n < 0 {
+		return nil, fmt.Errorf("cannot calculate [%s], since it requires more than %dMB of memory", pup.pu.String(), pup.maxStateSize/(1<<20))
+	}
+
+	// Filter out maps without entries
+	ms := perShardMaps[0]
+	result := ms[:0]
+	for _, m := range ms {
+		if len(m) > 0 {
+			result = append(result, m)
+		}
+	}
+
+	return result, nil
 }
 
-type pipeUniqWriteContext struct {
-	pup *pipeUniqProcessor
-	rcs []resultColumn
-	br  blockResult
+func updatePipeUniqMap(m map[string]*uint64, k string, pHitsSrc *uint64) int64 {
+	pHitsDst := m[k]
+	if pHitsDst != nil {
+		*pHitsDst += *pHitsSrc
+		return 0
+	}
 
-	// rowsWritten is the total number of rows passed to writeRow.
-	rowsWritten uint64
+	m[k] = pHitsSrc
+	return int64(unsafe.Sizeof(k) + unsafe.Sizeof(pHitsSrc))
+}
+
+type pipeUniqWriteContext struct {
+	workerID uint
+	pup      *pipeUniqProcessor
+	rcs      []resultColumn
+	br       blockResult
 
 	// rowsCount is the number of rows in the current block
 	rowsCount int
@@ -399,11 +552,6 @@ type pipeUniqWriteContext struct {
 }
 
 func (wctx *pipeUniqWriteContext) writeRow(rowFields []Field) {
-	if limit := wctx.pup.pu.limit; limit > 0 && wctx.rowsWritten >= limit {
-		return
-	}
-	wctx.rowsWritten++
-
 	rcs := wctx.rcs
 
 	areEqualColumns := len(rcs) == len(rowFields)
@@ -447,7 +595,7 @@ func (wctx *pipeUniqWriteContext) flush() {
 	// Flush rcs to ppNext
 	br.setResultColumns(rcs, wctx.rowsCount)
 	wctx.rowsCount = 0
-	wctx.pup.ppNext.writeBlock(0, br)
+	wctx.pup.ppNext.writeBlock(wctx.workerID, br)
 	br.reset()
 	for i := range rcs {
 		rcs[i].resetValues()
diff --git a/lib/logstorage/storage.go b/lib/logstorage/storage.go
index a448bacc4..a5de2b88e 100644
--- a/lib/logstorage/storage.go
+++ b/lib/logstorage/storage.go
@@ -11,9 +11,7 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/memory"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/workingsetcache"
 )
 
 // StorageStats represents stats for the storage. It may be obtained by calling Storage.UpdateStats().
@@ -136,12 +134,12 @@ type Storage struct {
 	//
 	// It reduces the load on persistent storage during data ingestion by skipping
 	// the check whether the given stream is already registered in the persistent storage.
-	streamIDCache *workingsetcache.Cache
+	streamIDCache *cache
 
 	// filterStreamCache caches streamIDs keyed by (partition, []TenanID, StreamFilter).
 	//
 	// It reduces the load on persistent storage during querying by _stream:{...} filter.
-	filterStreamCache *workingsetcache.Cache
+	filterStreamCache *cache
 }
 
 type partitionWrapper struct {
@@ -243,11 +241,8 @@ func MustOpenStorage(path string, cfg *StorageConfig) *Storage {
 	flockF := fs.MustCreateFlockFile(path)
 
 	// Load caches
-	mem := memory.Allowed()
-	streamIDCachePath := filepath.Join(path, cacheDirname, streamIDCacheFilename)
-	streamIDCache := workingsetcache.Load(streamIDCachePath, mem/16)
-
-	filterStreamCache := workingsetcache.New(mem / 10)
+	streamIDCache := newCache()
+	filterStreamCache := newCache()
 
 	s := &Storage{
 		path:                   path,
@@ -457,15 +452,17 @@ func (s *Storage) MustClose() {
 	s.partitions = nil
 	s.ptwHot = nil
 
-	// Save caches
-	streamIDCachePath := filepath.Join(s.path, cacheDirname, streamIDCacheFilename)
-	if err := s.streamIDCache.Save(streamIDCachePath); err != nil {
-		logger.Panicf("FATAL: cannot save streamID cache to %q: %s", streamIDCachePath, err)
-	}
-	s.streamIDCache.Stop()
+	// Stop caches
+
+	// Do not persist caches, since they may become out of sync with partitions
+	// if partitions are deleted, restored from backups or copied from other sources
+	// between VictoriaLogs restarts. This may result in various issues
+	// during data ingestion and querying.
+
+	s.streamIDCache.MustStop()
 	s.streamIDCache = nil
 
-	s.filterStreamCache.Stop()
+	s.filterStreamCache.MustStop()
 	s.filterStreamCache = nil
 
 	// release lock file
diff --git a/lib/logstorage/storage_search.go b/lib/logstorage/storage_search.go
index 2e269f85e..4144475ee 100644
--- a/lib/logstorage/storage_search.go
+++ b/lib/logstorage/storage_search.go
@@ -337,8 +337,8 @@ func (s *Storage) GetStreamFieldNames(ctx context.Context, tenantIDs []TenantID,
 
 	m := make(map[string]*uint64)
 	forEachStreamField(streams, func(f Field, hits uint64) {
-		pHits, ok := m[f.Name]
-		if !ok {
+		pHits := m[f.Name]
+		if pHits == nil {
 			nameCopy := strings.Clone(f.Name)
 			hitsLocal := uint64(0)
 			pHits = &hitsLocal
@@ -364,8 +364,8 @@ func (s *Storage) GetStreamFieldValues(ctx context.Context, tenantIDs []TenantID
 		if f.Name != fieldName {
 			return
 		}
-		pHits, ok := m[f.Value]
-		if !ok {
+		pHits := m[f.Value]
+		if pHits == nil {
 			valueCopy := strings.Clone(f.Value)
 			hitsLocal := uint64(0)
 			pHits = &hitsLocal


Test patch:
diff --git a/lib/logstorage/cache_test.go b/lib/logstorage/cache_test.go
new file mode 100644
index 000000000..dd34d6e39
--- /dev/null
+++ b/lib/logstorage/cache_test.go
@@ -0,0 +1,59 @@
+package logstorage
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestCache(t *testing.T) {
+	m := make(map[string]int)
+	for i := 0; i < 10; i++ {
+		k := fmt.Sprintf("key_%d", i)
+		m[k] = i
+	}
+
+	c := newCache()
+	defer c.MustStop()
+
+	for kStr := range m {
+		k := []byte(kStr)
+
+		if v, ok := c.Get(k); ok {
+			t.Fatalf("unexpected value obtained from the cache for key %q: %v", k, v)
+		}
+		c.Set(k, m[kStr])
+		v, ok := c.Get(k)
+		if !ok {
+			t.Fatalf("cannot obtain value for key %q", k)
+		}
+		if n := v.(int); n != m[kStr] {
+			t.Fatalf("unexpected value obtained for key %q; got %d; want %d", k, n, m[kStr])
+		}
+	}
+
+	// The cached entries should be still visible after a single clean() call.
+	c.clean()
+	for kStr := range m {
+		k := []byte(kStr)
+
+		v, ok := c.Get(k)
+		if !ok {
+			t.Fatalf("cannot obtain value for key %q", k)
+		}
+		if n := v.(int); n != m[kStr] {
+			t.Fatalf("unexpected value obtained for key %q; got %d; want %d", k, n, m[kStr])
+		}
+	}
+
+	// The cached entries must be dropped after two clean() calls.
+	c.clean()
+	c.clean()
+
+	for kStr := range m {
+		k := []byte(kStr)
+
+		if v, ok := c.Get(k); ok {
+			t.Fatalf("unexpected value obtained from the cache for key %q: %v", k, v)
+		}
+	}
+}
diff --git a/lib/logstorage/partition_test.go b/lib/logstorage/partition_test.go
index 7f67ffb45..de297a85c 100644
--- a/lib/logstorage/partition_test.go
+++ b/lib/logstorage/partition_test.go
@@ -6,7 +6,6 @@ import (
 	"time"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/workingsetcache"
 )
 
 func TestPartitionLifecycle(t *testing.T) {
@@ -183,8 +182,8 @@ func TestPartitionMustAddRowsConcurrent(t *testing.T) {
 //
 // When the storage is no longer needed, closeTestStorage() must be called.
 func newTestStorage() *Storage {
-	streamIDCache := workingsetcache.New(1024 * 1024)
-	filterStreamCache := workingsetcache.New(1024 * 1024)
+	streamIDCache := newCache()
+	filterStreamCache := newCache()
 	return &Storage{
 		flushInterval:     time.Second,
 		streamIDCache:     streamIDCache,
@@ -194,6 +193,6 @@ func newTestStorage() *Storage {
 
 // closeTestStorage closes storage created via newTestStorage().
 func closeTestStorage(s *Storage) {
-	s.streamIDCache.Stop()
-	s.filterStreamCache.Stop()
+	s.streamIDCache.MustStop()
+	s.filterStreamCache.MustStop()
 }