Repository: VictoriaMetrics/VictoriaMetrics
Pull request #7249 (2024-10-13 08:29:29)

docs/victorialogs: fixed HA examples links
### Describe Your Changes

Fixed VictoriaLogs HA examples references in docs

### Checklist

The following checks are **mandatory**:

- [ ] 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/VictoriaLogs/CHANGELOG.md b/docs/VictoriaLogs/CHANGELOG.md
index cece5d096..cfedbe96f 100644
--- a/docs/VictoriaLogs/CHANGELOG.md
+++ b/docs/VictoriaLogs/CHANGELOG.md
@@ -15,6 +15,9 @@ 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).
 
 
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 c9c5ad722..cd8d53b72 100644
--- a/docs/VictoriaLogs/README.md
+++ b/docs/VictoriaLogs/README.md
@@ -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/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 c95fe8597..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, "")
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")
 		}