From 1bace57a748867b1451849ba35db7bf8d2aaa3cc Mon Sep 17 00:00:00 2001 From: Huaijin Date: Fri, 3 Jul 2026 17:23:42 +0800 Subject: [PATCH] perf: move vortex to compactor (#13016) for metrics data, we only generate vortex file in the compactor, but others we generate parquet file on ingester and compactor. becuase metrics have many small files --- src/config/src/config.rs | 29 +++- src/infra/src/cache/file_data/mod.rs | 24 +++- src/job/files/mod.rs | 44 +++++- src/job/files/parquet.rs | 129 ++++++++---------- src/job/mod.rs | 2 +- src/service/compact/deleted.rs | 6 +- src/service/compact/flatten.rs | 35 ++++- src/service/compact/merge.rs | 18 ++- .../search/datafusion/merge/downsampling.rs | 9 +- src/service/search/datafusion/merge/mod.rs | 51 ++++++- 10 files changed, 243 insertions(+), 104 deletions(-) diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 1b952d621e..2fbc94d958 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -36,7 +36,10 @@ use serde::{Deserialize, Serialize}; use sha256::digest; use crate::{ - meta::{cluster, stream::QueryPartitionStrategy}, + meta::{ + cluster, + stream::{QueryPartitionStrategy, StreamType}, + }, utils::sysinfo, }; @@ -562,6 +565,14 @@ impl std::str::FromStr for FileFormat { } impl FileFormat { + pub fn for_ingester_stream(stream_type: StreamType, configured: Self) -> Self { + if stream_type == StreamType::Metrics { + Self::Parquet + } else { + configured + } + } + pub fn extension(&self) -> &'static str { match self { Self::Parquet => FILE_EXT_PARQUET, @@ -3780,6 +3791,22 @@ mod tests { assert_eq!(FileFormat::Vortex.extension(), ".vortex"); } + #[test] + fn test_file_format_for_ingester_stream() { + assert_eq!( + FileFormat::for_ingester_stream(StreamType::Metrics, FileFormat::Vortex), + FileFormat::Parquet + ); + assert_eq!( + FileFormat::for_ingester_stream(StreamType::Logs, FileFormat::Vortex), + FileFormat::Vortex + ); + assert_eq!( + FileFormat::for_ingester_stream(StreamType::Traces, FileFormat::Parquet), + FileFormat::Parquet + ); + } + #[test] fn test_file_format_from_extension() { assert_eq!( diff --git a/src/infra/src/cache/file_data/mod.rs b/src/infra/src/cache/file_data/mod.rs index d1e1bb09cf..32be605448 100644 --- a/src/infra/src/cache/file_data/mod.rs +++ b/src/infra/src/cache/file_data/mod.rs @@ -52,6 +52,7 @@ enum CacheStrategy { enum FileType { Parquet, Ttv, + Vortex, } impl CacheStrategy { @@ -220,6 +221,15 @@ async fn validate_file(bytes: &[u8], ftype: FileType) -> Result<(), anyhow::Erro return Err(anyhow::anyhow!("payload size mismatch")); } } + FileType::Vortex => { + if bytes.len() < 12 { + return Err(anyhow::anyhow!("invalid vortex file")); + } + const VORTEX_MAGIC: &[u8; 4] = b"VTXF"; + if &bytes[..4] != VORTEX_MAGIC || &bytes[bytes.len() - 4..] != VORTEX_MAGIC { + return Err(anyhow::anyhow!("vortex magic bytes mismatch")); + } + } } Ok(()) } @@ -289,16 +299,20 @@ async fn download_from_storage( // so we check if the footer is valid. If it is, then the db entry is invalid // and we reset it. If footer is invalid, the store has a corrupted file // so we mark it as deleted, and return error. + // data files (parquet/vortex) are tracked in file_list, ttv index files are not + let is_data_file = file.ends_with(".parquet") || file.ends_with(".vortex"); let valid_parquet = file.ends_with(".parquet") && validate_file(&data_bytes, FileType::Parquet).await.is_ok(); + let valid_vortex = file.ends_with(".vortex") + && validate_file(&data_bytes, FileType::Vortex).await.is_ok(); let valid_ttv = file.ends_with(".ttv") && validate_file(&data_bytes, FileType::Ttv).await.is_ok(); - if valid_parquet || valid_ttv { + if valid_parquet || valid_vortex || valid_ttv { log::warn!( "download file {file} found size mismatch, remote : {expected_blob_size}, db: {size}, correcting db as valid file", ); - // only update for parquet files, not ttv files - if file.ends_with(".parquet") { + // only update for data files, not ttv files + if is_data_file { crate::file_list::update_compressed_size(file, data_len as i64).await?; crate::file_list::LOCAL_CACHE .update_compressed_size(file, data_len as i64) @@ -309,8 +323,8 @@ async fn download_from_storage( log::warn!( "download file {file} found corrupt file, remote: {expected_blob_size}, db: {size}, deleting entry from file_list " ); - // only update for parquet files, not ttv files - if file.ends_with(".parquet") { + // only update for data files, not ttv files + if is_data_file { crate::file_list::remove(file).await?; crate::file_list::LOCAL_CACHE.remove(file).await?; } diff --git a/src/job/files/mod.rs b/src/job/files/mod.rs index c9b3183604..1de7e2547c 100644 --- a/src/job/files/mod.rs +++ b/src/job/files/mod.rs @@ -14,6 +14,7 @@ // along with this program. If not, see . use config::{ + FileFormat, cluster::{LOCAL_NODE, is_offline}, ider, meta::stream::StreamType, @@ -54,11 +55,12 @@ async fn clean_empty_dirs() -> Result<(), anyhow::Error> { Ok(()) } -pub fn generate_storage_file_name( +pub fn generate_ingester_storage_file_key( org_id: &str, stream_type: StreamType, stream_name: &str, wal_file_name: &str, + file_format: FileFormat, ) -> String { // eg: 0/2023/08/21/08/8b8a5451bbe1c44b/ip=1234/7099303408192061440f3XQ2p.json let file_columns = wal_file_name.splitn(7, '/').collect::>(); @@ -76,6 +78,42 @@ pub fn generate_storage_file_name( } else { format!("{}/{}", &file_name[..file_name_pos], id) }; - let file_format = config::get_config().common.file_format.extension(); - format!("files/{stream_key}/{file_date}/{file_name}{file_format}") + format!( + "files/{stream_key}/{file_date}/{file_name}{}", + file_format.extension() + ) +} + +#[cfg(test)] +mod tests { + use config::{FileFormat, meta::stream::StreamType}; + + use super::generate_ingester_storage_file_key; + + #[test] + fn test_ingester_storage_file_key_uses_given_format_extension() { + let file_name = generate_ingester_storage_file_key( + "default", + StreamType::Metrics, + "quickstart", + "0/2026/07/02/12/hash/service_name=ingester/source.json", + FileFormat::Parquet, + ); + + assert!( + file_name.starts_with( + "files/default/metrics/quickstart/2026/07/02/12/service_name=ingester/" + ) + ); + assert!(file_name.ends_with(".parquet")); + + let file_name = generate_ingester_storage_file_key( + "default", + StreamType::Logs, + "quickstart", + "0/2026/07/02/12/hash/service_name=ingester/source.json", + FileFormat::Vortex, + ); + assert!(file_name.ends_with(".vortex")); + } } diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs index c5377e9719..dcd3c090a0 100644 --- a/src/job/files/parquet.rs +++ b/src/job/files/parquet.rs @@ -22,8 +22,6 @@ use std::{ use arrow_schema::Schema; use bytes::Bytes; use chrono::{Duration, Utc}; -#[cfg(feature = "enterprise")] -use config::utils::parquet::get_recordbatch_reader_from_bytes; use config::{ FxIndexMap, cluster, get_config, meta::{ @@ -51,6 +49,19 @@ use tokio::{ fs::remove_file, sync::{Mutex, RwLock}, }; +#[cfg(feature = "enterprise")] +use { + config::{FileFormat, cluster::LOCAL_NODE, utils::parquet::get_recordbatch_reader_from_bytes}, + o2_enterprise::enterprise::{ + common::config::get_config as get_enterprise_config, + service_streams::{ + batch_processor::queue_services, cache::get_coverage_deficit, meta::ServiceMetadata, + processor::StreamProcessor, sampler::should_process_file, + }, + }, + std::collections::HashMap, + tokio::sync::mpsc, +}; use crate::{ common::infra::wal, @@ -800,8 +811,12 @@ async fn merge_files( } }; - let (buf, mut new_file_meta) = match buf { - MergeParquetResult::Single(buf, meta) => (buf, meta), + let (buf, mut new_file_meta, file_format) = match buf { + MergeParquetResult::Single { + buf, + file_meta, + file_format, + } => (buf, file_meta, file_format), MergeParquetResult::Multiple { .. } => { // ingester should not support multiple files, it will be handled in compactor mode panic!("[INGESTER:JOB] merge_parquet_files error: multiple files"); @@ -813,8 +828,13 @@ async fn merge_files( "merge_parquet_files error: compressed_size is 0" )); } - let new_file_key = - super::generate_storage_file_name(&org_id, stream_type, &stream_name, &file_name); + let new_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(), @@ -840,9 +860,7 @@ async fn merge_files( // This runs BEFORE indexing checks to ensure all stream types are discovered #[cfg(feature = "enterprise")] { - use config::cluster::LOCAL_NODE; - let service_streams_config = - &o2_enterprise::enterprise::common::config::get_config().service_streams; + let service_streams_config = &get_enterprise_config().service_streams; let valid_stream_type = stream_type == StreamType::Logs || stream_type == StreamType::Metrics @@ -863,25 +881,20 @@ async fn merge_files( // active), tapering back to normal as coverage fills in. // Passing (0, 0) when disabled is the documented no-op sentinel for the sampler. let coverage_deficit = if service_streams_config.coverage_catchup_enabled { - o2_enterprise::enterprise::service_streams::cache::get_coverage_deficit( - &org_id, - stream_type, - ) - .await + get_coverage_deficit(&org_id, stream_type).await } else { (0, 0) }; // Check if we should process this file (adaptive per-type sampling with fast lane) - let should_process = - o2_enterprise::enterprise::service_streams::sampler::should_process_file( - &org_id, - stream_type, - &stream_name, - &new_file_key, - stream_count, - coverage_deficit, - ); + let should_process = should_process_file( + &org_id, + stream_type, + &stream_name, + &new_file_key, + stream_count, + coverage_deficit, + ); if should_process { let buf_clone = buf.clone(); @@ -890,20 +903,17 @@ async fn merge_files( // Queue services for batched processing (non-blocking) tokio::spawn(async move { - if let Err(e) = queue_services_from_parquet( + if let Err(e) = queue_services_from_data_file( &org_id_clone, stream_type, &stream_name_clone, - &buf_clone, + file_format, + buf_clone, ) .await { log::error!( - "[ServiceStreams] Failed to queue services for {}/{}/{}: {}", - org_id_clone, - stream_type, - stream_name_clone, - e + "[ServiceStreams] Failed to queue services for {org_id_clone}/{stream_type}/{stream_name_clone}: {e}" ); } }); @@ -961,16 +971,13 @@ fn split_perfix(prefix: &str) -> (String, StreamType, String, String) { } #[cfg(feature = "enterprise")] -async fn queue_services_from_parquet( +async fn queue_services_from_data_file( org_id: &str, stream_type: StreamType, stream_name: &str, - parquet_data: &[u8], + file_format: FileFormat, + file_data: Bytes, ) -> Result<(), anyhow::Error> { - use std::collections::HashMap; - - use tokio::sync::mpsc; - let start = std::time::Instant::now(); // Get semantic field groups upfront (before spawning tasks) @@ -982,27 +989,21 @@ async fn queue_services_from_parquet( // Create bounded channel for backpressure - drops records if consumer can't keep up // ARROW-NATIVE: Channel now sends RecordBatch directly (no HashMap conversion!) let (tx, mut rx) = mpsc::channel::( - o2_enterprise::enterprise::common::config::get_config() - .service_streams - .channel_capacity, + get_enterprise_config().service_streams.channel_capacity, ); - // Clone data needed for producer task - let parquet_bytes = Bytes::copy_from_slice(parquet_data); - - // Spawn producer task to read parquet and send Arrow batches through channel + // Spawn producer task to read the data file and send Arrow batches through channel. // ARROW-NATIVE: No HashMap conversion! Sends RecordBatch directly. let producer_handle = tokio::spawn(async move { let mut records_sent = 0u64; let mut records_dropped = 0u64; let mut batches_sent = 0u64; - let file_format = config::get_config().common.file_format; - let reader_result = get_recordbatch_reader_from_bytes(file_format, parquet_bytes).await; + let reader_result = get_recordbatch_reader_from_bytes(file_format, file_data).await; let (_schema, mut reader) = match reader_result { Ok(r) => r, Err(e) => { - log::error!("[ServiceStreams] Failed to read parquet: {e}"); + log::error!("[ServiceStreams] Failed to read {file_format} file: {e}"); return (records_sent, records_dropped); } }; @@ -1051,16 +1052,9 @@ async fn queue_services_from_parquet( }); // Consumer: Process batches as they arrive - let processor = o2_enterprise::enterprise::service_streams::processor::StreamProcessor::new( - org_id.to_string(), - semantic_groups, - identity_config, - ); + let processor = StreamProcessor::new(org_id.to_string(), semantic_groups, identity_config); - let mut all_services: HashMap< - String, - o2_enterprise::enterprise::service_streams::meta::ServiceMetadata, - > = HashMap::new(); + let mut all_services: HashMap = HashMap::new(); let mut total_records_processed = 0u64; // Process Arrow batches from channel @@ -1089,10 +1083,7 @@ async fn queue_services_from_parquet( if records_dropped > 0 { log::warn!( - "[ServiceStreams] Dropped {} records due to backpressure for {}/{}", - records_dropped, - org_id, - stream_name + "[ServiceStreams] Dropped {records_dropped} records due to backpressure for {org_id}/{stream_name}", ); metrics::SERVICE_STREAMS_RECORDS_DROPPED .with_label_values(&[org_id, &stream_type.to_string()]) @@ -1115,22 +1106,16 @@ async fn queue_services_from_parquet( .observe(duration.as_secs_f64()); log::debug!( - "[ServiceStreams] Processed {} records, discovered {} services in {:?} for {}/{} (sent: {}, dropped: {})", - total_records_processed, - service_count, - duration, - org_id, - stream_name, - records_sent, - records_dropped + "[ServiceStreams] Processed {total_records_processed} records, discovered {service_count} services in {duration:?} for {org_id}/{stream_name} (sent: {records_sent}, dropped: {records_dropped})", ); // Queue services for batched processing - o2_enterprise::enterprise::service_streams::batch_processor::queue_services( - org_id.to_string(), - all_services, - ) - .await; + queue_services(org_id.to_string(), all_services).await; + + let duration = start.elapsed().as_millis(); + log::info!( + "[ServiceStreams] queue_services_from_data_file completed in {duration} ms for {org_id}/{stream_name} (processed: {total_records_processed}, services: {service_count}, sent: {records_sent}, dropped: {records_dropped})", + ); Ok(()) } diff --git a/src/job/mod.rs b/src/job/mod.rs index 07070ddc2a..cda2be9b4b 100644 --- a/src/job/mod.rs +++ b/src/job/mod.rs @@ -751,7 +751,7 @@ pub async fn init() -> Result<(), anyhow::Error> { tokio::task::spawn(alert_grouping::process_expired_batches()); tokio::task::spawn(file_downloader::run()); // Note: Service discovery extraction runs automatically during parquet file processing - // See src/job/files/parquet.rs:queue_services_from_parquet for implementation + // See src/job/files/parquet.rs:queue_services_from_data_file for implementation #[cfg(feature = "enterprise")] spawn_pausable_job!( "service_streams_batch_processor", diff --git a/src/service/compact/deleted.rs b/src/service/compact/deleted.rs index 69457aa3d8..92e0469e0f 100644 --- a/src/service/compact/deleted.rs +++ b/src/service/compact/deleted.rs @@ -86,11 +86,7 @@ pub async fn delete(org_id: &str, time_max: i64) -> Result { if file.flattened { Some(( file.account.to_string(), - format!( - "files{}/{}", - config::get_config().common.column_all, - file.file.strip_prefix("files/").unwrap() - ), + super::flatten::generate_flatten_file_key(&file.file), )) } else { None diff --git a/src/service/compact/flatten.rs b/src/service/compact/flatten.rs index 26975e661b..48b750849c 100644 --- a/src/service/compact/flatten.rs +++ b/src/service/compact/flatten.rs @@ -149,6 +149,21 @@ pub async fn generate_by_stream( Ok(()) } +/// Generates the storage key of the flattened version of a data file. +pub fn generate_flatten_file_key(file_key: &str) -> String { + let key = file_key.strip_prefix("files/").unwrap_or(file_key); + let key = match FileFormat::from_extension(key) { + Some(format) => key.strip_suffix(format.extension()).unwrap_or(key), + None => key, + }; + format!( + "files{}/{}{}", + get_config().common.column_all, + key, + FileFormat::Parquet.extension() + ) +} + pub async fn generate_file(file: &FileKey) -> Result<(), anyhow::Error> { let start = std::time::Instant::now(); log::debug!("[FLATTEN_COMPACTOR] generate flatten file for {}", file.key); @@ -172,11 +187,7 @@ pub async fn generate_file(file: &FileKey) -> Result<(), anyhow::Error> { file.key )); } - let new_file = format!( - "files{}/{}", - get_config().common.column_all, - file.key.strip_prefix("files/").unwrap() - ); + let new_file = generate_flatten_file_key(&file.key); let org_id = columns[1]; let stream_type = StreamType::from(columns[2]); let stream_name = columns[3]; @@ -304,6 +315,20 @@ mod tests { use super::*; + #[test] + fn test_generate_flatten_file_key_always_uses_parquet_extension() { + let column_all = &get_config().common.column_all; + assert_eq!( + generate_flatten_file_key("files/default/logs/quickstart/2026/07/02/12/abc.parquet"), + format!("files{column_all}/default/logs/quickstart/2026/07/02/12/abc.parquet") + ); + // flattened output is written as parquet even when the source is vortex + assert_eq!( + generate_flatten_file_key("files/default/logs/quickstart/2026/07/02/12/abc.vortex"), + format!("files{column_all}/default/logs/quickstart/2026/07/02/12/abc.parquet") + ); + } + // Helper function to create a record batch with the "_all" field fn create_test_batch_with_all_field( all_field_values: Vec>, diff --git a/src/service/compact/merge.rs b/src/service/compact/merge.rs index 87436d05d6..4fc48342d0 100644 --- a/src/service/compact/merge.rs +++ b/src/service/compact/merge.rs @@ -879,7 +879,11 @@ pub async fn merge_files( let mut new_files = Vec::new(); match buf { - MergeParquetResult::Single(buf, mut new_file_meta) => { + MergeParquetResult::Single { + buf, + file_meta: mut new_file_meta, + file_format, + } => { if new_file_meta.compressed_size == 0 { return Err(anyhow::anyhow!( "merge_parquet_files error: compressed_size is 0" @@ -887,8 +891,7 @@ pub async fn merge_files( } let id = ider::generate_file_name(); - let file_format = get_config().common.file_format.extension(); - let new_file_key = format!("{prefix}/{id}{file_format}"); + let new_file_key = format!("{prefix}/{id}{}", file_format.extension()); log::info!( "[COMPACTOR:WORKER:{thread_id}] merged {} files into a new file: {new_file_key}, original_size: {}, compressed_size: {}, took: {} ms", retain_file_list.len(), @@ -930,7 +933,11 @@ pub async fn merge_files( } new_files.push(FileKey::new(0, account, new_file_key, new_file_meta, false)); } - MergeParquetResult::Multiple { bufs, file_metas } => { + MergeParquetResult::Multiple { + bufs, + file_metas, + file_format, + } => { for (buf, file_meta) in bufs.into_iter().zip(file_metas) { let mut new_file_meta = file_meta; new_file_meta.compressed_size = buf.len() as i64; @@ -941,8 +948,7 @@ pub async fn merge_files( } let id = ider::generate_file_name(); - let file_format = get_config().common.file_format.extension(); - let new_file_key = format!("{prefix}/{id}{file_format}"); + let new_file_key = format!("{prefix}/{id}{}", file_format.extension()); // upload file to storage let buf = Bytes::from(buf); diff --git a/src/service/search/datafusion/merge/downsampling.rs b/src/service/search/datafusion/merge/downsampling.rs index 0ce9cc3993..c651f34417 100644 --- a/src/service/search/datafusion/merge/downsampling.rs +++ b/src/service/search/datafusion/merge/downsampling.rs @@ -57,6 +57,7 @@ pub async fn merge_parquet_files_with_downsampling( bloom_filter_fields: &[String], rule: &DownsamplingRule, metadata: &FileMeta, + file_format: FileFormat, ) -> Result { let start = std::time::Instant::now(); let cfg = get_config(); @@ -107,7 +108,7 @@ pub async fn merge_parquet_files_with_downsampling( }); // Write batches to the appropriate format - let (bufs, file_metas) = match cfg.common.file_format { + let (bufs, file_metas) = match file_format { FileFormat::Parquet => { write_downsampled_parquet(rx, &schema, bloom_filter_fields, metadata, &cfg).await? } @@ -133,7 +134,11 @@ pub async fn merge_parquet_files_with_downsampling( start.elapsed().as_millis() ); - Ok(MergeParquetResult::Multiple { bufs, file_metas }) + Ok(MergeParquetResult::Multiple { + bufs, + file_metas, + file_format, + }) } async fn write_downsampled_parquet( diff --git a/src/service/search/datafusion/merge/mod.rs b/src/service/search/datafusion/merge/mod.rs index 53fb50aa39..0ee49c3d1c 100644 --- a/src/service/search/datafusion/merge/mod.rs +++ b/src/service/search/datafusion/merge/mod.rs @@ -42,11 +42,16 @@ use { }; pub enum MergeParquetResult { - Single(Vec, FileMeta), + Single { + buf: Vec, + file_meta: FileMeta, + file_format: FileFormat, + }, #[allow(unused)] Multiple { bufs: Vec>, file_metas: Vec, + file_format: FileFormat, }, } @@ -62,6 +67,8 @@ pub async fn merge_parquet_files( let start = std::time::Instant::now(); let cfg = get_config(); + let file_format = merge_output_file_format(stream_type, is_ingester, cfg.common.file_format); + #[cfg(feature = "enterprise")] if stream_type == StreamType::Metrics && !is_ingester { let rule = get_largest_downsampling_rule(stream_name, metadata.max_ts); @@ -75,6 +82,7 @@ pub async fn merge_parquet_files( bloom_filter_fields, rule, &metadata, + file_format, ) .await; } @@ -150,8 +158,7 @@ pub async fn merge_parquet_files( Ok(()) }); - // write batches to the appropriate format - let buf = match cfg.common.file_format { + let buf = match file_format { FileFormat::Parquet => { write_parquet( &schema, @@ -172,7 +179,23 @@ pub async fn merge_parquet_files( ); metadata.compressed_size = buf.len() as i64; - Ok(MergeParquetResult::Single(buf, metadata)) + Ok(MergeParquetResult::Single { + buf, + file_meta: metadata, + file_format, + }) +} + +fn merge_output_file_format( + stream_type: StreamType, + is_ingester: bool, + configured: FileFormat, +) -> FileFormat { + if is_ingester { + FileFormat::for_ingester_stream(stream_type, configured) + } else { + configured + } } async fn write_parquet( @@ -274,6 +297,26 @@ mod tests { ])) } + #[test] + fn test_merge_output_file_format_uses_parquet_for_ingester_metrics() { + assert_eq!( + merge_output_file_format(StreamType::Metrics, true, FileFormat::Vortex), + FileFormat::Parquet + ); + assert_eq!( + merge_output_file_format(StreamType::Logs, true, FileFormat::Vortex), + FileFormat::Vortex + ); + assert_eq!( + merge_output_file_format(StreamType::Metrics, false, FileFormat::Vortex), + FileFormat::Vortex + ); + assert_eq!( + merge_output_file_format(StreamType::Logs, false, FileFormat::Parquet), + FileFormat::Parquet + ); + } + #[tokio::test] async fn test_merge_parquet_files_error_handling() { // Test with empty tables vector