Repository: VictoriaMetrics/VictoriaMetrics
Pull request #7242 (2024-10-11 13:20:48)

docs: make all statement of active time series consistent
### Describe Your Changes

https://github.com/VictoriaMetrics/VictoriaMetrics/issues/7212

### Checklist

The following checks are **mandatory**:

- [x] My change adheres [VictoriaMetrics contributing guidelines](https://docs.victoriametrics.com/contributing/).



Patch:
diff --git a/app/victoria-logs/main.go b/app/victoria-logs/main.go
index 32834904a..19216c428 100644
--- a/app/victoria-logs/main.go
+++ b/app/victoria-logs/main.go
@@ -92,6 +92,9 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 	if vlselect.RequestHandler(w, r) {
 		return true
 	}
+	if vlstorage.RequestHandler(w, r) {
+		return true
+	}
 	return false
 }
 
diff --git a/app/vlstorage/main.go b/app/vlstorage/main.go
index 628919ce5..d4ac5a171 100644
--- a/app/vlstorage/main.go
+++ b/app/vlstorage/main.go
@@ -37,6 +37,8 @@ var (
 		"see https://docs.victoriametrics.com/victorialogs/data-ingestion/ ; see also -logNewStreams")
 	minFreeDiskSpaceBytes = flagutil.NewBytes("storage.minFreeDiskSpaceBytes", 10e6, "The minimum free disk space at -storageDataPath after which "+
 		"the storage stops accepting new data")
+
+	forceMergeAuthKey = flagutil.NewPassword("forceMergeAuthKey", "authKey, which must be passed in query string to /internal/force_merge pages. It overrides -httpAuth.*")
 )
 
 // Init initializes vlstorage.
@@ -87,6 +89,28 @@ func Stop() {
 	strg = nil
 }
 
+// RequestHandler is a storage request handler.
+func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
+	path := r.URL.Path
+	if path == "/internal/force_merge" {
+		if !httpserver.CheckAuthFlag(w, r, forceMergeAuthKey) {
+			return true
+		}
+		// Run force merge in background
+		partitionNamePrefix := r.FormValue("partition_prefix")
+		go func() {
+			activeForceMerges.Inc()
+			defer activeForceMerges.Dec()
+			logger.Infof("forced merge for partition_prefix=%q has been started", partitionNamePrefix)
+			startTime := time.Now()
+			strg.MustForceMerge(partitionNamePrefix)
+			logger.Infof("forced merge for partition_prefix=%q has been successfully finished in %.3f seconds", partitionNamePrefix, time.Since(startTime).Seconds())
+		}()
+		return true
+	}
+	return false
+}
+
 var strg *logstorage.Storage
 var storageMetrics *metrics.Set
 
@@ -205,3 +229,5 @@ func writeStorageMetrics(w io.Writer, strg *logstorage.Storage) {
 	metrics.WriteCounterUint64(w, `vl_rows_dropped_total{reason="too_big_timestamp"}`, ss.RowsDroppedTooBigTimestamp)
 	metrics.WriteCounterUint64(w, `vl_rows_dropped_total{reason="too_small_timestamp"}`, ss.RowsDroppedTooSmallTimestamp)
 }
