From 08d31371ccc8531538bd2b4292766964339f2c99 Mon Sep 17 00:00:00 2001 From: Huaijin Date: Fri, 21 Aug 2026 04:58:50 +0000 Subject: [PATCH] refactor(metrics): prepare indexed metrics groundwork (#13907) ## Summary - rename the existing TSID-major config, modules, types, merge modes, and persisted file prefixes to indexed-metrics terminology - add the promql-service dependency declarations needed by the follow-up metrics-index work - carry the exact Cargo.lock content from #13899, including h2 0.4.16 and the promql-service dependency entries ## Scope This is no-new-feature groundwork split from #13899. It retains the existing metrics layout mechanics under the new names and does not add .midx sidecar writing, index lookup/query pruning, or index lifecycle handling. The old environment variable and file prefixes are intentionally not kept as compatibility aliases because this experimental path has not been used in a deployment. ## Validation - cargo fmt --all -- --check - cargo metadata --locked --no-deps --format-version 1 - cargo check -p promql-service - cargo test -p config --lib -- layout (5 passed) - cargo test -p search --lib -- datafusion::merge (8 passed) - cargo test -p compaction --lib -- merge (32 passed) - cargo check -p openobserve-jobs --- Cargo.lock | 7 +- src/compaction/src/merge/metrics.rs | 134 ++++++++++++++++++ src/compaction/src/merge/mod.rs | 2 +- src/compaction/src/merge/stream.rs | 22 +-- src/compaction/src/merge/tsid_major.rs | 131 ----------------- src/config/src/config.rs | 6 +- .../meta/promql/{tsid_layout.rs => layout.rs} | 67 +++++---- src/config/src/meta/promql/mod.rs | 2 +- src/jobs/src/job/files/pack.rs | 2 +- src/promql_service/Cargo.toml | 5 + .../src/datafusion/merge/downsampling.rs | 4 +- .../merge/{tsid_major.rs => metrics.rs} | 24 ++-- src/search/src/datafusion/merge/mod.rs | 8 +- src/search/src/datafusion/merge/mode.rs | 65 +++++---- src/search/src/datafusion/sort_order.rs | 6 +- .../src/datafusion/table_provider/helpers.rs | 2 +- 16 files changed, 247 insertions(+), 240 deletions(-) create mode 100644 src/compaction/src/merge/metrics.rs delete mode 100644 src/compaction/src/merge/tsid_major.rs rename src/config/src/meta/promql/{tsid_layout.rs => layout.rs} (73%) rename src/search/src/datafusion/merge/{tsid_major.rs => metrics.rs} (92%) 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>,