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
This commit is contained in:
Huaijin
2026-08-21 04:58:50 +00:00
committed by GitHub
parent 8ef1fd4cbb
commit 08d31371cc
16 changed files with 247 additions and 240 deletions
Generated
+5 -2
View File
@@ -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",
+134
View File
@@ -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 <http://www.gnu.org/licenses/>.
//! 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<FileKey> = (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<FileKey> = (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<FileKey> = (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
);
}
}
+1 -1
View File
@@ -15,8 +15,8 @@
mod file;
mod job;
mod metrics;
mod stream;
mod tsid_major;
pub use file::merge_files;
pub use job::{
+11 -11
View File
@@ -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()
);
}
-131
View File
@@ -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 <http://www.gnu.org/licenses/>.
//! 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<FileKey> = (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<FileKey> = (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<FileKey> = (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
);
}
}
+3 -3
View File
@@ -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(
@@ -13,31 +13,30 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
//! 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"
);
}
}
+1 -1
View File
@@ -66,7 +66,7 @@ where
}
pub mod grpc;
pub mod tsid_layout;
pub mod layout;
pub mod value;
pub const NAME_LABEL: &str = "__name__";
+1 -1
View File
@@ -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,
+5
View File
@@ -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
@@ -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,
@@ -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<Schema>,
bloom_filter_fields: &[String],
@@ -45,11 +45,11 @@ pub(super) async fn write_files(
) -> Result<Vec<MergedFile>> {
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<ActiveSizeTsidWriter> = None;
let mut active: Option<ActiveIndexedMetricsWriter> = None;
let mut files: Vec<MergedFile> = 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<Vec<u8>>,
file_meta: FileMeta,
timestamp_index: usize,
}
impl ActiveSizeTsidWriter {
impl ActiveIndexedMetricsWriter {
fn new(
schema: &Arc<Schema>,
bloom_filter_fields: &[String],
@@ -123,7 +123,7 @@ impl ActiveSizeTsidWriter {
.downcast_ref::<Int64Array>()
.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
}));
+4 -4
View File
@@ -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,
+32 -33
View File
@@ -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!(
+3 -3
View File
@@ -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,
@@ -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<Item = Range<usize>>,