+
+var activeForceMerges = metrics.NewCounter("vl_active_force_merges")
diff --git a/docs/Cluster-VictoriaMetrics.md b/docs/Cluster-VictoriaMetrics.md
index 9adcaeea9..edfe8f843 100644
--- a/docs/Cluster-VictoriaMetrics.md
+++ b/docs/Cluster-VictoriaMetrics.md
@@ -77,7 +77,7 @@ or [vmgateway](https://docs.victoriametrics.com/vmgateway/). [Contact us](mailto
 - Data for all the tenants is evenly spread among available `vmstorage` nodes. This guarantees even load among `vmstorage` nodes
 when different tenants have different amounts of data and different query load.
 
-- The database performance and resource usage doesn't depend on the number of tenants. It depends mostly on the total number of active time series in all the tenants. A time series is considered active if it received at least a single sample during the last hour or it has been touched by queries during the last hour.
+- The database performance and resource usage doesn't depend on the number of tenants. It depends mostly on the total number of [active time series](https://docs.victoriametrics.com/faq/#what-is-an-active-time-series) in all the tenants. A time series is considered active if it received at least a single sample during the last hour or it has been touched by queries during the last hour.
 
 - The list of registered tenants can be obtained via `http://<vmselect>:8481/admin/tenants` url. See [these docs](#url-format).
 
diff --git a/docs/FAQ.md b/docs/FAQ.md
index 33066a1ea..a9a12e5b5 100644
--- a/docs/FAQ.md
+++ b/docs/FAQ.md
@@ -315,7 +315,7 @@ Yes. See [these docs](https://docs.victoriametrics.com/#graphite-api-usage).
 ## What is an active time series?
 
 A time series is uniquely identified by its name plus a set of its labels. For example, `temperature{city="NY",country="US"}` and `temperature{city="SF",country="US"}`
-are two distinct series, since they differ by the `city` label. A time series is considered active if it receives at least a single new sample during the last hour.
+are two distinct series, since they differ by the `city` label. A time series is considered active if it received at least a single sample during the last hour or it has been touched by queries during the last hour.
 The number of active time series is displayed on the official Grafana dashboard for VictoriaMetrics - see [these docs](https://docs.victoriametrics.com/#monitoring) for details.
 
 ## What is high churn rate?
diff --git a/docs/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md
index 533f5488e..cfedbe96f 100644
--- a/docs/VictoriaLogs/CHANGELOG.md
+++ b/docs/VictoriaLogs/CHANGELOG.md
@@ -15,6 +15,12 @@ according to [these docs](https://docs.victoriametrics.com/victorialogs/quicksta
 
 ## tip
 
+* FEATURE: add support for forced merge. See [these docs](https://docs.victoriametrics.com/victorialogs/#forced-merge).
+
+* BUGFIX: avoid possible panic when logs for a new day are ingested during execution of concurrent queries.
+* BUGFIX: avoid panic at `lib/logstorage.(*blockResultColumn).forEachDictValue()` when [stats with additional filters](https://docs.victoriametrics.com/victorialogs/logsql/#stats-with-additional-filters). The panic has been introduced in [v0.33.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.33.0-victorialogs) in [this commit](https://github.com/VictoriaMetrics/VictoriaMetrics/commit/a350be48b68330ee1a487e1fb09b002d3be45163).
+
+
 ## [v0.35.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v0.35.0-victorialogs)
 
 * FEATURE: [vlogscli](https://docs.victoriametrics.com/victorialogs/querying/vlogscli/): add ability to live tail query results - see [these docs](https://docs.victoriametrics.com/victorialogs/querying/vlogscli/#live-tailing).
diff --git a/docs/VictoriaLogs/LogsQL.md b/docs/VictoriaLogs/LogsQL.md
index 332b009cf..be981f085 100644
--- a/docs/VictoriaLogs/LogsQL.md
+++ b/docs/VictoriaLogs/LogsQL.md
@@ -468,7 +468,7 @@ See also:
 
 ### _stream_id filter
 
-Every [log stream](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) in VictoriaMetrics is uniquely identified by `_stream_id` field.
+Every [log stream](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields) in VictoriaLogs is uniquely identified by `_stream_id` field.
 The `_stream_id:...` filter allows quickly selecting all the logs belonging to the particular stream.
 
 For example, the following query selects all the logs, which belong to the [log stream](https://docs.victoriametrics.com/victorialogs/keyconcepts/#stream-fields)
diff --git a/docs/VictoriaLogs/README.md b/docs/VictoriaLogs/README.md
index 16005acb3..cd8d53b72 100644
--- a/docs/VictoriaLogs/README.md
+++ b/docs/VictoriaLogs/README.md
@@ -117,7 +117,7 @@ Set the `-retentionPeriod` to some big value (e.g. `100y` - 100 years) if logs s
 For example:
 
 ```sh
-/path/to/victoria-logs -retention.maxDiskSpaceUsageBytes=10TiB -retention=100y
+/path/to/victoria-logs -retention.maxDiskSpaceUsageBytes=10TiB -retentionPeriod=100y
 ```
 
 ## Storage
@@ -131,6 +131,19 @@ For example, the following command starts VictoriaLogs, which stores the data at
 
 VictoriaLogs automatically creates the `-storageDataPath` directory on the first run if it is missing.
 
+## Forced merge
+
+VictoriaLogs performs data compactions in background in order to keep good performance characteristics when accepting new data.
+These compactions (merges) are performed independently on per-day partitions.
+This means that compactions are stopped for per-day partitions if no new data is ingested into these partitions.
+Sometimes it is necessary to trigger compactions for old partitions. In this case forced compaction may be initiated on the specified per-month partition
+by sending request to `/internal/force_merge?partition_prefix=YYYYMMDD`,
+where `YYYYMMDD` is per-day partition name. For example, `http://victoria-logs:9428/internal/force_merge?partition_prefix=20240921` would initiate forced
+merge for September 21, 2024 partition. The call to `/internal/force_merge` returns immediately, while the corresponding forced merge continues running in background.
+
+Forced merges may require additional CPU, disk IO and storage space resources. It is unnecessary to run forced merge under normal conditions,
+since VictoriaLogs automatically performs optimal merges in background when new data is ingested into it.
+
 ## High Availability
 
 ### High Availability (HA) Setup with VictoriaLogs Single-Node Instances
@@ -147,9 +160,9 @@ This schema outlines how to configure a High Availability (HA) setup using Victo
 
 Here are the working example of HA configuration for VictoriaLogs using Docker Compose:
 
-- [Fluent Bit + VictoriaLogs Single-Node + vmauth](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker/victorialogs/fluentbit-ha-single-node)
-- [Logstash + VictoriaLogs Single-Node + vmauth](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker/victorialogs/logstash-ha-single-node)
-- [Vector + VictoriaLogs Single-Node + vmauth](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker/victorialogs/vector-ha-single-node)
+- [Fluent Bit + VictoriaLogs Single-Node + vmauth](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker/victorialogs/fluentbit/jsonline-ha)
+- [Logstash + VictoriaLogs Single-Node + vmauth](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker/victorialogs/logstash/jsonline-ha)
+- [Vector + VictoriaLogs Single-Node + vmauth](https://github.com/VictoriaMetrics/VictoriaMetrics/tree/master/deployment/docker/victorialogs/vector/jsonline-ha)
 
 ## Backup and restore
 
diff --git a/docs/VictoriaLogs/Roadmap.md b/docs/VictoriaLogs/Roadmap.md
index 3dbabcb95..5d2f28f37 100644
--- a/docs/VictoriaLogs/Roadmap.md
+++ b/docs/VictoriaLogs/Roadmap.md
@@ -21,11 +21,8 @@ See [these docs](https://docs.victoriametrics.com/victorialogs/) for details.
 The following functionality is planned in the future versions of VictoriaLogs:
 
 - Support for [data ingestion](https://docs.victoriametrics.com/victorialogs/data-ingestion/) from popular log collectors and formats:
-  - [x] [OpenTelemetry for logs](https://docs.victoriametrics.com/victorialogs/data-ingestion/opentelemetry/)
-  - [x] [Fluentd](https://docs.victoriametrics.com/victorialogs/data-ingestion/fluentd/)
   - [ ] [Journald](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/4618) (systemd)
   - [ ] [Datadog protocol for logs](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6632)
-  - [x] [Telegraf](https://docs.victoriametrics.com/victorialogs/data-ingestion/telegraf/)
 - [ ] Integration with Grafana. Partially done, check the [documentation](https://docs.victoriametrics.com/victorialogs/victorialogs-datasource/) and [datasource repository](https://github.com/VictoriaMetrics/victorialogs-datasource).
 - [ ] Ability to make instant snapshots and backups in the way [similar to VictoriaMetrics](https://docs.victoriametrics.com/#how-to-work-with-snapshots).
 - [ ] Cluster version of VictoriaLogs.
diff --git a/docs/VictoriaLogs/querying/vlogscli.md b/docs/VictoriaLogs/querying/vlogscli.md
index 8f29f6857..5bfe8321c 100644
--- a/docs/VictoriaLogs/querying/vlogscli.md
+++ b/docs/VictoriaLogs/querying/vlogscli.md
@@ -100,7 +100,7 @@ See also:
 
 If the query response exceeds vertical screen space, `vlogsql` pipes query response to `less` utility,
 so you can scroll the response as needed. This allows executing queries, which potentially
-may return billions of rows, without any problems at both VictoriaMetrics and `vlogsql` sides,
+may return billions of rows, without any problems at both VictoriaLogs and `vlogsql` sides,
 thanks to the way how `less` interacts with [`/select/logsql/query`](https://docs.victoriametrics.com/victorialogs/querying/#querying-logs):
 
 - `less` reads the response when needed, e.g. when you scroll it down.
@@ -112,7 +112,7 @@ thanks to the way how `less` interacts with [`/select/logsql/query`](https://doc
   after the response stream is closed.
 
 See also [`less` docs](https://man7.org/linux/man-pages/man1/less.1.html) and
-[command-line integration docs for VictoriaMetrics](https://docs.victoriametrics.com/victorialogs/querying/#command-line).
+[command-line integration docs for VictoriaLogs](https://docs.victoriametrics.com/victorialogs/querying/#command-line).
 
 
 ## Live tailing
diff --git a/docs/changelog/CHANGELOG.md b/docs/changelog/CHANGELOG.md
index ee34c8c28..f65d4e085 100644
--- a/docs/changelog/CHANGELOG.md
+++ b/docs/changelog/CHANGELOG.md
@@ -20,7 +20,7 @@ See also [LTS releases](https://docs.victoriametrics.com/lts-releases/).
 
 * FEATURE: add Darwin binaries for [VictoriaMetrics cluster](https://docs.victoriametrics.com/cluster-victoriametrics/) to the release flow. The binaries will be available in the new release.
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/): allow using HTTP/2 client for Kubernetes service discovery if `-promscrape.kubernetes.useHTTP2Client` cmd-line flag is set. This could help to reduce the amount of opened connections to the Kubernetes API server. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/5971) for the details.
-* FEATURE: [vmalert](https://docs.victoriametrics.com/vmalert/): `-rule` cmd-line flag now supports multi-document YAML files. This could be usefule when rules are retrieved from via HTTP where multiple rule files were merged together in one response. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6753). Thanks to @Irene-123 for [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6995).
+* FEATURE: [vmalert](https://docs.victoriametrics.com/vmalert/): `-rule` cmd-line flag now supports multi-document YAML files. This could be useful when rules are retrieved via HTTP URL where multiple rule files were merged together in one response. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/6753). Thanks to @Irene-123 for [the pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6995).
 * FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent/) and [Single-node VictoriaMetrics](https://docs.victoriametrics.com/): add support of [exponential histograms](https://opentelemetry.io/docs/specs/otel/metrics/data-model/#exponentialhistogram) ingested via [OpenTelemetry protocol for metrics](https://docs.victoriametrics.com/#sending-data-via-opentelemetry). Such histograms will be automatically converted to [VictoriaMetrics histogram format](https://valyala.medium.com/improving-histogram-usability-for-prometheus-and-grafana-bc7e5df0e350). See [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/6354).
 
 ## [v1.104.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.104.0)
diff --git a/lib/logstorage/block_data.go b/lib/logstorage/block_data.go
index 9f009c098..0ad517c02 100644
--- a/lib/logstorage/block_data.go
+++ b/lib/logstorage/block_data.go
@@ -158,12 +158,10 @@ func (bd *blockData) mustReadFrom(a *arena, bh *blockHeader, sr *streamReaders)
 	bb.B = bytesutil.ResizeNoCopyMayOverallocate(bb.B, int(columnsHeaderSize))
 	sr.columnsHeaderReader.MustReadFull(bb.B)
 
-	cshA := getArena()
 	csh := getColumnsHeader()
-	if err := csh.unmarshal(cshA, bb.B); err != nil {
+	if err := csh.unmarshalNoArena(bb.B); err != nil {
 		logger.Panicf("FATAL: %s: cannot unmarshal columnsHeader: %s", sr.columnsHeaderReader.Path(), err)
 	}
-	longTermBufPool.Put(bb)
 	chs := csh.columnHeaders
 	cds := bd.resizeColumnsData(len(chs))
 	for i := range chs {
@@ -171,7 +169,7 @@ func (bd *blockData) mustReadFrom(a *arena, bh *blockHeader, sr *streamReaders)
 	}
 	bd.constColumns = appendFields(a, bd.constColumns[:0], csh.constColumns)
 	putColumnsHeader(csh)
-	putArena(cshA)
+	longTermBufPool.Put(bb)
 }
 
 // timestampsData contains the encoded timestamps data.
diff --git a/lib/logstorage/block_header.go b/lib/logstorage/block_header.go
index 578cae535..96ae05282 100644
--- a/lib/logstorage/block_header.go
+++ b/lib/logstorage/block_header.go
@@ -235,34 +235,6 @@ func (csh *columnsHeader) reset() {
 	csh.constColumns = ccs[:0]
 }
 
-func (csh *columnsHeader) getConstColumnValue(name string) string {
-	if name == "_msg" {
-		name = ""
-	}
-	ccs := csh.constColumns
-	for i := range ccs {
-		cc := &ccs[i]
-		if cc.Name == name {
-			return cc.Value
-		}
-	}
-	return ""
-}
-
-func (csh *columnsHeader) getColumnHeader(name string) *columnHeader {
-	if name == "_msg" {
-		name = ""
-	}
-	chs := csh.columnHeaders
-	for i := range chs {
-		ch := &chs[i]
-		if ch.name == name {
-			return ch
-		}
-	}
-	return nil
-}
-
 func (csh *columnsHeader) resizeConstColumns(columnsLen int) []Field {
 	csh.constColumns = slicesutil.SetLength(csh.constColumns, columnsLen)
 	return csh.constColumns
@@ -289,10 +261,10 @@ func (csh *columnsHeader) marshal(dst []byte) []byte {
 	return dst
 }
 
-// unmarshal unmarshals csh from src.
+// unmarshalNoArena unmarshals csh from src.
 //
-// csh is valid until a.reset() is called.
-func (csh *columnsHeader) unmarshal(a *arena, src []byte) error {
+// csh is valid until src is changed.
+func (csh *columnsHeader) unmarshalNoArena(src []byte) error {
 	csh.reset()
 
 	// unmarshal columnHeaders
@@ -307,7 +279,7 @@ func (csh *columnsHeader) unmarshal(a *arena, src []byte) error {
 
 	chs := csh.resizeColumnHeaders(int(n))
 	for i := range chs {
-		tail, err := chs[i].unmarshal(a, src)
+		tail, err := chs[i].unmarshalNoArena(src)
 		if err != nil {
 			return fmt.Errorf("cannot unmarshal columnHeader %d out of %d columnHeaders: %w", i, len(chs), err)
 		}
@@ -327,7 +299,7 @@ func (csh *columnsHeader) unmarshal(a *arena, src []byte) error {
 
 	ccs := csh.resizeConstColumns(int(n))
 	for i := range ccs {
-		tail, err := ccs[i].unmarshal(a, src)
+		tail, err := ccs[i].unmarshalNoArena(src)
 		if err != nil {
 			return fmt.Errorf("cannot unmarshal constColumn %d out of %d columns: %w", i, len(ccs), err)
 		}
@@ -497,10 +469,10 @@ func (ch *columnHeader) marshalBloomFilters(dst []byte) []byte {
 	return dst
 }
 
-// unmarshal unmarshals ch from src and returns the tail left after unmarshaling.
+// unmarshalNoArena unmarshals ch from src and returns the tail left after unmarshaling.
 //
-// ch is valid until a.reset() is called.
-func (ch *columnHeader) unmarshal(a *arena, src []byte) ([]byte, error) {
+// ch is valid until src is changed.
+func (ch *columnHeader) unmarshalNoArena(src []byte) ([]byte, error) {
 	ch.reset()
 
 	srcOrig := src
@@ -511,7 +483,7 @@ func (ch *columnHeader) unmarshal(a *arena, src []byte) ([]byte, error) {
 		return srcOrig, fmt.Errorf("cannot unmarshal column name")
 	}
 	src = src[nSize:]
-	ch.name = a.copyBytesToString(data)
+	ch.name = bytesutil.ToUnsafeString(data)
 
 	// Unmarshal value type
 	if len(src) < 1 {
@@ -529,7 +501,7 @@ func (ch *columnHeader) unmarshal(a *arena, src []byte) ([]byte, error) {
 		}
 		src = tail
 	case valueTypeDict:
-		tail, err := ch.valuesDict.unmarshal(a, src)
+		tail, err := ch.valuesDict.unmarshalNoArena(src)
 		if err != nil {
 			return srcOrig, fmt.Errorf("cannot unmarshal dict at valueTypeDict for column %q: %w", ch.name, err)
 		}
diff --git a/lib/logstorage/block_result.go b/lib/logstorage/block_result.go
index 319df03fb..8eb2b1c72 100644
--- a/lib/logstorage/block_result.go
+++ b/lib/logstorage/block_result.go
@@ -305,11 +305,10 @@ func (br *blockResult) initAllColumns() {
 
 	if !slices.Contains(unneededColumnNames, "_msg") {
 		// Add _msg column
-		csh := br.bs.getColumnsHeader()
-		v := csh.getConstColumnValue("_msg")
+		v := br.bs.getConstColumnValue("_msg")
 		if v != "" {
 			br.addConstColumn("_msg", v)
-		} else if ch := csh.getColumnHeader("_msg"); ch != nil {
+		} else if ch := br.bs.getColumnHeader("_msg"); ch != nil {
 			br.addColumn(ch)
 		} else {
 			br.addConstColumn("_msg", "")
@@ -317,9 +316,9 @@ func (br *blockResult) initAllColumns() {
 	}
 
 	// Add other const columns
-	csh := br.bs.getColumnsHeader()
-	for _, cc := range csh.constColumns {
-		if isMsgFieldName(cc.Name) {
+	ccs := br.bs.getConstColumns()
+	for _, cc := range ccs {
+		if cc.Name == "" {
 			continue
 		}
 		if !slices.Contains(unneededColumnNames, cc.Name) {
@@ -328,10 +327,10 @@ func (br *blockResult) initAllColumns() {
 	}
 
 	// Add other non-const columns
-	chs := csh.columnHeaders
+	chs := br.bs.getColumnHeaders()
 	for i := range chs {
 		ch := &chs[i]
-		if isMsgFieldName(ch.name) {
+		if ch.name == "" {
 			continue
 		}
 		if !slices.Contains(unneededColumnNames, ch.name) {
@@ -357,11 +356,10 @@ func (br *blockResult) initRequestedColumns() {
 		case "_time":
 			br.addTimeColumn()
 		default:
-			csh := br.bs.getColumnsHeader()
-			v := csh.getConstColumnValue(columnName)
+			v := br.bs.getConstColumnValue(columnName)
 			if v != "" {
 				br.addConstColumn(columnName, v)
-			} else if ch := csh.getColumnHeader(columnName); ch != nil {
+			} else if ch := br.bs.getColumnHeader(columnName); ch != nil {
 				br.addColumn(ch)
 			} else {
 				br.addConstColumn(columnName, "")
@@ -1733,7 +1731,7 @@ func (c *blockResultColumn) forEachDictValue(br *blockResult, f func(v string))
 	if c.valueType != valueTypeDict {
 		logger.Panicf("BUG: unexpected column valueType=%d; want %d", c.valueType, valueTypeDict)
 	}
-	if uint64(br.rowsLen) == br.bs.bsw.bh.rowsCount {
+	if br.bs != nil && uint64(br.rowsLen) == br.bs.bsw.bh.rowsCount {
 		// Fast path - there is no need in reading encoded values
 		for _, v := range c.dictValues {
 			f(v)
diff --git a/lib/logstorage/block_search.go b/lib/logstorage/block_search.go
index 80b708a4c..d954bf0fb 100644
--- a/lib/logstorage/block_search.go
+++ b/lib/logstorage/block_search.go
@@ -113,16 +113,15 @@ type blockSearch struct {
 	// sbu is used for unmarshaling local columns
 	sbu stringsBlockUnmarshaler
 
-	// cshCached is the columnsHeader associated with the given block
+	// cshBlockCache holds columnsHeader data for the given block.
 	//
 	// it is initialized lazily by calling getColumnsHeader().
-	cshCached columnsHeader
+	cshBlockCache []byte
 
-	// cshInitialized is set to true if cshCached is initialized.
-	cshInitialized bool
-
-	// a is used for storing unmarshaled data in cshCached
-	a arena
+	// cshCache is the columnsHeader associated with the given block
+	//
+	// it is initialized lazily by calling getColumnsHeader().
+	cshCache *columnsHeader
 
 	// seenStreams contains seen streamIDs for the recent searches.
 	// It is used for speeding up fetching _stream column.
@@ -152,10 +151,12 @@ func (bs *blockSearch) reset() {
 
 	bs.sbu.reset()
 
-	bs.cshCached.reset()
-	bs.cshInitialized = false
+	bs.cshBlockCache = bs.cshBlockCache[:0]
 
-	bs.a.reset()
+	if bs.cshCache != nil {
+		putColumnsHeader(bs.cshCache)
+		bs.cshCache = nil
+	}
 
 	// Do not reset seenStreams, since its' lifetime is managed by blockResult.addStreamColumn() code.
 }
@@ -189,26 +190,67 @@ func (bs *blockSearch) search(bsw *blockSearchWork, bm *bitmap) {
 	}
 }
 
+func (bs *blockSearch) getConstColumnValue(name string) string {
+	if name == "_msg" {
+		name = ""
+	}
+
+	csh := bs.getColumnsHeader()
+	for _, cc := range csh.constColumns {
+		if cc.Name == name {
+			return cc.Value
+		}
+	}
+	return ""
+}
+
+func (bs *blockSearch) getColumnHeader(name string) *columnHeader {
+	if name == "_msg" {
+		name = ""
+	}
+
+	csh := bs.getColumnsHeader()
+	chs := csh.columnHeaders
+	for i := range chs {
+		ch := &chs[i]
+		if ch.name == name {
+			return ch
+		}
+	}
+	return nil
+}
+
+func (bs *blockSearch) getConstColumns() []Field {
+	csh := bs.getColumnsHeader()
+	return csh.constColumns
+}
+
+func (bs *blockSearch) getColumnHeaders() []columnHeader {
+	csh := bs.getColumnsHeader()
+	return csh.columnHeaders
+}
+
 func (bs *blockSearch) getColumnsHeader() *columnsHeader {
-	if !bs.cshInitialized {
-		bs.cshCached.initFromBlockHeader(&bs.a, bs.bsw.p, &bs.bsw.bh)
+	if bs.cshCache == nil {
+		bs.cshBlockCache = readColumnsHeaderBlock(bs.cshBlockCache[:0], bs.bsw.p, &bs.bsw.bh)
+
+		bs.cshCache = getColumnsHeader()
+		if err := bs.cshCache.unmarshalNoArena(bs.cshBlockCache); err != nil {
+			logger.Panicf("FATAL: %s: cannot unmarshal columns header: %s", bs.bsw.p.path, err)
+		}
 	}
-	return &bs.cshCached
+	return bs.cshCache
 }
 
-func (csh *columnsHeader) initFromBlockHeader(a *arena, p *part, bh *blockHeader) {
-	bb := longTermBufPool.Get()
+func readColumnsHeaderBlock(dst []byte, p *part, bh *blockHeader) []byte {
 	columnsHeaderSize := bh.columnsHeaderSize
 	if columnsHeaderSize > maxColumnsHeaderSize {
 		logger.Panicf("FATAL: %s: columns header size cannot exceed %d bytes; got %d bytes", p.path, maxColumnsHeaderSize, columnsHeaderSize)
 	}
-	bb.B = bytesutil.ResizeNoCopyMayOverallocate(bb.B, int(columnsHeaderSize))
-	p.columnsHeaderFile.MustReadAt(bb.B, int64(bh.columnsHeaderOffset))
-
-	if err := csh.unmarshal(a, bb.B); err != nil {
-		logger.Panicf("FATAL: %s: cannot unmarshal columns header: %s", p.path, err)
-	}
-	longTermBufPool.Put(bb)
+	dstLen := len(dst)
+	dst = bytesutil.ResizeNoCopyMayOverallocate(dst, int(columnsHeaderSize)+dstLen)
+	p.columnsHeaderFile.MustReadAt(dst[dstLen:], int64(bh.columnsHeaderOffset))
+	return dst
 }
 
 // getBloomFilterForColumn returns bloom filter for the given ch.
diff --git a/lib/logstorage/datadb.go b/lib/logstorage/datadb.go
index 5d559bf43..99a1d6f7a 100644
--- a/lib/logstorage/datadb.go
+++ b/lib/logstorage/datadb.go
@@ -459,6 +459,7 @@ func assertIsInMerge(pws []*partWrapper) {
 // mustMergeParts merges pws to a single resulting part.
 //
 // if isFinal is set, then the resulting part is guaranteed to be saved to disk.
+// if isFinal is set, then the merge process cannot be interrupted.
 // The pws may remain unmerged after returning from the function if there is no enough disk space.
 //
 // All the parts inside pws must have isInMerge field set to true.
@@ -1226,3 +1227,48 @@ func getBlocksCount(pws []*partWrapper) uint64 {
 	}
 	return n
 }
+
+func (ddb *datadb) mustForceMergeAllParts() {
+	// Flush inmemory parts to files before forced merge
+	ddb.mustFlushInmemoryPartsToFiles(true)
+
+	var pws []*partWrapper
+
+	// Collect all the file parts for forced merge
+	ddb.partsLock.Lock()
+	pws = appendAllPartsForMergeLocked(pws, ddb.smallParts)
+	pws = appendAllPartsForMergeLocked(pws, ddb.bigParts)
+	ddb.partsLock.Unlock()
+
+	// If len(pws) == 1, then the merge must run anyway.
+	// This allows applying the configured retention, removing the deleted data, etc.
+
+	// Merge pws optimally
+	wg := getWaitGroup()
+	for len(pws) > 0 {
+		pwsToMerge, pwsRemaining := getPartsForOptimalMerge(pws)
+		wg.Add(1)
+		bigPartsConcurrencyCh <- struct{}{}
+		go func(pwsChunk []*partWrapper) {
+			defer func() {
+				<-bigPartsConcurrencyCh
+				wg.Done()
+			}()
+
+			ddb.mustMergeParts(pwsChunk, false)
+		}(pwsToMerge)
+		pws = pwsRemaining
+	}
+	wg.Wait()
+	putWaitGroup(wg)
+}
+
+func appendAllPartsForMergeLocked(dst, src []*partWrapper) []*partWrapper {
+	for _, pw := range src {
+		if !pw.isInMerge {
+			pw.isInMerge = true
+			dst = append(dst, pw)
+		}
+	}
+	return dst
+}
diff --git a/lib/logstorage/filter_and.go b/lib/logstorage/filter_and.go
index 3d0ac2091..624cb9c59 100644
--- a/lib/logstorage/filter_and.go
+++ b/lib/logstorage/filter_and.go
@@ -81,8 +81,7 @@ func (fa *filterAnd) matchBloomFilters(bs *blockSearch) bool {
 		fieldName := ft.field
 		tokens := ft.tokens
 
-		csh := bs.getColumnsHeader()
-		v := csh.getConstColumnValue(fieldName)
+		v := bs.getConstColumnValue(fieldName)
 		if v != "" {
 			if matchStringByAllTokens(v, tokens) {
 				continue
@@ -90,7 +89,7 @@ func (fa *filterAnd) matchBloomFilters(bs *blockSearch) bool {
 			return false
 		}
 
-		ch := csh.getColumnHeader(fieldName)
+		ch := bs.getColumnHeader(fieldName)
 		if ch == nil {
 			return false
 		}
diff --git a/lib/logstorage/filter_any_case_phrase.go b/lib/logstorage/filter_any_case_phrase.go
index 63f2ee338..d1d54feae 100644
--- a/lib/logstorage/filter_any_case_phrase.go
+++ b/lib/logstorage/filter_any_case_phrase.go
@@ -86,8 +86,7 @@ func (fp *filterAnyCasePhrase) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	phraseLowercase := fp.getPhraseLowercase()
 
 	// Verify whether fp matches const column
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchAnyCasePhrase(v, phraseLowercase) {
 			bm.resetBits()
@@ -96,7 +95,7 @@ func (fp *filterAnyCasePhrase) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether fp matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		// It matches anything only for empty phrase.
diff --git a/lib/logstorage/filter_any_case_prefix.go b/lib/logstorage/filter_any_case_prefix.go
index 1e569612d..c471b6f3f 100644
--- a/lib/logstorage/filter_any_case_prefix.go
+++ b/lib/logstorage/filter_any_case_prefix.go
@@ -90,8 +90,7 @@ func (fp *filterAnyCasePrefix) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	prefixLowercase := fp.getPrefixLowercase()
 
 	// Verify whether fp matches const column
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchAnyCasePrefix(v, prefixLowercase) {
 			bm.resetBits()
@@ -100,7 +99,7 @@ func (fp *filterAnyCasePrefix) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether fp matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		bm.resetBits()
diff --git a/lib/logstorage/filter_exact.go b/lib/logstorage/filter_exact.go
index 3d7359047..d68f96b62 100644
--- a/lib/logstorage/filter_exact.go
+++ b/lib/logstorage/filter_exact.go
@@ -174,8 +174,7 @@ func (fe *filterExact) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	fieldName := fe.fieldName
 	value := fe.value
 
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if value != v {
 			bm.resetBits()
@@ -184,7 +183,7 @@ func (fe *filterExact) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether filter matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		// It matches anything only for empty value.
diff --git a/lib/logstorage/filter_exact_prefix.go b/lib/logstorage/filter_exact_prefix.go
index bf894166f..efc75d007 100644
--- a/lib/logstorage/filter_exact_prefix.go
+++ b/lib/logstorage/filter_exact_prefix.go
@@ -51,8 +51,7 @@ func (fep *filterExactPrefix) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	fieldName := fep.fieldName
 	prefix := fep.prefix
 
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchExactPrefix(v, prefix) {
 			bm.resetBits()
@@ -61,7 +60,7 @@ func (fep *filterExactPrefix) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether filter matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		if !matchExactPrefix("", prefix) {
diff --git a/lib/logstorage/filter_in.go b/lib/logstorage/filter_in.go
index 0c062f28d..fa3855e7e 100644
--- a/lib/logstorage/filter_in.go
+++ b/lib/logstorage/filter_in.go
@@ -358,8 +358,7 @@ func (fi *filterIn) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 		return
 	}
 
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		stringValues := fi.getStringValues()
 		if _, ok := stringValues[v]; !ok {
@@ -369,7 +368,7 @@ func (fi *filterIn) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether filter matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		// It matches anything only for empty phrase.
diff --git a/lib/logstorage/filter_ipv4_range.go b/lib/logstorage/filter_ipv4_range.go
index a382cd5de..8c9b6fb57 100644
--- a/lib/logstorage/filter_ipv4_range.go
+++ b/lib/logstorage/filter_ipv4_range.go
@@ -102,8 +102,7 @@ func (fr *filterIPv4Range) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 		return
 	}
 
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchIPv4Range(v, minValue, maxValue) {
 			bm.resetBits()
@@ -112,7 +111,7 @@ func (fr *filterIPv4Range) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether filter matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		bm.resetBits()
diff --git a/lib/logstorage/filter_len_range.go b/lib/logstorage/filter_len_range.go
index b80ab877e..d6fb7dad0 100644
--- a/lib/logstorage/filter_len_range.go
+++ b/lib/logstorage/filter_len_range.go
@@ -125,8 +125,7 @@ func (fr *filterLenRange) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 		return
 	}
 
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchLenRange(v, minLen, maxLen) {
 			bm.resetBits()
@@ -135,7 +134,7 @@ func (fr *filterLenRange) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether filter matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		if !matchLenRange("", minLen, maxLen) {
diff --git a/lib/logstorage/filter_or.go b/lib/logstorage/filter_or.go
index 7e67717ef..1987ee9f4 100644
--- a/lib/logstorage/filter_or.go
+++ b/lib/logstorage/filter_or.go
@@ -93,8 +93,7 @@ func (fo *filterOr) matchBloomFilters(bs *blockSearch) bool {
 		fieldName := ft.field
 		tokens := ft.tokens
 
-		csh := bs.getColumnsHeader()
-		v := csh.getConstColumnValue(fieldName)
+		v := bs.getConstColumnValue(fieldName)
 		if v != "" {
 			if matchStringByAllTokens(v, tokens) {
 				return true
@@ -102,7 +101,7 @@ func (fo *filterOr) matchBloomFilters(bs *blockSearch) bool {
 			continue
 		}
 
-		ch := csh.getColumnHeader(fieldName)
+		ch := bs.getColumnHeader(fieldName)
 		if ch == nil {
 			continue
 		}
diff --git a/lib/logstorage/filter_phrase.go b/lib/logstorage/filter_phrase.go
index 676711b52..f25f1e971 100644
--- a/lib/logstorage/filter_phrase.go
+++ b/lib/logstorage/filter_phrase.go
@@ -61,8 +61,7 @@ func (fp *filterPhrase) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	phrase := fp.phrase
 
 	// Verify whether fp matches const column
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchPhrase(v, phrase) {
 			bm.resetBits()
@@ -71,7 +70,7 @@ func (fp *filterPhrase) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether fp matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		// It matches anything only for empty phrase.
diff --git a/lib/logstorage/filter_prefix.go b/lib/logstorage/filter_prefix.go
index a6a086c00..50897a090 100644
--- a/lib/logstorage/filter_prefix.go
+++ b/lib/logstorage/filter_prefix.go
@@ -59,8 +59,7 @@ func (fp *filterPrefix) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	prefix := fp.prefix
 
 	// Verify whether fp matches const column
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchPrefix(v, prefix) {
 			bm.resetBits()
@@ -69,7 +68,7 @@ func (fp *filterPrefix) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether fp matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		bm.resetBits()
diff --git a/lib/logstorage/filter_range.go b/lib/logstorage/filter_range.go
index b173c2deb..a4ed1e936 100644
--- a/lib/logstorage/filter_range.go
+++ b/lib/logstorage/filter_range.go
@@ -173,8 +173,7 @@ func (fr *filterRange) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 		return
 	}
 
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchRange(v, minValue, maxValue) {
 			bm.resetBits()
@@ -183,7 +182,7 @@ func (fr *filterRange) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether filter matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		bm.resetBits()
diff --git a/lib/logstorage/filter_regexp.go b/lib/logstorage/filter_regexp.go
index de9121ea1..1f71ea9c9 100644
--- a/lib/logstorage/filter_regexp.go
+++ b/lib/logstorage/filter_regexp.go
@@ -78,8 +78,7 @@ func (fr *filterRegexp) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	re := fr.re
 
 	// Verify whether filter matches const column
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !re.MatchString(v) {
 			bm.resetBits()
@@ -88,7 +87,7 @@ func (fr *filterRegexp) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether filter matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		if !re.MatchString("") {
diff --git a/lib/logstorage/filter_sequence.go b/lib/logstorage/filter_sequence.go
index fd8cdccc5..2eec23a58 100644
--- a/lib/logstorage/filter_sequence.go
+++ b/lib/logstorage/filter_sequence.go
@@ -87,8 +87,7 @@ func (fs *filterSequence) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 		return
 	}
 
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchSequence(v, phrases) {
 			bm.resetBits()
@@ -97,7 +96,7 @@ func (fs *filterSequence) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether filter matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		// Fast path - there are no matching columns.
 		// It matches anything only for empty phrase.
diff --git a/lib/logstorage/filter_string_range.go b/lib/logstorage/filter_string_range.go
index 071ade0be..988aced50 100644
--- a/lib/logstorage/filter_string_range.go
+++ b/lib/logstorage/filter_string_range.go
@@ -52,8 +52,7 @@ func (fr *filterStringRange) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 		return
 	}
 
-	csh := bs.getColumnsHeader()
-	v := csh.getConstColumnValue(fieldName)
+	v := bs.getConstColumnValue(fieldName)
 	if v != "" {
 		if !matchStringRange(v, minValue, maxValue) {
 			bm.resetBits()
@@ -62,7 +61,7 @@ func (fr *filterStringRange) applyToBlockSearch(bs *blockSearch, bm *bitmap) {
 	}
 
 	// Verify whether filter matches other columns
-	ch := csh.getColumnHeader(fieldName)
+	ch := bs.getColumnHeader(fieldName)
 	if ch == nil {
 		if !matchStringRange("", minValue, maxValue) {
 			bm.resetBits()
diff --git a/lib/logstorage/partition.go b/lib/logstorage/partition.go
index fc88510ec..439ca002f 100644
--- a/lib/logstorage/partition.go
+++ b/lib/logstorage/partition.go
@@ -195,3 +195,8 @@ func (pt *partition) updateStats(ps *PartitionStats) {
 	pt.ddb.updateStats(&ps.DatadbStats)
 	pt.idb.updateStats(&ps.IndexdbStats)
 }
+
+// mustForceMerge runs forced merge for all the parts in pt.
+func (pt *partition) mustForceMerge() {
+	pt.ddb.mustForceMergeAllParts()
+}
diff --git a/lib/logstorage/rows.go b/lib/logstorage/rows.go
index 9d0d77e9c..a90e54a08 100644
--- a/lib/logstorage/rows.go
+++ b/lib/logstorage/rows.go
@@ -36,7 +36,7 @@ func (f *Field) marshal(dst []byte) []byte {
 	return dst
 }
 
-func (f *Field) unmarshal(a *arena, src []byte) ([]byte, error) {
+func (f *Field) unmarshalNoArena(src []byte) ([]byte, error) {
 	srcOrig := src
 
 	// Unmarshal field name
@@ -45,7 +45,7 @@ func (f *Field) unmarshal(a *arena, src []byte) ([]byte, error) {
 		return srcOrig, fmt.Errorf("cannot unmarshal field name")
 	}
 	src = src[nSize:]
-	f.Name = a.copyBytesToString(b)
+	f.Name = bytesutil.ToUnsafeString(b)
 
 	// Unmarshal field value
 	b, nSize = encoding.UnmarshalBytes(src)
@@ -53,7 +53,7 @@ func (f *Field) unmarshal(a *arena, src []byte) ([]byte, error) {
 		return srcOrig, fmt.Errorf("cannot unmarshal field value")
 	}
 	src = src[nSize:]
-	f.Value = a.copyBytesToString(b)
+	f.Value = bytesutil.ToUnsafeString(b)
 
 	return src, nil
 }
diff --git a/lib/logstorage/storage.go b/lib/logstorage/storage.go
index b729ffe30..a448bacc4 100644
--- a/lib/logstorage/storage.go
+++ b/lib/logstorage/storage.go
@@ -4,6 +4,7 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -113,6 +114,8 @@ type Storage struct {
 	// partitions is a list of partitions for the Storage.
 	//
 	// It must be accessed under partitionsLock.
+	//
+	// partitions are sorted by time.
 	partitions []*partitionWrapper
 
 	// ptwHot is the "hot" partition, were the last rows were ingested.
@@ -472,6 +475,33 @@ func (s *Storage) MustClose() {
 	s.path = ""
 }
 
+// MustForceMerge force-merges parts in s partitions with names starting from the given partitionNamePrefix.
+//
+// Partitions are merged sequentially in order to reduce load on the system.
+func (s *Storage) MustForceMerge(partitionNamePrefix string) {
+	var ptws []*partitionWrapper
+
+	s.partitionsLock.Lock()
+	for _, ptw := range s.partitions {
+		if strings.HasPrefix(ptw.pt.name, partitionNamePrefix) {
+			ptw.incRef()
+			ptws = append(ptws, ptw)
+		}
+	}
+	s.partitionsLock.Unlock()
+
+	s.wg.Add(1)
+	defer s.wg.Done()
+
+	for _, ptw := range ptws {
+		logger.Infof("started force merge for partition %s", ptw.pt.name)
+		startTime := time.Now()
+		ptw.pt.mustForceMerge()
+		ptw.decRef()
+		logger.Infof("finished force merge for partition %s in %.3fs", ptw.pt.name, time.Since(startTime).Seconds())
+	}
+}
+
 // MustAddRows adds lr to s.
 //
 // It is recommended checking whether the s is in read-only mode by calling IsReadOnly()
diff --git a/lib/logstorage/storage_search.go b/lib/logstorage/storage_search.go
index 941c195c8..2e269f85e 100644
--- a/lib/logstorage/storage_search.go
+++ b/lib/logstorage/storage_search.go
@@ -685,6 +685,10 @@ func (s *Storage) search(workersCount int, so *genericSearchOptions, stopCh <-ch
 		return ptws[i].day > maxDay
 	})
 	ptws = ptws[:n]
+
+	// Copy the selected partitions, so they don't interfere with s.partitions.
+	ptws = append([]*partitionWrapper{}, ptws...)
+
 	for _, ptw := range ptws {
 		ptw.incRef()
 	}
diff --git a/lib/logstorage/values_encoder.go b/lib/logstorage/values_encoder.go
index dc7dfebee..9185f8a74 100644
--- a/lib/logstorage/values_encoder.go
+++ b/lib/logstorage/values_encoder.go
@@ -1116,10 +1116,10 @@ func (vd *valuesDict) marshal(dst []byte) []byte {
 	return dst
 }
 
-// unmarshal unmarshals vd from src.
+// unmarshalNoArena unmarshals vd from src.
 //
-// vd is valid until a.reset() is called.
-func (vd *valuesDict) unmarshal(a *arena, src []byte) ([]byte, error) {
+// vd is valid until src is changed.
+func (vd *valuesDict) unmarshalNoArena(src []byte) ([]byte, error) {
 	vd.reset()
 
 	srcOrig := src
@@ -1135,7 +1135,7 @@ func (vd *valuesDict) unmarshal(a *arena, src []byte) ([]byte, error) {
 		}
 		src = src[nSize:]
 
-		v := a.copyBytesToString(data)
+		v := bytesutil.ToUnsafeString(data)
 		vd.values = append(vd.values, v)
 	}
 	return src, nil


Test patch:
diff --git a/lib/logstorage/block_header_test.go b/lib/logstorage/block_header_test.go
index bec76b3de..a358f25cc 100644
--- a/lib/logstorage/block_header_test.go
+++ b/lib/logstorage/block_header_test.go
@@ -56,15 +56,12 @@ func TestColumnsHeaderMarshalUnmarshal(t *testing.T) {
 	f := func(csh *columnsHeader, marshaledLen int) {
 		t.Helper()
 
-		a := getArena()
-		defer putArena(a)
-
 		data := csh.marshal(nil)
 		if len(data) != marshaledLen {
 			t.Fatalf("unexpected lengths of the marshaled columnsHeader; got %d; want %d", len(data), marshaledLen)
 		}
 		csh2 := &columnsHeader{}
-		err := csh2.unmarshal(a, data)
+		err := csh2.unmarshalNoArena(data)
 		if err != nil {
 			t.Fatalf("unexpected error in unmarshal: %s", err)
 		}
@@ -155,12 +152,9 @@ func TestColumnsHeaderUnmarshalFailure(t *testing.T) {
 	f := func(data []byte) {
 		t.Helper()
 
-		a := getArena()
-		defer putArena(a)
-
 		csh := getColumnsHeader()
 		defer putColumnsHeader(csh)
-		err := csh.unmarshal(a, data)
+		err := csh.unmarshalNoArena(data)
 		if err == nil {
 			t.Fatalf("expecting non-nil error")
 		}
@@ -326,15 +320,12 @@ func TestColumnHeaderMarshalUnmarshal(t *testing.T) {
 	f := func(ch *columnHeader, marshaledLen int) {
 		t.Helper()
 
-		a := getArena()
-		defer putArena(a)
-
 		data := ch.marshal(nil)
 		if len(data) != marshaledLen {
 			t.Fatalf("unexpected marshaled length of columnHeader; got %d; want %d", len(data), marshaledLen)
 		}
 		var ch2 columnHeader
-		tail, err := ch2.unmarshal(a, data)
+		tail, err := ch2.unmarshalNoArena(data)
 		if err != nil {
 			t.Fatalf("unexpected error in umarshal(%v): %s", ch, err)
 		}
@@ -365,12 +356,9 @@ func TestColumnHeaderUnmarshalFailure(t *testing.T) {
 	f := func(data []byte) {
 		t.Helper()
 
-		a := getArena()
-		defer putArena(a)
-
 		dataOrig := append([]byte{}, data...)
 		var ch columnHeader
-		tail, err := ch.unmarshal(a, data)
+		tail, err := ch.unmarshalNoArena(data)
 		if err == nil {
 			t.Fatalf("expecting non-nil error")
 		}