diff --git a/Cargo.lock b/Cargo.lock index e1ebdb2b16..aa145766e9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1122,6 +1122,7 @@ dependencies = [ "aws-credential-types", "aws-sigv4", "aws-smithy-async", + "aws-smithy-eventstream", "aws-smithy-http", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1138,6 +1139,33 @@ dependencies = [ "uuid", ] +[[package]] +name = "aws-sdk-lambda" +version = "1.138.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b002aa94953a1aab8534280436ea3cf7805b821d51ebb49a89da41764650ce0b" +dependencies = [ + "arc-swap", + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-eventstream", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-observability", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-schema", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http 0.2.12", + "http 1.5.0", + "regex-lite", + "tracing", +] + [[package]] name = "aws-sdk-sns" version = "1.107.0" @@ -1251,6 +1279,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "723c2234ad7511ceef63eab016b7ba6ff7c55590fefb96fa8467af014a07309f" dependencies = [ "aws-credential-types", + "aws-smithy-eventstream", "aws-smithy-http", "aws-smithy-runtime-api", "aws-smithy-types", @@ -1277,12 +1306,24 @@ dependencies = [ "tokio", ] +[[package]] +name = "aws-smithy-eventstream" +version = "0.61.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a9381123ab62d20c13082b151f30f962a3b112b727345394536dfa39a482944" +dependencies = [ + "aws-smithy-types", + "bytes", + "crc32fast", +] + [[package]] name = "aws-smithy-http" version = "0.64.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37843d9add67c3aff5856f409c6dc315d3cdff60f9c0cb5b670dab1e9920306d" dependencies = [ + "aws-smithy-eventstream", "aws-smithy-runtime-api", "aws-smithy-types", "bytes", @@ -7882,6 +7923,7 @@ dependencies = [ "openobserve-node", "openobserve-org-storage", "openobserve-sourcemaps", + "openobserve-synthetics", "opentelemetry", "pprof", "promql-service", @@ -8121,6 +8163,7 @@ dependencies = [ "openobserve-core", "openobserve-node", "openobserve-org-storage", + "openobserve-synthetics", "prometheus", "promql-service", "proto", @@ -8217,6 +8260,30 @@ dependencies = [ "zip 8.6.0", ] +[[package]] +name = "openobserve-synthetics" +version = "0.1.0" +dependencies = [ + "anyhow", + "aws-config", + "aws-sdk-lambda", + "chrono", + "config", + "dashmap", + "infra", + "log", + "o2_enterprise", + "o2_openfga", + "reqwest 0.13.4", + "sea-orm", + "serde", + "serde_json", + "svix-ksuid", + "tokio", + "tracing", + "urlencoding", +] + [[package]] name = "openobserve-tls" version = "0.1.0" diff --git a/src/compaction/src/merge.rs b/src/compaction/src/merge/mod.rs similarity index 97% rename from src/compaction/src/merge.rs rename to src/compaction/src/merge/mod.rs index 1b7d43a241..71c61504df 100644 --- a/src/compaction/src/merge.rs +++ b/src/compaction/src/merge/mod.rs @@ -22,9 +22,12 @@ use config::{ FileFormat, cluster::LOCAL_NODE, get_config, ider, is_local_disk_storage, - meta::stream::{ - FileKey, FileListDeleted, FileMeta, MergeStrategy, PartitionTimeLevel, StorageType, - StreamType, + meta::{ + promql::tsid_layout::MetricsFileLayout, + stream::{ + FileKey, FileListDeleted, FileMeta, MergeStrategy, PartitionTimeLevel, StorageType, + StreamType, + }, }, metrics, utils::{ @@ -58,6 +61,9 @@ use tokio::{ }; use super::worker::{MergeBatch, MergeSender}; +use crate::merge::tsid_major::{TsidMajorMergeScope, tsid_major_merge_scope}; + +mod tsid_major; /// Last microsecond of the range a merge job at `offset` covers: the hour of /// `offset`, or the whole day for daily-partitioned streams. No file of the @@ -427,7 +433,13 @@ pub async fn merge_by_stream( // the merge only read it — from the end of the range, so a whole-hour // mode applies to the hour as a unit. let range_end_ts = job_range_end(offset, partition_time_level); - let mode = MergeMode::for_compactor(stream_type, stream_name, range_end_ts, !is_incremental); + let mode = MergeMode::for_compactor( + stream_type, + stream_name, + &schema, + range_end_ts, + !is_incremental, + ); // a whole-hour merge needs every file of the hour, even those already // above the size target that a normal merge would leave alone let max_original_size = if mode.merges_whole_batch() { @@ -496,6 +508,27 @@ 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 => { + files_with_size.retain(|f| { + MetricsFileLayout::of(&f.key) != MetricsFileLayout::TsidMajor + }); + log::debug!( + "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] tsid_major late merge of {} files, finalized files untouched", + files_with_size.len() + ); + } + TsidMajorMergeScope::WholeHour => { + log::debug!( + "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] tsid_major fragmentation cap hit, full rewrite of {} files", + files_with_size.len() + ); + } + } + } // group files need to merge let mut batch_groups = Vec::new(); @@ -867,8 +900,13 @@ pub async fn merge_files( target_partitions: 2, }; + let input_sort_order = mode.input_sort_order(&files); + log::debug!( + "[COMPACTOR:WORKER:{thread_id}] merge [{mode}] input sort order: {input_sort_order}, files: {}", + files.len() + ); let tables = match TableBuilder::new() - .sort_order(mode.input_sort_order()) + .sort_order(input_sort_order) .build(session, files.clone(), schema.clone()) .await { @@ -935,6 +973,7 @@ pub async fn merge_files( for MergedFile { buf, meta: mut new_file_meta, + layout, } in merged_files { new_file_meta.compressed_size = buf.len() as i64; @@ -945,7 +984,7 @@ pub async fn merge_files( } let id = ider::generate_file_name(); - let new_file_key = format!("{prefix}/{id}{}", file_format.extension()); + let new_file_key = format!("{prefix}/{}", layout.file_name(&id, file_format)); // upload file to storage let buf = Bytes::from(buf); diff --git a/src/compaction/src/merge/tsid_major.rs b/src/compaction/src/merge/tsid_major.rs new file mode 100644 index 0000000000..80ea302202 --- /dev/null +++ b/src/compaction/src/merge/tsid_major.rs @@ -0,0 +1,131 @@ +// 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 2a5eeb3b91..32ccffdca7 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -2627,6 +2627,12 @@ pub struct Limit { pub struct Compact { #[env_config(name = "ZO_COMPACT_ENABLED", default = true)] pub enabled: bool, + #[env_config( + name = "ZO_METRICS_TSID_MAJOR_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." + )] + pub metrics_tsid_major_enabled: bool, #[env_config(name = "ZO_COMPACT_INTERVAL", default = 10)] // seconds pub interval: u64, #[env_config( diff --git a/src/config/src/meta/promql/mod.rs b/src/config/src/meta/promql/mod.rs index d37f6c5be5..dd851bbe88 100644 --- a/src/config/src/meta/promql/mod.rs +++ b/src/config/src/meta/promql/mod.rs @@ -66,6 +66,7 @@ where } pub mod grpc; +pub mod tsid_layout; pub mod value; pub const NAME_LABEL: &str = "__name__"; diff --git a/src/config/src/meta/promql/tsid_layout.rs b/src/config/src/meta/promql/tsid_layout.rs new file mode 100644 index 0000000000..068f43085a --- /dev/null +++ b/src/config/src/meta/promql/tsid_layout.rs @@ -0,0 +1,193 @@ +// 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 . + +//! 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. + +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 +/// `(__hash__, _timestamp)`, so readers must not assume a `_timestamp` order. +pub fn metrics_tsid_major_enabled(stream_type: StreamType) -> bool { + if stream_type != StreamType::Metrics { + return false; + } + let cfg = crate::get_config(); + cfg.compact.metrics_tsid_major_enabled + && cfg.common.file_format.for_stream(stream_type) == FileFormat::Parquet +} + +/// [`metrics_tsid_major_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) + && schema + .field_with_name(HASH_LABEL) + .is_ok_and(|field| field.data_type() == &DataType::UInt64) +} + +/// Physical layout of a metrics data file, encoded in its file-name prefix so +/// readers and later merges know the row order without opening the file. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum MetricsFileLayout { + /// Classic `_timestamp DESC` file (`{id}.parquet` / `{id}.vortex`). + 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`). + HashSorted, + /// Size-bounded Parquet ordered by `(__hash__ ASC, _timestamp ASC)`, + /// written by the compactor's hour-end merge (`tsid-major-v3-{id}.parquet`). + TsidMajor, +} + +impl MetricsFileLayout { + const HASH_SORTED_PREFIX: &'static str = "tsid-sorted-"; + const TSID_MAJOR_PREFIX: &'static str = "tsid-major-v3-"; + + /// 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), + ] { + if let Some(id) = file_name.strip_prefix(prefix) + && let Some(id) = id.strip_suffix(".parquet") + && !id.is_empty() + { + return layout; + } + } + Self::Legacy + } + + fn prefix(self) -> &'static str { + match self { + Self::Legacy => "", + Self::HashSorted => Self::HASH_SORTED_PREFIX, + Self::TsidMajor => Self::TSID_MAJOR_PREFIX, + } + } + + /// True when the rows are ordered by `(__hash__, _timestamp)`. + pub fn is_hash_ordered(self) -> bool { + !matches!(self, Self::Legacy) + } + + /// File name for a new file of this layout. Marked layouts are always + /// Parquet; `Legacy` keeps the extension of `file_format`. + pub fn file_name(self, id: &str, file_format: FileFormat) -> String { + match self { + Self::Legacy => format!("{id}{}", file_format.extension()), + _ => format!("{}{id}.parquet", self.prefix()), + } + } + + /// Add this layout's marker to the file name of an existing key + /// (`files/.../7099.parquet` -> `files/.../tsid-sorted-7099.parquet`). + pub fn mark_file_key(self, key: &str) -> String { + match key.rfind('/') { + Some(pos) => format!("{}/{}{}", &key[..pos], self.prefix(), &key[pos + 1..]), + None => format!("{}{key}", self.prefix()), + } + } +} + +#[cfg(test)] +mod metrics_file_layout_tests { + use FileFormat; + + use super::*; + + #[test] + fn recognizes_layouts_from_file_names() { + assert_eq!( + MetricsFileLayout::of("files/default/metrics/cpu/2026/08/18/10/7099.parquet"), + MetricsFileLayout::Legacy + ); + assert_eq!( + MetricsFileLayout::of("7099.vortex"), + MetricsFileLayout::Legacy + ); + assert_eq!( + MetricsFileLayout::of( + "files/default/metrics/cpu/2026/08/18/10/tsid-sorted-7099.parquet" + ), + MetricsFileLayout::HashSorted + ); + assert_eq!( + MetricsFileLayout::of( + "files/default/metrics/test/2026/08/13/10/tsid-major-v3-456.parquet" + ), + MetricsFileLayout::TsidMajor + ); + // 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", + ] { + assert_eq!( + MetricsFileLayout::of(name), + MetricsFileLayout::Legacy, + "{name}" + ); + } + assert!(MetricsFileLayout::HashSorted.is_hash_ordered()); + assert!(MetricsFileLayout::TsidMajor.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" + ); + assert_eq!( + MetricsFileLayout::HashSorted.file_name("456", FileFormat::Parquet), + "tsid-sorted-456.parquet" + ); + assert_eq!( + MetricsFileLayout::Legacy.file_name("456", FileFormat::Vortex), + "456.vortex" + ); + let key = "files/default/metrics/cpu/2026/08/18/10/7099.parquet"; + let marked = MetricsFileLayout::HashSorted.mark_file_key(key); + assert_eq!( + marked, + "files/default/metrics/cpu/2026/08/18/10/tsid-sorted-7099.parquet" + ); + assert_eq!( + MetricsFileLayout::of(&marked), + MetricsFileLayout::HashSorted + ); + assert_eq!(MetricsFileLayout::Legacy.mark_file_key(key), key); + assert_eq!( + MetricsFileLayout::HashSorted.mark_file_key("1.parquet"), + "tsid-sorted-1.parquet" + ); + } +} diff --git a/src/jobs/src/job/files/pack.rs b/src/jobs/src/job/files/pack.rs index a6ae0be73f..c076118d6a 100644 --- a/src/jobs/src/job/files/pack.rs +++ b/src/jobs/src/job/files/pack.rs @@ -26,7 +26,10 @@ use bytes::Bytes; use chrono::Duration; use config::{ FileFormat, cluster, get_config, - meta::stream::{FileMeta, StreamType}, + meta::{ + promql::tsid_layout::MetricsFileLayout, + stream::{FileMeta, StreamType}, + }, metrics, utils::{parquet::generate_filename_with_time_range, time::now_micros}, }; @@ -388,7 +391,11 @@ async fn upload_chunk( bufs.push(data); } - let (buf, mut new_file_meta, file_format) = if chunk.len() == 1 { + // Segments are persisted in arrival order. A hash-sorted layout must be + // produced by the merge, so it never takes the as-is fast path. + let mode = MergeMode::for_ingester(stream_type, stream_name, &latest_schema); + let single_segment_as_is = chunk.len() == 1 && mode.file_layout() == MetricsFileLayout::Legacy; + let (buf, mut new_file_meta, file_format, layout) = if single_segment_as_is { // fast path: upload the parquet bytes as-is, no decode/re-encode let buf = Bytes::from(bufs.pop().unwrap()); let file_meta = FileMeta { @@ -399,7 +406,12 @@ async fn upload_chunk( compressed_size: buf.len() as i64, ..Default::default() }; - (buf, file_meta, FileFormat::Parquet) + ( + buf, + file_meta, + FileFormat::Parquet, + MetricsFileLayout::Legacy, + ) } else { // merge multiple segments in memory let mut segment_schemas = Vec::with_capacity(bufs.len()); @@ -437,13 +449,18 @@ async fn upload_chunk( tables, &bloom_filter_fields, new_file_meta, - &MergeMode::for_ingester(stream_type, stream_name), + &mode, MergeOutput::for_ingester(stream_type), ) .await?; // the ingester always merges into exactly one file let (merged, file_format) = merge_result.into_single()?; - (Bytes::from(merged.buf), merged.meta, file_format) + ( + Bytes::from(merged.buf), + merged.meta, + file_format, + merged.layout, + ) }; if new_file_meta.compressed_size == 0 { @@ -458,13 +475,13 @@ async fn upload_chunk( chunk[0].meta.partition_key, generate_filename_with_time_range(min_ts, max_ts, 0) ); - let new_file_key = super::generate_ingester_storage_file_key( + let new_file_key = layout.mark_file_key(&super::generate_ingester_storage_file_key( org_id, stream_type, stream_name, &wal_like_name, file_format, - ); + )); log::info!( "[INGESTER:PACK:JOB:{thread_id}] uploading {} segments into a new file: {new_file_key}, original_size: {}, compressed_size: {}, took: {} ms", diff --git a/src/jobs/src/job/files/parquet.rs b/src/jobs/src/job/files/parquet.rs index 1d20582530..6b2f050726 100644 --- a/src/jobs/src/job/files/parquet.rs +++ b/src/jobs/src/job/files/parquet.rs @@ -784,12 +784,13 @@ async fn merge_files( .await?; let start = std::time::Instant::now(); + let mode = MergeMode::for_ingester(stream_type, &stream_name, &schema); let merge_result = merge::merge_parquet_files( schema, tables, &bloom_filter_fields, new_file_meta, - &MergeMode::for_ingester(stream_type, &stream_name), + &mode, MergeOutput::for_ingester(stream_type), ) .await; @@ -809,20 +810,20 @@ async fn merge_files( // the ingester always merges into exactly one file let (merged, file_format) = buf.into_single()?; - let (buf, mut new_file_meta) = (merged.buf, merged.meta); + let (buf, mut new_file_meta, layout) = (merged.buf, merged.meta, merged.layout); if new_file_meta.compressed_size == 0 { return Err(anyhow::anyhow!( "merge_parquet_files error: compressed_size is 0" )); } - let new_file_key = super::generate_ingester_storage_file_key( + let new_file_key = layout.mark_file_key(&super::generate_ingester_storage_file_key( &org_id, stream_type, &stream_name, &file_name, file_format, - ); + )); log::info!( "[INGESTER:JOB:{thread_id}] merged {} files into a new file: {new_file_key}, original_size: {}, compressed_size: {}, took: {} ms", retain_file_list.len(), diff --git a/src/search/src/datafusion/merge/downsampling.rs b/src/search/src/datafusion/merge/downsampling.rs index 7e797fdc50..5b6ffa3b3d 100644 --- a/src/search/src/datafusion/merge/downsampling.rs +++ b/src/search/src/datafusion/merge/downsampling.rs @@ -19,7 +19,9 @@ use arrow::array::{Int64Array, RecordBatch}; use config::{ FileFormat, TIMESTAMP_COL_NAME, get_config, meta::{ - promql::{DownsamplingRule, Function, HASH_LABEL, VALUE_LABEL}, + promql::{ + DownsamplingRule, Function, HASH_LABEL, VALUE_LABEL, tsid_layout::MetricsFileLayout, + }, stream::FileMeta, }, utils::parquet::new_parquet_writer, @@ -72,7 +74,11 @@ pub(super) async fn write_files( Ok(bufs .into_iter() .zip(file_metas) - .map(|(buf, meta)| MergedFile { buf, meta }) + .map(|(buf, meta)| MergedFile { + buf, + meta, + layout: MetricsFileLayout::Legacy, + }) .collect()) } diff --git a/src/search/src/datafusion/merge/mod.rs b/src/search/src/datafusion/merge/mod.rs index 008f80e0ef..df71726199 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::stream::FileMeta, + meta::{promql::tsid_layout::MetricsFileLayout, stream::FileMeta}, utils::parquet::{VORTEX_FILE_META_KEY, encode_vortex_file_meta, new_parquet_writer}, }; use datafusion::{ @@ -52,6 +52,7 @@ use crate::datafusion::{ #[cfg(feature = "enterprise")] pub mod downsampling; pub mod mode; +mod tsid_major; pub use mode::{MergeMode, MergeOutput}; @@ -59,6 +60,8 @@ pub use mode::{MergeMode, MergeOutput}; pub struct MergedFile { pub buf: Vec, pub meta: FileMeta, + /// Physical layout the file was written in; decides its file name. + pub layout: MetricsFileLayout, } pub struct MergeParquetResult { @@ -98,7 +101,7 @@ pub async fn merge_parquet_files( let sql = mode.sql(&schema); log::debug!("merge_parquet_files [{mode}] sql: {sql}"); let (schema, mut rx, read_task) = - run_merge_query(&sql, mode.input_sort_order(), schema, tables).await?; + run_merge_query(&sql, mode.output_sort_order(), schema, tables).await?; let files = match mode { #[cfg(feature = "enterprise")] @@ -113,6 +116,17 @@ pub async fn merge_parquet_files( ) .await? } + MergeMode::TsidMajor => { + tsid_major::write_files( + &schema, + bloom_filter_fields, + &metadata, + get_config().compact.max_file_size, + &mut rx, + read_task, + ) + .await? + } _ => { let buf = match output.file_format { FileFormat::Parquet => { @@ -132,6 +146,7 @@ pub async fn merge_parquet_files( vec![MergedFile { buf, meta: metadata, + layout: mode.file_layout(), }] } }; @@ -149,9 +164,11 @@ pub async fn merge_parquet_files( /// Plan and start `sql` over the union of `tables`; the record batches arrive /// on the returned channel, the task reports the stream's completion / error. +/// `sort_order` only enables `split_file_groups_by_statistics` for that +/// order; the input tables declare what the files really carry. async fn run_merge_query( sql: &str, - input_sort_order: FileSortOrder, + sort_order: FileSortOrder, schema: Arc, tables: Vec>, ) -> Result<( @@ -162,7 +179,7 @@ async fn run_merge_query( let cfg = get_config(); let ctx = DataFusionContextBuilder::new() .trace_id("merge_parquet_files") - .sort_order(input_sort_order) + .sort_order(sort_order) .build(cfg.limit.datafusion_min_partition_num) .await?; // register union table @@ -378,7 +395,7 @@ mod tests { .unwrap(); let table = Arc::new(MemTable::try_new(schema.clone(), vec![vec![batch]]).unwrap()); - let mode = MergeMode::for_ingester(StreamType::Metadata, "trace_time_index_test"); + let mode = MergeMode::for_ingester(StreamType::Metadata, "trace_time_index_test", &schema); assert!(matches!(mode, MergeMode::TraceTimeIndex)); let merged = merge_parquet_files( schema, diff --git a/src/search/src/datafusion/merge/mode.rs b/src/search/src/datafusion/merge/mode.rs index 8cc56adc0c..12a68d7810 100644 --- a/src/search/src/datafusion/merge/mode.rs +++ b/src/search/src/datafusion/merge/mode.rs @@ -28,7 +28,11 @@ use arrow_schema::Schema; #[cfg(feature = "enterprise")] use config::meta::promql::DownsamplingRule; use config::{ - FileFormat, FileFormatConfig, TIMESTAMP_COL_NAME, get_config, meta::stream::StreamType, + FileFormat, FileFormatConfig, TIMESTAMP_COL_NAME, get_config, + meta::{ + promql::tsid_layout::{MetricsFileLayout, metrics_tsid_major_stream}, + stream::{FileKey, StreamType}, + }, utils::util::is_trace_time_index_stream, }; #[cfg(feature = "enterprise")] @@ -52,6 +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 + /// 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, } impl MergeMode { @@ -63,23 +74,32 @@ impl MergeMode { pub fn for_compactor( stream_type: StreamType, stream_name: &str, - max_ts: i64, + schema: &Schema, + _max_ts: i64, finalize: bool, ) -> Self { #[cfg(feature = "enterprise")] if finalize && stream_type == StreamType::Metrics - && let Some(rule) = get_largest_downsampling_rule(stream_name, max_ts) + && let Some(rule) = get_largest_downsampling_rule(stream_name, _max_ts) { return Self::Downsampling(rule.clone()); } - #[cfg(not(feature = "enterprise"))] - let _ = (max_ts, finalize); + if metrics_tsid_major_stream(stream_type, schema) { + return if finalize { + Self::TsidMajor + } else { + Self::HashSorted + }; + } Self::for_stream(stream_type, stream_name) } - /// Mode for the ingester movers: never downsamples. - pub fn for_ingester(stream_type: StreamType, stream_name: &str) -> Self { + /// 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; + } Self::for_stream(stream_type, stream_name) } @@ -91,6 +111,11 @@ impl MergeMode { } } + /// True for the TSID-major hour-end merge. + pub fn is_tsid_major(&self) -> bool { + matches!(self, Self::TsidMajor) + } + /// True when the whole hour must be merged as one batch — every file of /// the hour, including ones already above the size target, and regardless /// of `ZO_COMPACT_MAX_FILE_SIZE` (the writer splits the output itself). @@ -98,13 +123,47 @@ impl MergeMode { match self { #[cfg(feature = "enterprise")] Self::Downsampling(_) => true, + Self::TsidMajor => true, _ => false, } } - /// Physical order of the input files the merge query may rely on. - pub fn input_sort_order(&self) -> FileSortOrder { - FileSortOrder::TimestampDesc + /// Row order the merge writes. + pub fn output_sort_order(&self) -> FileSortOrder { + match self { + Self::HashSorted | Self::TsidMajor => FileSortOrder::HashTimestampAsc, + _ => FileSortOrder::TimestampDesc, + } + } + + /// Layout of the file(s) the merge writes. + pub fn file_layout(&self) -> MetricsFileLayout { + match self { + Self::HashSorted => MetricsFileLayout::HashSorted, + Self::TsidMajor => MetricsFileLayout::TsidMajor, + _ => MetricsFileLayout::Legacy, + } + } + + /// Physical order of the input `files` the merge query may rely on. + /// + /// A hash-ordered merge whose inputs are all hash-ordered declares that + /// order, so DataFusion merges the pre-sorted files instead of a full + /// sort. Any mix with other layouts declares no order; otherwise the + /// classic `_timestamp DESC`. + pub fn input_sort_order(&self, files: &[FileKey]) -> FileSortOrder { + let hash_ordered = files + .iter() + .filter(|f| MetricsFileLayout::of(&f.key).is_hash_ordered()) + .count(); + match self { + Self::HashSorted | Self::TsidMajor if hash_ordered == files.len() => { + FileSortOrder::HashTimestampAsc + } + _ if hash_ordered > 0 => FileSortOrder::None, + Self::HashSorted | Self::TsidMajor => FileSortOrder::None, + _ => FileSortOrder::TimestampDesc, + } } /// The merge query over the union table `tbl`. @@ -129,6 +188,12 @@ impl MergeMode { Self::Downsampling(rule) => { super::downsampling::generate_downsampling_sql(schema, rule) } + Self::HashSorted | Self::TsidMajor => format!( + "SELECT * FROM tbl ORDER BY {}", + FileSortOrder::HashTimestampAsc + .order_by_clause() + .expect("hash order has sort columns") + ), } } } @@ -141,6 +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"), } } } @@ -197,21 +264,67 @@ mod tests { #[test] fn mode_by_stream() { + let schema = Schema::empty(); assert!(matches!( - MergeMode::for_ingester(StreamType::Logs, "app"), + MergeMode::for_ingester(StreamType::Logs, "app", &schema), MergeMode::Classic )); assert!(matches!( - MergeMode::for_compactor(StreamType::Traces, "app", 0, true), + MergeMode::for_compactor(StreamType::Traces, "app", &schema, 0, true), MergeMode::Classic )); assert!(matches!( - MergeMode::for_compactor(StreamType::Filelist, "x", 0, false), + MergeMode::for_compactor(StreamType::Filelist, "x", &schema, 0, false), MergeMode::FileList )); assert!(!MergeMode::Classic.merges_whole_batch()); + assert!(MergeMode::TsidMajor.merges_whole_batch()); + assert!(!MergeMode::HashSorted.merges_whole_batch()); assert_eq!( - MergeMode::Classic.input_sort_order(), + MergeMode::Classic.input_sort_order(&[]), + FileSortOrder::TimestampDesc + ); + assert_eq!(MergeMode::Classic.file_layout(), MetricsFileLayout::Legacy); + assert_eq!( + MergeMode::HashSorted.file_layout(), + MetricsFileLayout::HashSorted + ); + assert_eq!( + MergeMode::TsidMajor.output_sort_order(), + FileSortOrder::HashTimestampAsc + ); + } + + #[test] + 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"); + // all inputs hash ordered: the hash modes merge them pre-sorted + for mode in [MergeMode::HashSorted, MergeMode::TsidMajor] { + assert_eq!( + mode.input_sort_order(std::slice::from_ref(&sorted)), + FileSortOrder::HashTimestampAsc + ); + assert_eq!( + mode.input_sort_order(&[sorted.clone(), major.clone()]), + FileSortOrder::HashTimestampAsc + ); + // a legacy file in the batch: nothing can be assumed + assert_eq!( + mode.input_sort_order(&[sorted.clone(), legacy.clone()]), + FileSortOrder::None + ); + } + // classic merge over hash-ordered files (layout switched off): no order + assert_eq!( + MergeMode::Classic.input_sort_order(&[legacy.clone(), sorted.clone()]), + FileSortOrder::None + ); + assert_eq!( + MergeMode::Classic.input_sort_order(std::slice::from_ref(&legacy)), FileSortOrder::TimestampDesc ); } @@ -264,6 +377,10 @@ mod tests { MergeMode::FileList.sql(&without_session), "SELECT * FROM tbl ORDER BY min_ts DESC" ); + assert_eq!( + MergeMode::TsidMajor.sql(&without_session), + "SELECT * FROM tbl ORDER BY __hash__ ASC, _timestamp ASC" + ); assert!( MergeMode::TraceTimeIndex .sql(&with_session) diff --git a/src/search/src/datafusion/merge/tsid_major.rs b/src/search/src/datafusion/merge/tsid_major.rs new file mode 100644 index 0000000000..5f213a5dcd --- /dev/null +++ b/src/search/src/datafusion/merge/tsid_major.rs @@ -0,0 +1,293 @@ +// 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 . + +use std::sync::Arc; + +use arrow::{ + array::{Array, Int64Array, RecordBatch}, + compute::{max, min}, +}; +use config::{ + TIMESTAMP_COL_NAME, + meta::{promql::tsid_layout::MetricsFileLayout, stream::FileMeta}, + utils::parquet::new_parquet_writer, +}; +use datafusion::{ + arrow::datatypes::Schema, + error::{DataFusionError, Result}, +}; +use parquet::arrow::AsyncArrowWriter; + +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. +pub(super) async fn write_files( + schema: &Arc, + bloom_filter_fields: &[String], + metadata: &FileMeta, + max_file_size: usize, + rx: &mut tokio::sync::mpsc::Receiver, + read_task: tokio::task::JoinHandle>, +) -> Result> { + let timestamp_index = schema.index_of(TIMESTAMP_COL_NAME).map_err(|e| { + DataFusionError::Plan(format!( + "TSID-major 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 files: Vec = Vec::new(); + + while let Some(batch) = rx.recv().await { + if batch.num_rows() == 0 { + continue; + } + // rotate between input batches once the logical size target is reached + if let Some(full) = active.take_if(|writer| { + proportional_original_size(metadata, writer.file_meta.records) >= max_file_size + }) { + 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) + }); + writer.write(&batch).await?; + } + + read_task + .await + .map_err(|e| DataFusionError::External(Box::new(e)))??; + if let Some(active) = active.take() { + files.push(active.finish(metadata, max_file_size).await?); + } + if files.is_empty() { + return Err(DataFusionError::Execution( + "TSID-major merge produced no rows".to_string(), + )); + } + + Ok(files) +} + +/// One output file in progress: the Parquet writer and the running file meta. +struct ActiveSizeTsidWriter { + writer: AsyncArrowWriter>, + file_meta: FileMeta, + timestamp_index: usize, +} + +impl ActiveSizeTsidWriter { + fn new( + schema: &Arc, + bloom_filter_fields: &[String], + metadata: &FileMeta, + timestamp_index: usize, + ) -> Self { + let writer = new_parquet_writer( + Vec::new(), + schema, + bloom_filter_fields, + metadata, + false, + None, + ); + Self { + writer, + file_meta: FileMeta::default(), + timestamp_index, + } + } + + /// Append one (hash, ts)-ordered batch to the data file and its rows / + /// time range to the file meta. + async fn write(&mut self, batch: &RecordBatch) -> Result<()> { + self.writer.write(batch).await?; + + let timestamps = batch + .column(self.timestamp_index) + .as_any() + .downcast_ref::() + .ok_or_else(|| { + DataFusionError::Plan(format!( + "TSID-major layout requires Int64 {TIMESTAMP_COL_NAME}" + )) + })?; + let meta = &mut self.file_meta; + if let (Some(batch_min), Some(batch_max)) = (min(timestamps), max(timestamps)) { + if meta.records == 0 { + (meta.min_ts, meta.max_ts) = (batch_min, batch_max); + } else { + meta.min_ts = meta.min_ts.min(batch_min); + meta.max_ts = meta.max_ts.max(batch_max); + } + } + meta.records += batch.num_rows() as i64; + Ok(()) + } + + async fn finish(self, source_meta: &FileMeta, max_file_size: i64) -> Result { + let Self { + mut writer, + mut file_meta, + .. + } = self; + + // below the target so a finalized file never advertises >= max_file_size + file_meta.original_size = + proportional_original_size(source_meta, file_meta.records).min(max_file_size - 1); + append_metadata(&mut writer, &file_meta)?; + writer.finish().await?; + + let buf = writer.into_inner(); + file_meta.compressed_size = buf.len() as i64; + Ok(MergedFile { + buf, + meta: file_meta, + layout: MetricsFileLayout::TsidMajor, + }) + } +} + +/// The share of the source `original_size` that `records` rows carry. +fn proportional_original_size(source_meta: &FileMeta, records: i64) -> i64 { + let estimate = (i128::from(source_meta.original_size.max(0)) * i128::from(records.max(0))) + / i128::from(source_meta.records.max(1)); + i64::try_from(estimate).unwrap_or(i64::MAX) +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use arrow::array::{Float64Array, Int64Array, StringViewArray, UInt64Array}; + use arrow_schema::{DataType, Field, Schema}; + use config::meta::promql::{HASH_LABEL, VALUE_LABEL}; + use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; + + use super::*; + + #[tokio::test] + async fn test_size_split_tsid_major_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), + Field::new(VALUE_LABEL, DataType::Float64, false), + Field::new("path", DataType::Utf8View, true), + ])); + let batch1 = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(UInt64Array::from(vec![1, 1])), + Arc::new(Int64Array::from(vec![10, 20])), + Arc::new(Float64Array::from(vec![1.0, 2.0])), + Arc::new(StringViewArray::from(vec!["a", "a"])), + ], + ) + .unwrap(); + let batch2 = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(UInt64Array::from(vec![1, 2, 2])), + Arc::new(Int64Array::from(vec![30, 10, 20])), + Arc::new(Float64Array::from(vec![3.0, 4.0, 5.0])), + Arc::new(StringViewArray::from(vec!["a", "b", "b"])), + ], + ) + .unwrap(); + let batch3 = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(UInt64Array::from(vec![3])), + Arc::new(Int64Array::from(vec![10])), + Arc::new(Float64Array::from(vec![6.0])), + Arc::new(StringViewArray::from(vec!["c"])), + ], + ) + .unwrap(); + let metadata = FileMeta { + min_ts: 10, + max_ts: 30, + records: 6, + original_size: 600, + compressed_size: 300, + ..Default::default() + }; + let (tx, mut rx) = tokio::sync::mpsc::channel(3); + tx.send(batch1).await.unwrap(); + tx.send(batch2).await.unwrap(); + tx.send(batch3).await.unwrap(); + drop(tx); + + let max_file_size = 151; + let files = write_files( + &schema, + &[], + &metadata, + max_file_size, + &mut rx, + tokio::spawn(async { Ok(()) }), + ) + .await + .unwrap(); + + assert_eq!(files.len(), 3); + assert_eq!(files.iter().map(|f| f.meta.records).sum::(), 6); + assert_eq!( + files + .iter() + .map(|f| f.meta.original_size) + .collect::>(), + vec![150, 150, 100] + ); + assert!(files.iter().all(|f| { + f.layout == MetricsFileLayout::TsidMajor + && f.meta.original_size < max_file_size as i64 + && f.meta.compressed_size == f.buf.len() as i64 + })); + + let mut file_hashes = Vec::new(); + for file in files { + let bytes = bytes::Bytes::from(file.buf); + let footer = config::utils::parquet::read_metadata_from_bytes(&bytes) + .await + .unwrap(); + assert_eq!(footer.min_ts, file.meta.min_ts); + assert_eq!(footer.max_ts, file.meta.max_ts); + assert_eq!(footer.records, file.meta.records); + assert_eq!(footer.original_size, file.meta.original_size); + + let reader = ParquetRecordBatchReaderBuilder::try_new(bytes) + .unwrap() + .build() + .unwrap(); + let mut hashes_in_file = Vec::new(); + for batch in reader { + let batch = batch.unwrap(); + let hashes = batch + .column_by_name(HASH_LABEL) + .unwrap() + .as_any() + .downcast_ref::() + .unwrap(); + hashes_in_file.extend_from_slice(hashes.values()); + } + file_hashes.push(hashes_in_file); + } + assert_eq!(file_hashes, vec![vec![1, 1], vec![1, 2, 2], vec![3]]); + assert!(file_hashes[0].contains(&1) && file_hashes[1].contains(&1)); + } +} diff --git a/src/search/src/sql/mod.rs b/src/search/src/sql/mod.rs index be1b40a7b5..6706142255 100644 --- a/src/search/src/sql/mod.rs +++ b/src/search/src/sql/mod.rs @@ -182,7 +182,10 @@ impl Sql { } let need_sort_by_time = order_by.len() == 1 && order_by[0].0 == TIMESTAMP_COL_NAME - && order_by[0].1 == OrderBy::Desc; + && order_by[0].1 == OrderBy::Desc + && !total_schemas.iter().any(|(stream, _)| { + matches!(stream.get_stream_type(stream_type), StreamType::Metrics) + }); // check if need exact limit and offset if (limit == -1 || limit == 0)