diff --git a/Cargo.lock b/Cargo.lock
index aa145766e9..05420c101d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5163,9 +5163,9 @@ dependencies = [
[[package]]
name = "h2"
-version = "0.4.15"
+version = "0.4.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155"
+checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27"
dependencies = [
"atomic-waker",
"bytes",
@@ -9163,11 +9163,13 @@ version = "0.1.0"
dependencies = [
"arrow",
"async-trait",
+ "bytes",
"config",
"datafusion",
"db",
"futures",
"hashbrown 0.16.1",
+ "hashlink 0.11.1",
"infra",
"itertools 0.14.0",
"log",
@@ -9177,6 +9179,7 @@ dependencies = [
"prost 0.14.4",
"proto",
"rayon",
+ "regex",
"search",
"search_service",
"serde",
diff --git a/src/compaction/src/merge/metrics.rs b/src/compaction/src/merge/metrics.rs
new file mode 100644
index 0000000000..cba2293a8d
--- /dev/null
+++ b/src/compaction/src/merge/metrics.rs
@@ -0,0 +1,134 @@
+// Copyright 2026 OpenObserve Inc.
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+//! What the compactor merges in a closed indexed metrics hour: nothing when
+//! every file is indexed, the late files alone while fragmentation is low,
+//! else the whole hour.
+
+use config::meta::{promql::layout::MetricsFileLayout, stream::FileKey};
+
+/// Indexed `indexed-v1-*` files a closed hour may hold beyond the ideal
+/// `total_size / max_file_size` before the whole hour is rewritten.
+const METRICS_INDEX_REWRITE_SLACK: usize = 4;
+
+/// What the merge of a closed indexed metrics hour covers.
+#[derive(Debug, PartialEq, Eq)]
+pub(super) enum MetricsIndexMergeScope {
+ /// Every file is `indexed-v1-*` and the count fits the current
+ /// `max_file_size`: leave the hour alone.
+ Skip,
+ /// Merge only the late, non-indexed files into one new indexed file;
+ /// existing indexed files stay. Their hash ranges may overlap, which
+ /// readers handle per file.
+ LateFilesOnly,
+ /// Too many indexed files: rewrite the whole hour. Reached by late
+ /// merges piling up fragments (amortizes their full-rewrite cost) or by
+ /// a raised `max_file_size` shrinking the ideal count (such hours are
+ /// re-enqueued by the old-data sweep).
+ WholeHour,
+}
+
+pub(super) fn metrics_index_merge_scope(
+ files: &[FileKey],
+ max_file_size: usize,
+) -> MetricsIndexMergeScope {
+ let indexed_files = files
+ .iter()
+ .filter(|f| MetricsFileLayout::of(&f.key) == MetricsFileLayout::Indexed)
+ .count();
+ let total_original_size: i64 = files.iter().map(|f| f.meta.original_size.max(0)).sum();
+ let ideal_file_count = (total_original_size as usize).div_ceil(max_file_size.max(1));
+ if indexed_files > ideal_file_count + METRICS_INDEX_REWRITE_SLACK {
+ MetricsIndexMergeScope::WholeHour
+ } else if indexed_files == files.len() {
+ MetricsIndexMergeScope::Skip
+ } else {
+ MetricsIndexMergeScope::LateFilesOnly
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use config::meta::stream::FileMeta;
+
+ use super::*;
+
+ fn file_key(name: &str, original_size: i64) -> FileKey {
+ let mut key = FileKey::from_file_name(&format!("files/o/metrics/m/2026/08/18/10/{name}"));
+ key.meta = FileMeta {
+ min_ts: 10,
+ max_ts: 20,
+ records: 100,
+ original_size,
+ ..Default::default()
+ };
+ key
+ }
+
+ #[test]
+ fn test_metrics_index_merge_scope() {
+ let max_file_size = 100_usize;
+ let indexed = |id: usize| file_key(&format!("indexed-v1-{id}.parquet"), 10);
+ let late = |id: usize| file_key(&format!("hash-sorted-v1-{id}.parquet"), 10);
+
+ // every file indexed and acceptably sized: the hour is left alone
+ assert_eq!(
+ metrics_index_merge_scope(&[indexed(1), indexed(2)], max_file_size),
+ MetricsIndexMergeScope::Skip
+ );
+ // max_file_size raised (files written at a smaller target): a fully indexed
+ // hour above the cap is rewritten to the new size, no late data needed
+ let raised: Vec = (1..=10).map(indexed).collect();
+ assert_eq!(
+ metrics_index_merge_scope(&raised, max_file_size),
+ MetricsIndexMergeScope::WholeHour
+ );
+ // same files under the original target (ideal 10): left alone
+ assert_eq!(
+ metrics_index_merge_scope(&raised, 10),
+ MetricsIndexMergeScope::Skip
+ );
+ // no indexed files yet (first hour-end pass): merge everything
+ assert_eq!(
+ metrics_index_merge_scope(&[late(1), late(2)], max_file_size),
+ MetricsIndexMergeScope::LateFilesOnly
+ );
+ // few indexed files: only the late ones merge
+ assert_eq!(
+ metrics_index_merge_scope(
+ &[indexed(1), indexed(2), indexed(3), late(4)],
+ max_file_size,
+ ),
+ MetricsIndexMergeScope::LateFilesOnly
+ );
+ // fragmentation cap: 8 indexed files, total size 90 -> ideal 1,
+ // 8 > 1 + METRICS_INDEX_REWRITE_SLACK -> rewrite the whole hour
+ let mut fragmented: Vec = (1..=8).map(indexed).collect();
+ fragmented.push(late(9));
+ assert_eq!(
+ metrics_index_merge_scope(&fragmented, max_file_size),
+ MetricsIndexMergeScope::WholeHour
+ );
+ // same file count but a large hour (ideal 9): still a late-only merge
+ let mut large: Vec = (1..=8)
+ .map(|id| file_key(&format!("indexed-v1-{id}.parquet"), 100))
+ .collect();
+ large.push(late(9));
+ assert_eq!(
+ metrics_index_merge_scope(&large, max_file_size),
+ MetricsIndexMergeScope::LateFilesOnly
+ );
+ }
+}
diff --git a/src/compaction/src/merge/mod.rs b/src/compaction/src/merge/mod.rs
index 8d60729dd9..ba2a79393e 100644
--- a/src/compaction/src/merge/mod.rs
+++ b/src/compaction/src/merge/mod.rs
@@ -15,8 +15,8 @@
mod file;
mod job;
+mod metrics;
mod stream;
-mod tsid_major;
pub use file::merge_files;
pub use job::{
diff --git a/src/compaction/src/merge/stream.rs b/src/compaction/src/merge/stream.rs
index e81cfae55c..56e408f440 100644
--- a/src/compaction/src/merge/stream.rs
+++ b/src/compaction/src/merge/stream.rs
@@ -18,7 +18,7 @@ use chrono::{DateTime, TimeZone, Utc};
use config::{
get_config,
meta::{
- promql::tsid_layout::MetricsFileLayout,
+ promql::layout::MetricsFileLayout,
stream::{FileKey, FileListDeleted, MergeStrategy, PartitionTimeLevel, StreamType},
},
metrics,
@@ -35,7 +35,7 @@ use tokio::{
use super::{
job::job_range_end,
- tsid_major::{TsidMajorMergeScope, tsid_major_merge_scope},
+ metrics::{MetricsIndexMergeScope, metrics_index_merge_scope},
};
use crate::worker::{MergeBatch, MergeSender};
@@ -175,22 +175,22 @@ pub async fn merge_by_stream(
if files_with_size.len() <= 1 && !mode.merges_whole_batch() {
return Ok(vec![]);
}
- // what a closed TSID-major hour merges, see [`TsidMajorMergeScope`]
- if mode.is_tsid_major() {
- match tsid_major_merge_scope(&files_with_size, cfg.compact.max_file_size) {
- TsidMajorMergeScope::Skip => return Ok(vec![]),
- TsidMajorMergeScope::LateFilesOnly => {
+ // what a closed indexed metrics hour merges, see [`MetricsIndexMergeScope`]
+ if mode.is_metrics_indexed() {
+ match metrics_index_merge_scope(&files_with_size, cfg.compact.max_file_size) {
+ MetricsIndexMergeScope::Skip => return Ok(vec![]),
+ MetricsIndexMergeScope::LateFilesOnly => {
files_with_size.retain(|f| {
- MetricsFileLayout::of(&f.key) != MetricsFileLayout::TsidMajor
+ MetricsFileLayout::of(&f.key) != MetricsFileLayout::Indexed
});
log::debug!(
- "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] tsid_major late merge of {} files, finalized files untouched",
+ "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] metrics_indexed late merge of {} files, indexed files untouched",
files_with_size.len()
);
}
- TsidMajorMergeScope::WholeHour => {
+ MetricsIndexMergeScope::WholeHour => {
log::debug!(
- "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] tsid_major fragmentation cap hit, full rewrite of {} files",
+ "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] metrics_indexed fragmentation cap hit, full rewrite of {} files",
files_with_size.len()
);
}
diff --git a/src/compaction/src/merge/tsid_major.rs b/src/compaction/src/merge/tsid_major.rs
deleted file mode 100644
index 80ea302202..0000000000
--- a/src/compaction/src/merge/tsid_major.rs
+++ /dev/null
@@ -1,131 +0,0 @@
-// Copyright 2026 OpenObserve Inc.
-//
-// This program is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Affero General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// This program is distributed in the hope that it will be useful
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Affero General Public License for more details.
-//
-// You should have received a copy of the GNU Affero General Public License
-// along with this program. If not, see .
-
-//! What the compactor merges in a closed TSID-major hour: nothing when it is
-//! finalized, the late files alone while fragmentation is low, else the
-//! whole hour.
-
-use config::meta::{promql::tsid_layout::MetricsFileLayout, stream::FileKey};
-
-/// Finalized `tsid-major-v3-*` files a closed hour may hold beyond the ideal
-/// `total_size / max_file_size` before the whole hour is rewritten.
-const TSID_MAJOR_REWRITE_SLACK: usize = 4;
-
-/// What the merge of a closed TSID-major hour covers.
-#[derive(Debug, PartialEq, Eq)]
-pub(super) enum TsidMajorMergeScope {
- /// Every file is `tsid-major-v3-*` and the count fits the current
- /// `max_file_size`: leave the hour alone.
- Skip,
- /// Merge only the late (non-v3) files into one new v3; finalized files
- /// stay. The hour's v3 hash ranges then overlap, which readers handle
- /// per file.
- LateFilesOnly,
- /// Too many finalized files: rewrite the whole hour. Reached by late
- /// merges piling up fragments (amortizes their full-rewrite cost) or by
- /// a raised `max_file_size` shrinking the ideal count (such hours are
- /// re-enqueued by the old-data sweep).
- WholeHour,
-}
-
-pub(super) fn tsid_major_merge_scope(
- files: &[FileKey],
- max_file_size: usize,
-) -> TsidMajorMergeScope {
- let finalized = files
- .iter()
- .filter(|f| MetricsFileLayout::of(&f.key) == MetricsFileLayout::TsidMajor)
- .count();
- let total_original_size: i64 = files.iter().map(|f| f.meta.original_size.max(0)).sum();
- let ideal_file_count = (total_original_size as usize).div_ceil(max_file_size.max(1));
- if finalized > ideal_file_count + TSID_MAJOR_REWRITE_SLACK {
- TsidMajorMergeScope::WholeHour
- } else if finalized == files.len() {
- TsidMajorMergeScope::Skip
- } else {
- TsidMajorMergeScope::LateFilesOnly
- }
-}
-
-#[cfg(test)]
-mod tests {
- use config::meta::stream::FileMeta;
-
- use super::*;
-
- fn file_key(name: &str, original_size: i64) -> FileKey {
- let mut key = FileKey::from_file_name(&format!("files/o/metrics/m/2026/08/18/10/{name}"));
- key.meta = FileMeta {
- min_ts: 10,
- max_ts: 20,
- records: 100,
- original_size,
- ..Default::default()
- };
- key
- }
-
- #[test]
- fn test_tsid_major_merge_scope() {
- let max_file_size = 100_usize;
- let v3 = |id: usize| file_key(&format!("tsid-major-v3-{id}.parquet"), 10);
- let late = |id: usize| file_key(&format!("tsid-sorted-{id}.parquet"), 10);
-
- // every file finalized and acceptably sized: the hour is left alone
- assert_eq!(
- tsid_major_merge_scope(&[v3(1), v3(2)], max_file_size),
- TsidMajorMergeScope::Skip
- );
- // max_file_size raised (files written at a smaller target): an all-v3
- // hour above the cap is rewritten to the new size, no late data needed
- let raised: Vec = (1..=10).map(v3).collect();
- assert_eq!(
- tsid_major_merge_scope(&raised, max_file_size),
- TsidMajorMergeScope::WholeHour
- );
- // same files under the original target (ideal 10): left alone
- assert_eq!(
- tsid_major_merge_scope(&raised, 10),
- TsidMajorMergeScope::Skip
- );
- // no finalized files yet (first hour-end pass): merge everything
- assert_eq!(
- tsid_major_merge_scope(&[late(1), late(2)], max_file_size),
- TsidMajorMergeScope::LateFilesOnly
- );
- // few finalized files: only the late ones merge
- assert_eq!(
- tsid_major_merge_scope(&[v3(1), v3(2), v3(3), late(4)], max_file_size),
- TsidMajorMergeScope::LateFilesOnly
- );
- // fragmentation cap: 8 finalized files, total size 90 -> ideal 1,
- // 8 > 1 + TSID_MAJOR_REWRITE_SLACK -> rewrite the whole hour
- let mut fragmented: Vec = (1..=8).map(v3).collect();
- fragmented.push(late(9));
- assert_eq!(
- tsid_major_merge_scope(&fragmented, max_file_size),
- TsidMajorMergeScope::WholeHour
- );
- // same file count but a large hour (ideal 9): still a late-only merge
- let mut large: Vec = (1..=8)
- .map(|id| file_key(&format!("tsid-major-v3-{id}.parquet"), 100))
- .collect();
- large.push(late(9));
- assert_eq!(
- tsid_major_merge_scope(&large, max_file_size),
- TsidMajorMergeScope::LateFilesOnly
- );
- }
-}
diff --git a/src/config/src/config.rs b/src/config/src/config.rs
index 32ccffdca7..640046febe 100644
--- a/src/config/src/config.rs
+++ b/src/config/src/config.rs
@@ -2628,11 +2628,11 @@ pub struct Compact {
#[env_config(name = "ZO_COMPACT_ENABLED", default = true)]
pub enabled: bool,
#[env_config(
- name = "ZO_METRICS_TSID_MAJOR_ENABLED",
+ name = "ZO_METRICS_INDEX_ENABLED",
default = false,
- help = "Experimental TSID-major metrics layout. The ingester writes Parquet metrics files ordered by (__hash__, _timestamp) instead of _timestamp DESC and marks them with a `tsid-sorted-` file name prefix; the compactor merges a closed hour into size-split `tsid-major-v3-` Parquet files in the same order. Only affects newly written Parquet metrics files of streams whose __hash__ column is UInt64; SQL queries on metrics streams must not assume a _timestamp order while it is on."
+ help = "Experimental indexed metrics layout. The ingester writes Parquet metrics files ordered by (__hash__, _timestamp) instead of _timestamp DESC and marks them with a `hash-sorted-v1-` file name prefix; the compactor merges a closed hour into size-split `indexed-v1-` Parquet files in the same order. Only affects newly written Parquet metrics files of streams whose __hash__ column is UInt64; SQL queries on metrics streams must not assume a _timestamp order while it is on."
)]
- pub metrics_tsid_major_enabled: bool,
+ pub metrics_index_enabled: bool,
#[env_config(name = "ZO_COMPACT_INTERVAL", default = 10)] // seconds
pub interval: u64,
#[env_config(
diff --git a/src/config/src/meta/promql/tsid_layout.rs b/src/config/src/meta/promql/layout.rs
similarity index 73%
rename from src/config/src/meta/promql/tsid_layout.rs
rename to src/config/src/meta/promql/layout.rs
index 068f43085a..38284f779a 100644
--- a/src/config/src/meta/promql/tsid_layout.rs
+++ b/src/config/src/meta/promql/layout.rs
@@ -13,31 +13,30 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see .
-//! The TSID-major metrics layout (`ZO_METRICS_TSID_MAJOR_ENABLED`): when it
-//! applies to a stream and how the physical layout of a metrics file is
-//! encoded in its name.
+//! Indexed metrics layout (`ZO_METRICS_INDEX_ENABLED`): when it applies to a
+//! stream and how the physical layout of a metrics file is encoded in its name.
use arrow_schema::{DataType, Schema};
use super::HASH_LABEL;
use crate::{FileFormat, meta::stream::StreamType};
-/// True when `stream_type` uses the TSID-major layout
-/// (`ZO_METRICS_TSID_MAJOR_ENABLED`): Parquet metrics files ordered by
+/// True when `stream_type` uses the indexed metrics layout
+/// (`ZO_METRICS_INDEX_ENABLED`): Parquet metrics files ordered by
/// `(__hash__, _timestamp)`, so readers must not assume a `_timestamp` order.
-pub fn metrics_tsid_major_enabled(stream_type: StreamType) -> bool {
+pub fn metrics_index_enabled(stream_type: StreamType) -> bool {
if stream_type != StreamType::Metrics {
return false;
}
let cfg = crate::get_config();
- cfg.compact.metrics_tsid_major_enabled
+ cfg.compact.metrics_index_enabled
&& cfg.common.file_format.for_stream(stream_type) == FileFormat::Parquet
}
-/// [`metrics_tsid_major_enabled`] narrowed to one stream: the layout also
+/// [`metrics_index_enabled`] narrowed to one stream: the layout also
/// needs a `__hash__` column of type `UInt64` (remote-write / OTLP metrics).
-pub fn metrics_tsid_major_stream(stream_type: StreamType, schema: &Schema) -> bool {
- metrics_tsid_major_enabled(stream_type)
+pub fn metrics_index_stream(stream_type: StreamType, schema: &Schema) -> bool {
+ metrics_index_enabled(stream_type)
&& schema
.field_with_name(HASH_LABEL)
.is_ok_and(|field| field.data_type() == &DataType::UInt64)
@@ -51,23 +50,23 @@ pub enum MetricsFileLayout {
Legacy,
/// Parquet ordered by `(__hash__ ASC, _timestamp ASC)` but not finalized:
/// written by the ingester and by incremental compactor merges of the
- /// still-open hour (`tsid-sorted-{id}.parquet`).
+ /// still-open hour (`hash-sorted-v1-{id}.parquet`).
HashSorted,
/// Size-bounded Parquet ordered by `(__hash__ ASC, _timestamp ASC)`,
- /// written by the compactor's hour-end merge (`tsid-major-v3-{id}.parquet`).
- TsidMajor,
+ /// written by the compactor's hour-end merge (`indexed-v1-{id}.parquet`).
+ Indexed,
}
impl MetricsFileLayout {
- const HASH_SORTED_PREFIX: &'static str = "tsid-sorted-";
- const TSID_MAJOR_PREFIX: &'static str = "tsid-major-v3-";
+ const HASH_SORTED_PREFIX: &'static str = "hash-sorted-v1-";
+ const INDEXED_PREFIX: &'static str = "indexed-v1-";
/// Layout of the file at `path` (a full object key or a bare file name).
pub fn of(path: &str) -> Self {
let file_name = path.rsplit('/').next().unwrap_or(path);
for (prefix, layout) in [
(Self::HASH_SORTED_PREFIX, Self::HashSorted),
- (Self::TSID_MAJOR_PREFIX, Self::TsidMajor),
+ (Self::INDEXED_PREFIX, Self::Indexed),
] {
if let Some(id) = file_name.strip_prefix(prefix)
&& let Some(id) = id.strip_suffix(".parquet")
@@ -83,7 +82,7 @@ impl MetricsFileLayout {
match self {
Self::Legacy => "",
Self::HashSorted => Self::HASH_SORTED_PREFIX,
- Self::TsidMajor => Self::TSID_MAJOR_PREFIX,
+ Self::Indexed => Self::INDEXED_PREFIX,
}
}
@@ -102,7 +101,7 @@ impl MetricsFileLayout {
}
/// Add this layout's marker to the file name of an existing key
- /// (`files/.../7099.parquet` -> `files/.../tsid-sorted-7099.parquet`).
+ /// (`files/.../7099.parquet` -> `files/.../hash-sorted-v1-7099.parquet`).
pub fn mark_file_key(self, key: &str) -> String {
match key.rfind('/') {
Some(pos) => format!("{}/{}{}", &key[..pos], self.prefix(), &key[pos + 1..]),
@@ -129,25 +128,25 @@ mod metrics_file_layout_tests {
);
assert_eq!(
MetricsFileLayout::of(
- "files/default/metrics/cpu/2026/08/18/10/tsid-sorted-7099.parquet"
+ "files/default/metrics/cpu/2026/08/18/10/hash-sorted-v1-7099.parquet"
),
MetricsFileLayout::HashSorted
);
assert_eq!(
MetricsFileLayout::of(
- "files/default/metrics/test/2026/08/13/10/tsid-major-v3-456.parquet"
+ "files/default/metrics/test/2026/08/13/10/indexed-v1-456.parquet"
),
- MetricsFileLayout::TsidMajor
+ MetricsFileLayout::Indexed
);
// wrong format, empty id, other versions, marker in a directory name
for name in [
- "tsid-major-v3-x.vortex",
- "tsid-sorted-1.vortex",
- "tsid-major-v3-.parquet",
- "tsid-sorted-.parquet",
- "tsid-major-v1-x.parquet",
- "tsid-range-v3-b04-p000a-x.parquet",
- "files/tsid-sorted-dir/1.parquet",
+ "indexed-v1-x.vortex",
+ "hash-sorted-v1-1.vortex",
+ "indexed-v1-.parquet",
+ "hash-sorted-v1-.parquet",
+ "metrics-indexed-v2-x.parquet",
+ "metrics-range-v1-b04-p000a-x.parquet",
+ "files/hash-sorted-v1-dir/1.parquet",
] {
assert_eq!(
MetricsFileLayout::of(name),
@@ -156,19 +155,19 @@ mod metrics_file_layout_tests {
);
}
assert!(MetricsFileLayout::HashSorted.is_hash_ordered());
- assert!(MetricsFileLayout::TsidMajor.is_hash_ordered());
+ assert!(MetricsFileLayout::Indexed.is_hash_ordered());
assert!(!MetricsFileLayout::Legacy.is_hash_ordered());
}
#[test]
fn builds_and_marks_file_names() {
assert_eq!(
- MetricsFileLayout::TsidMajor.file_name("456", FileFormat::Parquet),
- "tsid-major-v3-456.parquet"
+ MetricsFileLayout::Indexed.file_name("456", FileFormat::Parquet),
+ "indexed-v1-456.parquet"
);
assert_eq!(
MetricsFileLayout::HashSorted.file_name("456", FileFormat::Parquet),
- "tsid-sorted-456.parquet"
+ "hash-sorted-v1-456.parquet"
);
assert_eq!(
MetricsFileLayout::Legacy.file_name("456", FileFormat::Vortex),
@@ -178,7 +177,7 @@ mod metrics_file_layout_tests {
let marked = MetricsFileLayout::HashSorted.mark_file_key(key);
assert_eq!(
marked,
- "files/default/metrics/cpu/2026/08/18/10/tsid-sorted-7099.parquet"
+ "files/default/metrics/cpu/2026/08/18/10/hash-sorted-v1-7099.parquet"
);
assert_eq!(
MetricsFileLayout::of(&marked),
@@ -187,7 +186,7 @@ mod metrics_file_layout_tests {
assert_eq!(MetricsFileLayout::Legacy.mark_file_key(key), key);
assert_eq!(
MetricsFileLayout::HashSorted.mark_file_key("1.parquet"),
- "tsid-sorted-1.parquet"
+ "hash-sorted-v1-1.parquet"
);
}
}
diff --git a/src/config/src/meta/promql/mod.rs b/src/config/src/meta/promql/mod.rs
index dd851bbe88..93013e4872 100644
--- a/src/config/src/meta/promql/mod.rs
+++ b/src/config/src/meta/promql/mod.rs
@@ -66,7 +66,7 @@ where
}
pub mod grpc;
-pub mod tsid_layout;
+pub mod layout;
pub mod value;
pub const NAME_LABEL: &str = "__name__";
diff --git a/src/jobs/src/job/files/pack.rs b/src/jobs/src/job/files/pack.rs
index c076118d6a..64676fb91f 100644
--- a/src/jobs/src/job/files/pack.rs
+++ b/src/jobs/src/job/files/pack.rs
@@ -27,7 +27,7 @@ use chrono::Duration;
use config::{
FileFormat, cluster, get_config,
meta::{
- promql::tsid_layout::MetricsFileLayout,
+ promql::layout::MetricsFileLayout,
stream::{FileMeta, StreamType},
},
metrics,
diff --git a/src/promql_service/Cargo.toml b/src/promql_service/Cargo.toml
index 411d5f960b..e146bc435d 100644
--- a/src/promql_service/Cargo.toml
+++ b/src/promql_service/Cargo.toml
@@ -21,11 +21,13 @@ vectorscan = ["search/vectorscan", "search_service/vectorscan"]
[dependencies]
arrow.workspace = true
async-trait.workspace = true
+bytes.workspace = true
config.workspace = true
datafusion.workspace = true
db.workspace = true
futures.workspace = true
hashbrown.workspace = true
+hashlink.workspace = true
infra.workspace = true
itertools.workspace = true
log.workspace = true
@@ -44,3 +46,6 @@ tracing.workspace = true
tracing-opentelemetry.workspace = true
usage_reporting.workspace = true
utoipa.workspace = true
+
+[dev-dependencies]
+regex.workspace = true
diff --git a/src/search/src/datafusion/merge/downsampling.rs b/src/search/src/datafusion/merge/downsampling.rs
index 5b6ffa3b3d..ca4da0e979 100644
--- a/src/search/src/datafusion/merge/downsampling.rs
+++ b/src/search/src/datafusion/merge/downsampling.rs
@@ -19,9 +19,7 @@ use arrow::array::{Int64Array, RecordBatch};
use config::{
FileFormat, TIMESTAMP_COL_NAME, get_config,
meta::{
- promql::{
- DownsamplingRule, Function, HASH_LABEL, VALUE_LABEL, tsid_layout::MetricsFileLayout,
- },
+ promql::{DownsamplingRule, Function, HASH_LABEL, VALUE_LABEL, layout::MetricsFileLayout},
stream::FileMeta,
},
utils::parquet::new_parquet_writer,
diff --git a/src/search/src/datafusion/merge/tsid_major.rs b/src/search/src/datafusion/merge/metrics.rs
similarity index 92%
rename from src/search/src/datafusion/merge/tsid_major.rs
rename to src/search/src/datafusion/merge/metrics.rs
index 5f213a5dcd..f8419890c0 100644
--- a/src/search/src/datafusion/merge/tsid_major.rs
+++ b/src/search/src/datafusion/merge/metrics.rs
@@ -21,7 +21,7 @@ use arrow::{
};
use config::{
TIMESTAMP_COL_NAME,
- meta::{promql::tsid_layout::MetricsFileLayout, stream::FileMeta},
+ meta::{promql::layout::MetricsFileLayout, stream::FileMeta},
utils::parquet::new_parquet_writer,
};
use datafusion::{
@@ -34,7 +34,7 @@ use super::{MergedFile, append_metadata};
/// Write a globally hash-sorted metrics stream into size-bounded files.
/// Rotation happens between input record batches. File and row-group
-/// boundaries deliberately remain independent of TSID boundaries.
+/// boundaries deliberately remain independent of metrics series boundaries.
pub(super) async fn write_files(
schema: &Arc,
bloom_filter_fields: &[String],
@@ -45,11 +45,11 @@ pub(super) async fn write_files(
) -> Result> {
let timestamp_index = schema.index_of(TIMESTAMP_COL_NAME).map_err(|e| {
DataFusionError::Plan(format!(
- "TSID-major layout requires {TIMESTAMP_COL_NAME}: {e}"
+ "indexed metrics layout requires {TIMESTAMP_COL_NAME}: {e}"
))
})?;
let max_file_size = i64::try_from(max_file_size).unwrap_or(i64::MAX).max(1);
- let mut active: Option = None;
+ let mut active: Option = None;
let mut files: Vec = Vec::new();
while let Some(batch) = rx.recv().await {
@@ -63,7 +63,7 @@ pub(super) async fn write_files(
files.push(full.finish(metadata, max_file_size).await?);
}
let writer = active.get_or_insert_with(|| {
- ActiveSizeTsidWriter::new(schema, bloom_filter_fields, metadata, timestamp_index)
+ ActiveIndexedMetricsWriter::new(schema, bloom_filter_fields, metadata, timestamp_index)
});
writer.write(&batch).await?;
}
@@ -76,7 +76,7 @@ pub(super) async fn write_files(
}
if files.is_empty() {
return Err(DataFusionError::Execution(
- "TSID-major merge produced no rows".to_string(),
+ "indexed metrics merge produced no rows".to_string(),
));
}
@@ -84,13 +84,13 @@ pub(super) async fn write_files(
}
/// One output file in progress: the Parquet writer and the running file meta.
-struct ActiveSizeTsidWriter {
+struct ActiveIndexedMetricsWriter {
writer: AsyncArrowWriter>,
file_meta: FileMeta,
timestamp_index: usize,
}
-impl ActiveSizeTsidWriter {
+impl ActiveIndexedMetricsWriter {
fn new(
schema: &Arc,
bloom_filter_fields: &[String],
@@ -123,7 +123,7 @@ impl ActiveSizeTsidWriter {
.downcast_ref::()
.ok_or_else(|| {
DataFusionError::Plan(format!(
- "TSID-major layout requires Int64 {TIMESTAMP_COL_NAME}"
+ "indexed metrics layout requires Int64 {TIMESTAMP_COL_NAME}"
))
})?;
let meta = &mut self.file_meta;
@@ -157,7 +157,7 @@ impl ActiveSizeTsidWriter {
Ok(MergedFile {
buf,
meta: file_meta,
- layout: MetricsFileLayout::TsidMajor,
+ layout: MetricsFileLayout::Indexed,
})
}
}
@@ -181,7 +181,7 @@ mod tests {
use super::*;
#[tokio::test]
- async fn test_size_split_tsid_major_rotates_at_batch_boundary() {
+ async fn test_size_split_metrics_rotates_at_batch_boundary() {
let schema = Arc::new(Schema::new(vec![
Field::new(HASH_LABEL, DataType::UInt64, false),
Field::new(TIMESTAMP_COL_NAME, DataType::Int64, false),
@@ -254,7 +254,7 @@ mod tests {
vec![150, 150, 100]
);
assert!(files.iter().all(|f| {
- f.layout == MetricsFileLayout::TsidMajor
+ f.layout == MetricsFileLayout::Indexed
&& f.meta.original_size < max_file_size as i64
&& f.meta.compressed_size == f.buf.len() as i64
}));
diff --git a/src/search/src/datafusion/merge/mod.rs b/src/search/src/datafusion/merge/mod.rs
index df71726199..3858dec78f 100644
--- a/src/search/src/datafusion/merge/mod.rs
+++ b/src/search/src/datafusion/merge/mod.rs
@@ -18,7 +18,7 @@ use std::sync::Arc;
use arrow::array::RecordBatch;
use config::{
FileFormat, get_config,
- meta::{promql::tsid_layout::MetricsFileLayout, stream::FileMeta},
+ meta::{promql::layout::MetricsFileLayout, stream::FileMeta},
utils::parquet::{VORTEX_FILE_META_KEY, encode_vortex_file_meta, new_parquet_writer},
};
use datafusion::{
@@ -51,8 +51,8 @@ use crate::datafusion::{
#[cfg(feature = "enterprise")]
pub mod downsampling;
+mod metrics;
pub mod mode;
-mod tsid_major;
pub use mode::{MergeMode, MergeOutput};
@@ -116,8 +116,8 @@ pub async fn merge_parquet_files(
)
.await?
}
- MergeMode::TsidMajor => {
- tsid_major::write_files(
+ MergeMode::MetricsIndexed => {
+ metrics::write_files(
&schema,
bloom_filter_fields,
&metadata,
diff --git a/src/search/src/datafusion/merge/mode.rs b/src/search/src/datafusion/merge/mode.rs
index 12a68d7810..23fb07cec9 100644
--- a/src/search/src/datafusion/merge/mode.rs
+++ b/src/search/src/datafusion/merge/mode.rs
@@ -30,7 +30,7 @@ use config::meta::promql::DownsamplingRule;
use config::{
FileFormat, FileFormatConfig, TIMESTAMP_COL_NAME, get_config,
meta::{
- promql::tsid_layout::{MetricsFileLayout, metrics_tsid_major_stream},
+ promql::layout::{MetricsFileLayout, metrics_index_stream},
stream::{FileKey, StreamType},
},
utils::util::is_trace_time_index_stream,
@@ -56,13 +56,13 @@ pub enum MergeMode {
/// as a whole.
#[cfg(feature = "enterprise")]
Downsampling(DownsamplingRule),
- /// TSID-major metrics stream, hour still open: the ingester and the
- /// incremental compactor merges write one `tsid-sorted-*` Parquet file
+ /// Metrics index stream, hour still open: the ingester and the incremental
+ /// compactor merges write one `hash-sorted-v1-*` Parquet file
/// ordered by `(__hash__, _timestamp)`.
- HashSorted,
- /// TSID-major metrics stream, closed hour: the whole hour merges into
- /// size-split `tsid-major-v3-*` files in the same order.
- TsidMajor,
+ MetricsHashSorted,
+ /// Metrics index stream, closed hour: the whole hour merges into
+ /// size-split `indexed-v1-*` files in the same order.
+ MetricsIndexed,
}
impl MergeMode {
@@ -85,11 +85,11 @@ impl MergeMode {
{
return Self::Downsampling(rule.clone());
}
- if metrics_tsid_major_stream(stream_type, schema) {
+ if metrics_index_stream(stream_type, schema) {
return if finalize {
- Self::TsidMajor
+ Self::MetricsIndexed
} else {
- Self::HashSorted
+ Self::MetricsHashSorted
};
}
Self::for_stream(stream_type, stream_name)
@@ -97,8 +97,8 @@ impl MergeMode {
/// Mode for the ingester movers: never downsamples, never finalizes.
pub fn for_ingester(stream_type: StreamType, stream_name: &str, schema: &Schema) -> Self {
- if metrics_tsid_major_stream(stream_type, schema) {
- return Self::HashSorted;
+ if metrics_index_stream(stream_type, schema) {
+ return Self::MetricsHashSorted;
}
Self::for_stream(stream_type, stream_name)
}
@@ -111,9 +111,9 @@ impl MergeMode {
}
}
- /// True for the TSID-major hour-end merge.
- pub fn is_tsid_major(&self) -> bool {
- matches!(self, Self::TsidMajor)
+ /// True for the indexed metrics hour-end merge.
+ pub fn is_metrics_indexed(&self) -> bool {
+ matches!(self, Self::MetricsIndexed)
}
/// True when the whole hour must be merged as one batch — every file of
@@ -123,7 +123,7 @@ impl MergeMode {
match self {
#[cfg(feature = "enterprise")]
Self::Downsampling(_) => true,
- Self::TsidMajor => true,
+ Self::MetricsIndexed => true,
_ => false,
}
}
@@ -131,7 +131,7 @@ impl MergeMode {
/// Row order the merge writes.
pub fn output_sort_order(&self) -> FileSortOrder {
match self {
- Self::HashSorted | Self::TsidMajor => FileSortOrder::HashTimestampAsc,
+ Self::MetricsHashSorted | Self::MetricsIndexed => FileSortOrder::HashTimestampAsc,
_ => FileSortOrder::TimestampDesc,
}
}
@@ -139,8 +139,8 @@ impl MergeMode {
/// Layout of the file(s) the merge writes.
pub fn file_layout(&self) -> MetricsFileLayout {
match self {
- Self::HashSorted => MetricsFileLayout::HashSorted,
- Self::TsidMajor => MetricsFileLayout::TsidMajor,
+ Self::MetricsHashSorted => MetricsFileLayout::HashSorted,
+ Self::MetricsIndexed => MetricsFileLayout::Indexed,
_ => MetricsFileLayout::Legacy,
}
}
@@ -157,11 +157,11 @@ impl MergeMode {
.filter(|f| MetricsFileLayout::of(&f.key).is_hash_ordered())
.count();
match self {
- Self::HashSorted | Self::TsidMajor if hash_ordered == files.len() => {
+ Self::MetricsHashSorted | Self::MetricsIndexed if hash_ordered == files.len() => {
FileSortOrder::HashTimestampAsc
}
_ if hash_ordered > 0 => FileSortOrder::None,
- Self::HashSorted | Self::TsidMajor => FileSortOrder::None,
+ Self::MetricsHashSorted | Self::MetricsIndexed => FileSortOrder::None,
_ => FileSortOrder::TimestampDesc,
}
}
@@ -188,7 +188,7 @@ impl MergeMode {
Self::Downsampling(rule) => {
super::downsampling::generate_downsampling_sql(schema, rule)
}
- Self::HashSorted | Self::TsidMajor => format!(
+ Self::MetricsHashSorted | Self::MetricsIndexed => format!(
"SELECT * FROM tbl ORDER BY {}",
FileSortOrder::HashTimestampAsc
.order_by_clause()
@@ -206,8 +206,8 @@ impl fmt::Display for MergeMode {
Self::FileList => write!(f, "file_list"),
#[cfg(feature = "enterprise")]
Self::Downsampling(rule) => write!(f, "downsampling(step={}s)", rule.step),
- Self::HashSorted => write!(f, "hash_sorted"),
- Self::TsidMajor => write!(f, "tsid_major"),
+ Self::MetricsHashSorted => write!(f, "metrics_hash_sorted"),
+ Self::MetricsIndexed => write!(f, "metrics_indexed"),
}
}
}
@@ -278,19 +278,19 @@ mod tests {
MergeMode::FileList
));
assert!(!MergeMode::Classic.merges_whole_batch());
- assert!(MergeMode::TsidMajor.merges_whole_batch());
- assert!(!MergeMode::HashSorted.merges_whole_batch());
+ assert!(MergeMode::MetricsIndexed.merges_whole_batch());
+ assert!(!MergeMode::MetricsHashSorted.merges_whole_batch());
assert_eq!(
MergeMode::Classic.input_sort_order(&[]),
FileSortOrder::TimestampDesc
);
assert_eq!(MergeMode::Classic.file_layout(), MetricsFileLayout::Legacy);
assert_eq!(
- MergeMode::HashSorted.file_layout(),
+ MergeMode::MetricsHashSorted.file_layout(),
MetricsFileLayout::HashSorted
);
assert_eq!(
- MergeMode::TsidMajor.output_sort_order(),
+ MergeMode::MetricsIndexed.output_sort_order(),
FileSortOrder::HashTimestampAsc
);
}
@@ -299,11 +299,10 @@ mod tests {
fn input_sort_order_by_file_layout() {
let legacy = FileKey::from_file_name("files/o/metrics/m/2026/08/18/10/1.parquet");
let sorted =
- FileKey::from_file_name("files/o/metrics/m/2026/08/18/10/tsid-sorted-2.parquet");
- let major =
- FileKey::from_file_name("files/o/metrics/m/2026/08/18/10/tsid-major-v3-3.parquet");
+ FileKey::from_file_name("files/o/metrics/m/2026/08/18/10/hash-sorted-v1-2.parquet");
+ let major = FileKey::from_file_name("files/o/metrics/m/2026/08/18/10/indexed-v1-3.parquet");
// all inputs hash ordered: the hash modes merge them pre-sorted
- for mode in [MergeMode::HashSorted, MergeMode::TsidMajor] {
+ for mode in [MergeMode::MetricsHashSorted, MergeMode::MetricsIndexed] {
assert_eq!(
mode.input_sort_order(std::slice::from_ref(&sorted)),
FileSortOrder::HashTimestampAsc
@@ -378,7 +377,7 @@ mod tests {
"SELECT * FROM tbl ORDER BY min_ts DESC"
);
assert_eq!(
- MergeMode::TsidMajor.sql(&without_session),
+ MergeMode::MetricsIndexed.sql(&without_session),
"SELECT * FROM tbl ORDER BY __hash__ ASC, _timestamp ASC"
);
assert!(
diff --git a/src/search/src/datafusion/sort_order.rs b/src/search/src/datafusion/sort_order.rs
index 7b4350739e..317fd05517 100644
--- a/src/search/src/datafusion/sort_order.rs
+++ b/src/search/src/datafusion/sort_order.rs
@@ -16,7 +16,7 @@
//! Physical sort order of the files (and in-memory batches) that back a table.
//!
//! Historically every table was either unsorted or sorted by `_timestamp DESC`
-//! (`sorted_by_time: bool`). TSID-major metrics files are instead sorted by
+//! (`sorted_by_time: bool`). Indexed metrics files are instead sorted by
//! `(__hash__ ASC, _timestamp ASC)`. This enum is the single place that knows
//! how each order maps to DataFusion logical / physical sort expressions, so
//! the table providers, `NewEmptyExec`, the memtable `SortExec` and the merge
@@ -77,9 +77,9 @@ pub enum FileSortOrder {
#[default]
None,
/// `_timestamp DESC` — the classic layout written by the ingester and the
- /// compactor for logs, traces and (non TSID-major) metrics.
+ /// compactor for logs, traces and non-indexed metrics.
TimestampDesc,
- /// `__hash__ ASC, _timestamp ASC` — TSID-major metrics layout: all samples of
+ /// `__hash__ ASC, _timestamp ASC` — indexed metrics layout: all samples of
/// one series are contiguous and time-ordered inside a file. Only used by
/// the compactor merge; distributed query plans never carry it.
HashTimestampAsc,
diff --git a/src/search/src/datafusion/table_provider/helpers.rs b/src/search/src/datafusion/table_provider/helpers.rs
index 3a2bba6758..0e3d4a504e 100644
--- a/src/search/src/datafusion/table_provider/helpers.rs
+++ b/src/search/src/datafusion/table_provider/helpers.rs
@@ -119,7 +119,7 @@ fn generate_parquet_access_plan(
/// Build a Parquet access plan directly from sorted, non-overlapping row
/// ranges. This avoids allocating a bitmap proportional to a high-cardinality
-/// metrics file when a series index already stores compact contiguous runs.
+/// metrics file when a metrics index already stores compact contiguous runs.
fn generate_parquet_access_plan_from_ranges(
file: &PartitionedFile,
ranges: impl IntoIterator- >,