From 8ef1fd4cbbeaa9be87cf8709e5c62f047b9684dd Mon Sep 17 00:00:00 2001 From: Huaijin Date: Thu, 20 Aug 2026 12:11:22 +0000 Subject: [PATCH] refactor(compaction): split merge/mod.rs by function into job/stream/file modules (#13900) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Pure code-move refactor: the 1795-line `src/compaction/src/merge/mod.rs` is split by responsibility into three modules under `merge/`: - **`job.rs`** — job generation: `generate_job_by_stream`, `generate_old_data_job_by_stream`, `generate_downsampling_job_by_stream_and_rule`, plus the shared `job_range_end` helper (now `pub(super)`) and its test - **`stream.rs`** — per-stream merge orchestration: `merge_by_stream` with its private helpers `write_file_list` and `sort_by_time_range`, plus all `sort_by_time_range` tests - **`file.rs`** — file-level merge execution (called by the worker): `merge_files` with its private helpers `cache_remote_files` and `generate_inverted_index` - **`mod.rs`** — now only module declarations and `pub use` re-exports; `tsid_major.rs` untouched No logic changes — only `use` blocks and relative paths were adjusted (`super::is_past_hour` → `crate::is_past_hour`, etc.). The public API paths are unchanged, so the call sites in `worker.rs` and `lib.rs` need no edits. ## Testing - `cargo check -p compaction` and `cargo clippy -p compaction` clean, zero warnings - All 30 merge-module tests pass (`cargo test -p compaction merge::`) --- src/compaction/src/merge/file.rs | 470 ++++++++ src/compaction/src/merge/job.rs | 362 ++++++ src/compaction/src/merge/mod.rs | 1787 +--------------------------- src/compaction/src/merge/stream.rs | 1012 ++++++++++++++++ 4 files changed, 1853 insertions(+), 1778 deletions(-) create mode 100644 src/compaction/src/merge/file.rs create mode 100644 src/compaction/src/merge/job.rs create mode 100644 src/compaction/src/merge/stream.rs diff --git a/src/compaction/src/merge/file.rs b/src/compaction/src/merge/file.rs new file mode 100644 index 0000000000..2d42646f9c --- /dev/null +++ b/src/compaction/src/merge/file.rs @@ -0,0 +1,470 @@ +// 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 ::datafusion::{arrow::datatypes::Schema, error::DataFusionError}; +use bytes::Bytes; +use config::{ + FileFormat, get_config, ider, is_local_disk_storage, + meta::stream::{FileKey, FileMeta, StorageType, StreamType}, + metrics, + utils::{parquet::read_schema_from_bytes, schema_ext::SchemaExt}, +}; +use hashbrown::{HashMap, HashSet}; +use infra::{ + cache::file_data, + file_list as infra_file_list, + runtime::DATAFUSION_RUNTIME, + schema::{ + SchemaCache, get_stream_setting_bloom_filter_fields, get_stream_setting_fts_fields, + get_stream_setting_index_fields, + }, + storage, +}; +use schema::generate_schema_for_defined_schema_fields; +use search::datafusion::{ + exec::TableBuilder, + merge::{self, MergeMode, MergeOutput, MergeParquetResult, MergedFile}, +}; +use tantivy_utils::index_builder::create_tantivy_index; +use tokio::sync::Semaphore; + +// merge small files into big file, upload to storage, returns the big file key and merged files +// params: +// - thread_id: the id of the thread +// - org_id: the id of the organization +// - stream_type: the type of the stream +// - stream_name: the name of the stream +// - prefix: the prefix of the files +// - files_with_size: the files to merge +// - mode: what the merge produces (decided by the scheduler) +// returns: +// - new_files: the files that are merged +// - retain_file_list: the files that are not merged +pub async fn merge_files( + thread_id: usize, + org_id: &str, + stream_type: StreamType, + stream_name: &str, + prefix: &str, + files_with_size: &[FileKey], + mode: &MergeMode, +) -> Result<(Vec, Vec), anyhow::Error> { + let start = std::time::Instant::now(); + // a whole-batch mode (downsampling) merges everything it is given, even a + // single file; otherwise 0/1 files means nothing to do + let merge_whole_batch = mode.merges_whole_batch(); + if files_with_size.len() <= 1 && !merge_whole_batch { + return Ok((Vec::new(), Vec::new())); + } + + let mut new_file_size = 0; + let mut new_compressed_file_size = 0; + let mut new_file_list = Vec::new(); + let cfg = get_config(); + for file in files_with_size.iter() { + if (new_file_size + file.meta.original_size > cfg.compact.max_file_size as i64 + || new_compressed_file_size + file.meta.compressed_size + > cfg.compact.max_file_size as i64) + && !merge_whole_batch + { + break; + } + new_file_size += file.meta.original_size; + new_compressed_file_size += file.meta.compressed_size; + new_file_list.push(file.clone()); + // metrics + metrics::COMPACT_MERGED_FILES + .with_label_values(&[org_id, stream_type.as_str()]) + .inc(); + metrics::COMPACT_MERGED_BYTES + .with_label_values(&[org_id, stream_type.as_str()]) + .inc_by(file.meta.original_size as u64); + } + // no files need to merge + if new_file_list.len() <= 1 && !merge_whole_batch { + return Ok((Vec::new(), Vec::new())); + } + + let retain_file_list = new_file_list.clone(); + + // cache parquet files + let deleted_files = cache_remote_files(&new_file_list).await?; + log::info!( + "[COMPACTOR:WORKER:{thread_id}] download {} parquet files, took: {} ms", + new_file_list.len(), + start.elapsed().as_millis() + ); + if !deleted_files.is_empty() { + new_file_list.retain(|f| !deleted_files.contains(&f.key)); + } + if new_file_list.len() <= 1 && !merge_whole_batch { + return Ok((Vec::new(), retain_file_list)); + } + + // get time range and stats for these files in a single iteration + let (min_ts, max_ts, total_records, new_file_size) = new_file_list.iter().fold( + (i64::MAX, i64::MIN, 0, 0), + |(min_ts, max_ts, records, size), file| { + ( + min_ts.min(file.meta.min_ts), + max_ts.max(file.meta.max_ts), + records + file.meta.records, + size + file.meta.original_size, + ) + }, + ); + let min_ts = if min_ts == i64::MAX { 0 } else { min_ts }; + let max_ts = if max_ts == i64::MIN { 0 } else { max_ts }; + let new_file_meta = FileMeta { + min_ts, + max_ts, + records: total_records, + original_size: new_file_size, + compressed_size: 0, + flattened: false, + index_size: 0, + bloom_ver: 0, + }; + if new_file_meta.records == 0 { + return Err(anyhow::anyhow!("merge_files error: records is 0")); + } + + // get latest version of schema + let latest_schema = infra::schema::get(org_id, stream_name, stream_type).await?; + let stream_settings = infra::schema::unwrap_stream_settings(&latest_schema); + let bloom_filter_fields = get_stream_setting_bloom_filter_fields(&stream_settings); + let full_text_search_fields = get_stream_setting_fts_fields(&stream_settings); + let index_fields = get_stream_setting_index_fields(&stream_settings); + let (defined_schema_fields, need_original, index_original_data, index_all_values, storage_type) = + match stream_settings { + Some(s) => ( + s.defined_schema_fields, + s.store_original_data, + s.index_original_data, + s.index_all_values, + s.storage_type, + ), + None => (Vec::new(), false, false, false, StorageType::Normal), + }; + let latest_schema = if !defined_schema_fields.is_empty() { + let latest_schema = SchemaCache::new(latest_schema); + let latest_schema = generate_schema_for_defined_schema_fields( + stream_type, + &latest_schema, + &defined_schema_fields, + need_original, + index_original_data, + index_all_values, + ); + latest_schema.schema().clone() + } else { + Arc::new(latest_schema) + }; + + // read schema from parquet file and group files by schema + let mut schemas = HashMap::new(); + let files = new_file_list.clone(); + let mut fi = 0; + for file in new_file_list.iter() { + fi += 1; + log::info!( + "[COMPACTOR:WORKER:{thread_id}:{fi}] merge small file: {}", + file.key + ); + let buf = file_data::get(&file.account, &file.key, None).await?; + let file_format = FileFormat::from_extension(&file.key) + .ok_or_else(|| anyhow::anyhow!("invalid file format: {}", file.key))?; + let schema = match read_schema_from_bytes(file_format, &buf).await { + Ok(schema) => schema, + Err(e) => { + log::error!( + "[COMPACTOR:WORKER:{thread_id}:{fi}] read schema error for file: {}, err: {e}", + file.key + ); + return Err(e); + } + }; + let schema = schema.as_ref().clone().with_metadata(Default::default()); + let schema_key = schema.hash_key(); + if !schemas.contains_key(&schema_key) { + schemas.insert(schema_key.clone(), schema); + } + } + + // generate the parquet schema + let all_fields = schemas + .values() + .flat_map(|s| s.fields().iter().map(|f| f.name().to_string())) + .collect::>(); + let schema = Arc::new(latest_schema.retain(all_fields)); + + // generate datafusion tables + let trace_id = ider::generate(); + let session = config::meta::search::Session { + id: trace_id.to_string(), + storage_type: config::meta::search::StorageType::Memory, + work_group: None, + 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(input_sort_order) + .build(session, files.clone(), schema.clone()) + .await + { + Ok(tables) => tables, + Err(e) => { + log::error!("create_parquet_table err: {e}, files: {files:?}, schema: {schema:?}"); + return Err(DataFusionError::Plan(format!("create_parquet_table err: {e}")).into()); + } + }; + + let merge_result = { + let mode = mode.clone(); + DATAFUSION_RUNTIME + .spawn(async move { + merge::merge_parquet_files( + schema, + tables, + &bloom_filter_fields, + new_file_meta, + &mode, + MergeOutput::for_compactor(stream_type), + ) + .await + }) + .await? + }; + + // clear session data + search::datafusion::storage::file_list::clear(&trace_id); + + let files = new_file_list.into_iter().map(|f| f.key).collect::>(); + let buf = match merge_result { + Ok(v) => v, + Err(e) => { + log::error!("merge_parquet_files err: {e}, files: {files:?}"); + return Err(DataFusionError::Plan(format!("merge_parquet_files err: {e}")).into()); + } + }; + + let latest_schema_fields = latest_schema + .fields() + .iter() + .map(|f| f.name()) + .collect::>(); + let need_index = full_text_search_fields + .iter() + .chain(index_fields.iter()) + .any(|f| latest_schema_fields.contains(f)); + if !need_index { + log::debug!("skip index generation for stream: {org_id}/{stream_type}/{stream_name}"); + } + + let MergeParquetResult { + files: merged_files, + file_format, + } = buf; + // an empty result would delete the source files without a replacement + if merged_files.is_empty() { + return Err(anyhow::anyhow!( + "merge_parquet_files error: produced no files" + )); + } + let mut new_files = Vec::with_capacity(merged_files.len()); + for MergedFile { + buf, + meta: mut new_file_meta, + layout, + } in merged_files + { + new_file_meta.compressed_size = buf.len() as i64; + if new_file_meta.compressed_size == 0 { + return Err(anyhow::anyhow!( + "merge_parquet_files error: compressed_size is 0" + )); + } + + let id = ider::generate_file_name(); + let new_file_key = format!("{prefix}/{}", layout.file_name(&id, file_format)); + + // upload file to storage + let buf = Bytes::from(buf); + if cfg.cache_latest_files.enabled + && cfg.cache_latest_files.cache_parquet + && cfg.cache_latest_files.download_from_node + { + infra::cache::file_data::disk::set(&new_file_key, buf.clone()).await?; + log::debug!("merge_files {new_file_key} file_data::disk::set success"); + } + + // TODO: check how compliance will interact with org storage + let account = storage::get_account(org_id, &new_file_key).unwrap_or_default(); + if cfg.s3.feature_force_infrequent_access && storage_type.is_compliance() { + storage::put_with_compliance(&account, &new_file_key, buf.clone()).await?; + } else { + storage::put(&account, &new_file_key, buf.clone()).await?; + } + + if cfg.search.inverted_index_enabled && stream_type.support_index() && need_index { + generate_inverted_index( + org_id, + &new_file_key, + &full_text_search_fields, + &index_fields, + &retain_file_list, + &mut new_file_meta, + latest_schema.clone(), + buf, + ) + .await?; + } + new_files.push(FileKey::new(0, account, new_file_key, new_file_meta, false)); + } + log::info!( + "[COMPACTOR:WORKER:{thread_id}] merged {} files into {} new file(s): {:?}, original_size: {}, compressed_size: {}, took: {} ms", + retain_file_list.len(), + new_files.len(), + new_files.iter().map(|f| f.key.as_str()).collect::>(), + new_files.iter().map(|f| f.meta.original_size).sum::(), + new_files + .iter() + .map(|f| f.meta.compressed_size) + .sum::(), + start.elapsed().as_millis(), + ); + + Ok((new_files, retain_file_list)) +} + +#[allow(clippy::too_many_arguments)] +async fn generate_inverted_index( + org_id: &str, + new_file_key: &str, + fts_fields: &[String], + index_fields: &[String], + retain_file_list: &[FileKey], + new_file_meta: &mut FileMeta, + latest_schema: Arc, + buf: Bytes, +) -> Result<(), anyhow::Error> { + let index_size = create_tantivy_index( + "COMPACTOR", + org_id, + new_file_key, + fts_fields, + index_fields, + latest_schema, // Use stream schema to include all configured fields + buf, + ) + .await + .map_err(|e| { + anyhow::anyhow!( + "create_tantivy_index_on_compactor for file: {new_file_key}, error: {e}, need delete files: {retain_file_list:?}", + ) + })?; + new_file_meta.index_size = index_size as i64; + + Ok(()) +} + +async fn cache_remote_files(files: &[FileKey]) -> Result, anyhow::Error> { + let cfg = get_config(); + let scan_size = files.iter().map(|f| f.meta.compressed_size).sum::(); + if is_local_disk_storage() + || !cfg.disk_cache.enabled + || scan_size >= cfg.disk_cache.skip_size as i64 + { + return Ok(Vec::new()); + }; + + let mut tasks = Vec::with_capacity(files.len()); + let semaphore = std::sync::Arc::new(Semaphore::new(cfg.limit.cpu_num)); + for file in files.iter() { + let file_account = file.account.to_string(); + let file_name = file.key.to_string(); + let file_size = file.meta.compressed_size as usize; + let permit = semaphore.clone().acquire_owned().await.unwrap(); + let task: tokio::task::JoinHandle> = tokio::task::spawn(async move { + let ret = if !file_data::disk::exist(&file_name).await { + file_data::disk::download(&file_account, &file_name, Some(file_size)).await + } else { + Ok(0) + }; + // In case where the parquet file is not found or has no data, we assume that it + // must have been deleted by some external entity, and hence we + // should remove the entry from file_list table. + let file_name = match ret { + Ok(data_len) => { + if data_len > 0 && data_len != file_size { + log::warn!( + "[COMPACT] download file {file_name} found size mismatch, expected: {file_size}, actual: {data_len}, will skip it", + ); + // skip this file for compact + Some(file_name) + } else { + None + } + } + Err(e) => { + if e.to_string().to_lowercase().contains("not found") + || e.to_string().to_lowercase().contains("data size is zero") + { + // delete file from file list + log::error!("[COMPACT] found invalid file: {file_name}, will delete it"); + if let Err(e) = + infra_file_list::delete_parquet_file(&file_account, &file_name, true) + .await + { + log::error!("[COMPACT] delete from file_list err: {e}"); + } + Some(file_name) + } else { + log::error!("[COMPACT] download file to cache err: {e}"); + // remove downloaded file + let _ = file_data::disk::remove(&file_name).await; + None + } + } + }; + drop(permit); + file_name + }); + tasks.push(task); + } + + let mut delete_files = Vec::new(); + for task in tasks { + match task.await { + Ok(file) => { + if let Some(file) = file { + delete_files.push(file); + } + } + Err(e) => { + log::error!("[COMPACTOR] load file task err: {e}"); + } + } + } + + Ok(delete_files) +} diff --git a/src/compaction/src/merge/job.rs b/src/compaction/src/merge/job.rs new file mode 100644 index 0000000000..926b538af0 --- /dev/null +++ b/src/compaction/src/merge/job.rs @@ -0,0 +1,362 @@ +// 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 chrono::{DateTime, Datelike, Duration, TimeZone, Utc}; +use config::{ + cluster::LOCAL_NODE, + get_config, + meta::stream::{PartitionTimeLevel, StreamType}, + utils::time::{day_micros, hour_micros}, +}; +use infra::{ + cluster::get_node_by_uuid, + dist_lock, file_list as infra_file_list, + schema::{get_partition_time_level, unwrap_stream_created_at}, +}; + +/// 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 +/// job can be newer, so "is the range old enough" is decided against this. +pub(super) fn job_range_end(offset: i64, partition_time_level: PartitionTimeLevel) -> i64 { + let offset = offset - offset % hour_micros(1); + if partition_time_level == PartitionTimeLevel::Daily { + offset - offset % day_micros(1) + day_micros(1) - 1 + } else { + offset + hour_micros(1) - 1 + } +} + +/// Generate merging job by stream +/// 1. get offset from db +/// 2. check if other node is processing +/// 3. create job or return +pub async fn generate_job_by_stream( + org_id: &str, + stream_type: StreamType, + stream_name: &str, +) -> Result<(), anyhow::Error> { + // get last compacted offset + let (mut offset, node) = db::compact::files::get_offset(org_id, stream_type, stream_name).await; + if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() { + return Ok(()); // other node is processing + } + + if node.is_empty() || LOCAL_NODE.uuid.ne(&node) { + let lock_key = format!("/compact/merge/{org_id}/{stream_type}/{stream_name}"); + let locker = dist_lock::lock(&lock_key, 0).await?; + // check the working node again, maybe other node locked it first + let (offset, node) = db::compact::files::get_offset(org_id, stream_type, stream_name).await; + if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() + { + dist_lock::unlock(&locker).await?; + return Ok(()); // other node is processing + } + // set to current node + let ret = db::compact::files::set_offset( + org_id, + stream_type, + stream_name, + offset, + Some(&LOCAL_NODE.uuid.clone()), + ) + .await; + dist_lock::unlock(&locker).await?; + drop(locker); + ret?; + } + + // get schema + let schema = infra::schema::get(org_id, stream_name, stream_type).await?; + let stream_created = unwrap_stream_created_at(&schema).unwrap_or_default(); + if offset == 0 && stream_created > 0 { + offset = stream_created + } else if offset == 0 { + return Ok(()); // no data + } + + // format to hour with zero minutes, seconds + let offset = offset - offset % hour_micros(1); + if !crate::is_past_hour(offset) { + return Ok(()); // the time is future, just wait + } + + log::debug!( + "[COMPACTOR] generate_job_by_stream [{org_id}/{stream_type}/{stream_name}] offset: {offset}" + ); + + // generate merging job + if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await { + return Err(anyhow::anyhow!( + "[COMPACTOR] add file_list_jobs failed: {e}" + )); + } + + // write new offset + let offset = offset + hour_micros(1); + db::compact::files::set_offset( + org_id, + stream_type, + stream_name, + offset, + Some(&LOCAL_NODE.uuid.clone()), + ) + .await?; + + Ok(()) +} + +/// Generate merging job by stream +/// 1. get old data by hour +/// 2. check if other node is processing +/// 3. create job or return +pub async fn generate_old_data_job_by_stream( + org_id: &str, + stream_type: StreamType, + stream_name: &str, +) -> Result<(), anyhow::Error> { + // get last compacted offset + let (offset, node) = db::compact::files::get_offset(org_id, stream_type, stream_name).await; + if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() { + return Ok(()); // other node is processing + } + + if node.is_empty() || LOCAL_NODE.uuid.ne(&node) { + let lock_key = format!("/compact/merge/{org_id}/{stream_type}/{stream_name}"); + let locker = dist_lock::lock(&lock_key, 0).await?; + // check the working node again, maybe other node locked it first + let (offset, node) = db::compact::files::get_offset(org_id, stream_type, stream_name).await; + if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() + { + dist_lock::unlock(&locker).await?; + return Ok(()); // other node is processing + } + // set to current node + let ret = db::compact::files::set_offset( + org_id, + stream_type, + stream_name, + offset, + Some(&LOCAL_NODE.uuid.clone()), + ) + .await; + dist_lock::unlock(&locker).await?; + drop(locker); + ret?; + } + + if offset == 0 { + return Ok(()); // no data + } + + let cfg = get_config(); + let stream_settings = infra::schema::get_settings(org_id, stream_name, stream_type) + .await + .unwrap_or_default(); + let mut stream_data_retention_days = cfg.compact.data_retention_days; + if stream_settings.data_retention > 0 { + stream_data_retention_days = stream_settings.data_retention; + } + if stream_data_retention_days > cfg.compact.old_data_max_days { + stream_data_retention_days = cfg.compact.old_data_max_days; + } + if stream_data_retention_days == 0 { + return Ok(()); // no need to check old data + } + + // get old data by hour, `offset - cfg.compact.old_data_min_hours hours` as old data + let end_time = offset - hour_micros(cfg.compact.old_data_min_hours); + let start_time = end_time + - Duration::try_days(stream_data_retention_days) + .unwrap() + .num_microseconds() + .unwrap(); + let hours = infra_file_list::query_old_data_hours( + org_id, + stream_type, + stream_name, + (start_time, end_time - 1), + ) + .await?; + + // generate merging job + for hour in hours { + let column = hour.split('/').collect::>(); + if column.len() != 4 { + return Err(anyhow::anyhow!( + "Unexpected hour format in {hour}, Expected format YYYY/MM/DD/HH", + )); + } + let offset = DateTime::parse_from_rfc3339(&format!( + "{}-{}-{}T{}:00:00Z", + column[0], column[1], column[2], column[3] + ))? + .with_timezone(&Utc); + let offset = offset.timestamp_micros(); + log::debug!( + "[COMPACTOR] generate_old_data_job_by_stream [{org_id}/{stream_type}/{stream_name}] hours: {hour}, offset: {offset}" + ); + if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await { + return Err(anyhow::anyhow!( + "[COMPACTOR] add file_list_jobs for old data failed: {e}" + )); + } + } + + Ok(()) +} + +/// Generate downsampling job by stream and rule +/// 1. get offset from db +/// 2. check if other node is processing +/// 3. create job or return +pub async fn generate_downsampling_job_by_stream_and_rule( + org_id: &str, + stream_type: StreamType, + stream_name: &str, + rule: (i64, i64), // offset, step +) -> Result<(), anyhow::Error> { + assert!(stream_type == StreamType::Metrics); + // get last compacted offset + let (mut offset, node) = + db::compact::downsampling::get_offset(org_id, stream_type, stream_name, rule).await; + if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() { + return Ok(()); // other node is processing + } + + if node.is_empty() || LOCAL_NODE.uuid.ne(&node) { + let lock_key = format!( + "/compact/downsampling/{org_id}/{stream_type}/{stream_name}/{}/{}", + rule.0, rule.1 + ); + let locker = dist_lock::lock(&lock_key, 0).await?; + // check the working node again, maybe other node locked it first + let (offset, node) = + db::compact::downsampling::get_offset(org_id, stream_type, stream_name, rule).await; + if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() + { + dist_lock::unlock(&locker).await?; + return Ok(()); // other node is processing + } + // set to current node + let ret = db::compact::downsampling::set_offset( + org_id, + stream_type, + stream_name, + rule, + offset, + Some(&LOCAL_NODE.uuid.clone()), + ) + .await; + dist_lock::unlock(&locker).await?; + drop(locker); + ret?; + } + + // get schema + let schema = infra::schema::get(org_id, stream_name, stream_type).await?; + let stream_created = unwrap_stream_created_at(&schema).unwrap_or_default(); + if offset == 0 { + offset = stream_created + } + if offset == 0 { + return Ok(()); // no data + } + + let cfg = get_config(); + // check offset + let time_now: DateTime = Utc::now(); + let time_now_day = Utc + .with_ymd_and_hms(time_now.year(), time_now.month(), time_now.day(), 0, 0, 0) + .unwrap() + .timestamp_micros(); + // must wait for at least 3 * max_file_retention_time + 1 day + // -- first period: the last hour local file upload to storage, write file list + // -- second period, the last hour file list upload to storage + // -- third period, we can do the merge, so, at least 3 times of + // -- 1 day, downsampling is in day level + // max_file_retention_time + // The job merges the hour of `offset` and decides the downsampling rule + // against the end of that hour (see merge_by_stream), so only enqueue it + // once the whole hour is older than the rule's offset; otherwise the job + // would run as a plain merge and the advanced offset would skip the hour. + let job_end_ts = job_range_end(offset, get_partition_time_level(stream_type)); + if offset >= time_now_day + || time_now.timestamp_micros() - offset + <= Duration::try_seconds(cfg.limit.max_file_retention_time as i64) + .unwrap() + .num_microseconds() + .unwrap() + * 3 + + day_micros(1) + || time_now.timestamp_micros() - rule.0 * 1_000_000 < job_end_ts + { + return Ok(()); // the time is future, just wait + } + + log::debug!( + "[DOWNSAMPLING] generate_downsampling_job_by_stream_and_rule [{org_id}/{stream_type}/{stream_name}] rule: {rule:?}, offset: {offset}" + ); + + // generate downsampling job + if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await { + return Err(anyhow::anyhow!( + "[DOWNSAMPLING] add file_list_jobs failed: {e}" + )); + } + + // write new offset + let offset = offset + day_micros(1); + // format to day with zero hour, minutes, seconds + let offset = offset - offset % day_micros(1); + db::compact::downsampling::set_offset( + org_id, + stream_type, + stream_name, + rule, + offset, + Some(&LOCAL_NODE.uuid.clone()), + ) + .await?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_job_range_end_hourly_and_daily() { + // 2026-08-18 10:10:00 UTC + let offset = Utc + .with_ymd_and_hms(2026, 8, 18, 10, 10, 0) + .unwrap() + .timestamp_micros(); + assert_eq!( + job_range_end(offset, PartitionTimeLevel::Hourly), + Utc.with_ymd_and_hms(2026, 8, 18, 11, 0, 0) + .unwrap() + .timestamp_micros() + - 1 + ); + assert_eq!( + job_range_end(offset, PartitionTimeLevel::Daily), + Utc.with_ymd_and_hms(2026, 8, 19, 0, 0, 0) + .unwrap() + .timestamp_micros() + - 1 + ); + } +} diff --git a/src/compaction/src/merge/mod.rs b/src/compaction/src/merge/mod.rs index 71c61504df..8d60729dd9 100644 --- a/src/compaction/src/merge/mod.rs +++ b/src/compaction/src/merge/mod.rs @@ -13,1783 +13,14 @@ // 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 ::datafusion::{arrow::datatypes::Schema, error::DataFusionError}; -use bytes::Bytes; -use chrono::{DateTime, Datelike, Duration, TimeZone, Utc}; -use config::{ - FileFormat, - cluster::LOCAL_NODE, - get_config, ider, is_local_disk_storage, - meta::{ - promql::tsid_layout::MetricsFileLayout, - stream::{ - FileKey, FileListDeleted, FileMeta, MergeStrategy, PartitionTimeLevel, StorageType, - StreamType, - }, - }, - metrics, - utils::{ - parquet::read_schema_from_bytes, - schema_ext::SchemaExt, - time::{day_micros, hour_micros}, - }, -}; -use hashbrown::{HashMap, HashSet}; -use infra::{ - cache::file_data, - cluster::get_node_by_uuid, - dist_lock, file_list as infra_file_list, - runtime::DATAFUSION_RUNTIME, - schema::{ - SchemaCache, get_partition_time_level, get_stream_setting_bloom_filter_fields, - get_stream_setting_fts_fields, get_stream_setting_index_fields, unwrap_stream_created_at, - }, - storage, -}; -use schema::generate_schema_for_defined_schema_fields; -use search::datafusion::{ - exec::TableBuilder, - merge::{self, MergeMode, MergeOutput, MergeParquetResult, MergedFile}, -}; -use search_service::file_list; -use tantivy_utils::index_builder::create_tantivy_index; -use tokio::{ - sync::{Semaphore, mpsc}, - task::JoinHandle, -}; - -use super::worker::{MergeBatch, MergeSender}; -use crate::merge::tsid_major::{TsidMajorMergeScope, tsid_major_merge_scope}; - +mod file; +mod job; +mod stream; 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 -/// job can be newer, so "is the range old enough" is decided against this. -fn job_range_end(offset: i64, partition_time_level: PartitionTimeLevel) -> i64 { - let offset = offset - offset % hour_micros(1); - if partition_time_level == PartitionTimeLevel::Daily { - offset - offset % day_micros(1) + day_micros(1) - 1 - } else { - offset + hour_micros(1) - 1 - } -} - -/// Generate merging job by stream -/// 1. get offset from db -/// 2. check if other node is processing -/// 3. create job or return -pub async fn generate_job_by_stream( - org_id: &str, - stream_type: StreamType, - stream_name: &str, -) -> Result<(), anyhow::Error> { - // get last compacted offset - let (mut offset, node) = db::compact::files::get_offset(org_id, stream_type, stream_name).await; - if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() { - return Ok(()); // other node is processing - } - - if node.is_empty() || LOCAL_NODE.uuid.ne(&node) { - let lock_key = format!("/compact/merge/{org_id}/{stream_type}/{stream_name}"); - let locker = dist_lock::lock(&lock_key, 0).await?; - // check the working node again, maybe other node locked it first - let (offset, node) = db::compact::files::get_offset(org_id, stream_type, stream_name).await; - if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() - { - dist_lock::unlock(&locker).await?; - return Ok(()); // other node is processing - } - // set to current node - let ret = db::compact::files::set_offset( - org_id, - stream_type, - stream_name, - offset, - Some(&LOCAL_NODE.uuid.clone()), - ) - .await; - dist_lock::unlock(&locker).await?; - drop(locker); - ret?; - } - - // get schema - let schema = infra::schema::get(org_id, stream_name, stream_type).await?; - let stream_created = unwrap_stream_created_at(&schema).unwrap_or_default(); - if offset == 0 && stream_created > 0 { - offset = stream_created - } else if offset == 0 { - return Ok(()); // no data - } - - // format to hour with zero minutes, seconds - let offset = offset - offset % hour_micros(1); - if !super::is_past_hour(offset) { - return Ok(()); // the time is future, just wait - } - - log::debug!( - "[COMPACTOR] generate_job_by_stream [{org_id}/{stream_type}/{stream_name}] offset: {offset}" - ); - - // generate merging job - if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await { - return Err(anyhow::anyhow!( - "[COMPACTOR] add file_list_jobs failed: {e}" - )); - } - - // write new offset - let offset = offset + hour_micros(1); - db::compact::files::set_offset( - org_id, - stream_type, - stream_name, - offset, - Some(&LOCAL_NODE.uuid.clone()), - ) - .await?; - - Ok(()) -} - -/// Generate merging job by stream -/// 1. get old data by hour -/// 2. check if other node is processing -/// 3. create job or return -pub async fn generate_old_data_job_by_stream( - org_id: &str, - stream_type: StreamType, - stream_name: &str, -) -> Result<(), anyhow::Error> { - // get last compacted offset - let (offset, node) = db::compact::files::get_offset(org_id, stream_type, stream_name).await; - if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() { - return Ok(()); // other node is processing - } - - if node.is_empty() || LOCAL_NODE.uuid.ne(&node) { - let lock_key = format!("/compact/merge/{org_id}/{stream_type}/{stream_name}"); - let locker = dist_lock::lock(&lock_key, 0).await?; - // check the working node again, maybe other node locked it first - let (offset, node) = db::compact::files::get_offset(org_id, stream_type, stream_name).await; - if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() - { - dist_lock::unlock(&locker).await?; - return Ok(()); // other node is processing - } - // set to current node - let ret = db::compact::files::set_offset( - org_id, - stream_type, - stream_name, - offset, - Some(&LOCAL_NODE.uuid.clone()), - ) - .await; - dist_lock::unlock(&locker).await?; - drop(locker); - ret?; - } - - if offset == 0 { - return Ok(()); // no data - } - - let cfg = get_config(); - let stream_settings = infra::schema::get_settings(org_id, stream_name, stream_type) - .await - .unwrap_or_default(); - let mut stream_data_retention_days = cfg.compact.data_retention_days; - if stream_settings.data_retention > 0 { - stream_data_retention_days = stream_settings.data_retention; - } - if stream_data_retention_days > cfg.compact.old_data_max_days { - stream_data_retention_days = cfg.compact.old_data_max_days; - } - if stream_data_retention_days == 0 { - return Ok(()); // no need to check old data - } - - // get old data by hour, `offset - cfg.compact.old_data_min_hours hours` as old data - let end_time = offset - hour_micros(cfg.compact.old_data_min_hours); - let start_time = end_time - - Duration::try_days(stream_data_retention_days) - .unwrap() - .num_microseconds() - .unwrap(); - let hours = infra_file_list::query_old_data_hours( - org_id, - stream_type, - stream_name, - (start_time, end_time - 1), - ) - .await?; - - // generate merging job - for hour in hours { - let column = hour.split('/').collect::>(); - if column.len() != 4 { - return Err(anyhow::anyhow!( - "Unexpected hour format in {hour}, Expected format YYYY/MM/DD/HH", - )); - } - let offset = DateTime::parse_from_rfc3339(&format!( - "{}-{}-{}T{}:00:00Z", - column[0], column[1], column[2], column[3] - ))? - .with_timezone(&Utc); - let offset = offset.timestamp_micros(); - log::debug!( - "[COMPACTOR] generate_old_data_job_by_stream [{org_id}/{stream_type}/{stream_name}] hours: {hour}, offset: {offset}" - ); - if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await { - return Err(anyhow::anyhow!( - "[COMPACTOR] add file_list_jobs for old data failed: {e}" - )); - } - } - - Ok(()) -} - -/// Generate downsampling job by stream and rule -/// 1. get offset from db -/// 2. check if other node is processing -/// 3. create job or return -pub async fn generate_downsampling_job_by_stream_and_rule( - org_id: &str, - stream_type: StreamType, - stream_name: &str, - rule: (i64, i64), // offset, step -) -> Result<(), anyhow::Error> { - assert!(stream_type == StreamType::Metrics); - // get last compacted offset - let (mut offset, node) = - db::compact::downsampling::get_offset(org_id, stream_type, stream_name, rule).await; - if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() { - return Ok(()); // other node is processing - } - - if node.is_empty() || LOCAL_NODE.uuid.ne(&node) { - let lock_key = format!( - "/compact/downsampling/{org_id}/{stream_type}/{stream_name}/{}/{}", - rule.0, rule.1 - ); - let locker = dist_lock::lock(&lock_key, 0).await?; - // check the working node again, maybe other node locked it first - let (offset, node) = - db::compact::downsampling::get_offset(org_id, stream_type, stream_name, rule).await; - if !node.is_empty() && LOCAL_NODE.uuid.ne(&node) && get_node_by_uuid(&node).await.is_some() - { - dist_lock::unlock(&locker).await?; - return Ok(()); // other node is processing - } - // set to current node - let ret = db::compact::downsampling::set_offset( - org_id, - stream_type, - stream_name, - rule, - offset, - Some(&LOCAL_NODE.uuid.clone()), - ) - .await; - dist_lock::unlock(&locker).await?; - drop(locker); - ret?; - } - - // get schema - let schema = infra::schema::get(org_id, stream_name, stream_type).await?; - let stream_created = unwrap_stream_created_at(&schema).unwrap_or_default(); - if offset == 0 { - offset = stream_created - } - if offset == 0 { - return Ok(()); // no data - } - - let cfg = get_config(); - // check offset - let time_now: DateTime = Utc::now(); - let time_now_day = Utc - .with_ymd_and_hms(time_now.year(), time_now.month(), time_now.day(), 0, 0, 0) - .unwrap() - .timestamp_micros(); - // must wait for at least 3 * max_file_retention_time + 1 day - // -- first period: the last hour local file upload to storage, write file list - // -- second period, the last hour file list upload to storage - // -- third period, we can do the merge, so, at least 3 times of - // -- 1 day, downsampling is in day level - // max_file_retention_time - // The job merges the hour of `offset` and decides the downsampling rule - // against the end of that hour (see merge_by_stream), so only enqueue it - // once the whole hour is older than the rule's offset; otherwise the job - // would run as a plain merge and the advanced offset would skip the hour. - let job_end_ts = job_range_end(offset, get_partition_time_level(stream_type)); - if offset >= time_now_day - || time_now.timestamp_micros() - offset - <= Duration::try_seconds(cfg.limit.max_file_retention_time as i64) - .unwrap() - .num_microseconds() - .unwrap() - * 3 - + day_micros(1) - || time_now.timestamp_micros() - rule.0 * 1_000_000 < job_end_ts - { - return Ok(()); // the time is future, just wait - } - - log::debug!( - "[DOWNSAMPLING] generate_downsampling_job_by_stream_and_rule [{org_id}/{stream_type}/{stream_name}] rule: {rule:?}, offset: {offset}" - ); - - // generate downsampling job - if let Err(e) = infra_file_list::add_job(org_id, stream_type, stream_name, offset).await { - return Err(anyhow::anyhow!( - "[DOWNSAMPLING] add file_list_jobs failed: {e}" - )); - } - - // write new offset - let offset = offset + day_micros(1); - // format to day with zero hour, minutes, seconds - let offset = offset - offset % day_micros(1); - db::compact::downsampling::set_offset( - org_id, - stream_type, - stream_name, - rule, - offset, - Some(&LOCAL_NODE.uuid.clone()), - ) - .await?; - - Ok(()) -} - -/// compactor run steps on a stream: -/// 3. get a cluster lock for compactor stream -/// 4. read last compacted offset: year/month/day/hour -/// 5. read current hour all files -/// 6. compact small files to big files -> COMPACTOR_MAX_FILE_SIZE -/// 7. write to storage -/// 8. delete small files keys & write big files keys, use transaction -/// 9. delete small files from storage -/// 10. update last compacted offset -/// 11. release cluster lock -pub async fn merge_by_stream( - worker_tx: mpsc::Sender<(MergeSender, MergeBatch)>, - org_id: &str, - stream_type: StreamType, - stream_name: &str, - job_id: i64, - offset: i64, -) -> Result<(), anyhow::Error> { - let cfg = get_config(); - let start = std::time::Instant::now(); - - // get schema - let schema = infra::schema::get(org_id, stream_name, stream_type).await?; - if schema == Schema::empty() { - // the stream was deleted, mark the job as done - if let Err(e) = infra_file_list::set_job_done(&[job_id]).await { - log::error!("[COMPACTOR] set_job_done failed: {e}"); - } - return Ok(()); - } - - log::debug!( - "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] offset: {offset}" - ); - - // A job whose offset hour has not yet settled (hour end + 3 * max_file_retention_time, - // see is_past_hour) is an incremental round on a hour that can still receive files - // (enqueued by the ingester, see service::compaction::incremental): only seal full-size - // groups and carry the remainder, so each file is merged into a sealed output exactly - // once. The scheduled hour-end pass seals whatever is left. - let offset = offset - offset % hour_micros(1); - let is_incremental = !super::is_past_hour(offset); - - // check offset - let partition_time_level = get_partition_time_level(stream_type); - let offset_time: DateTime = Utc.timestamp_nanos(offset * 1000); - let (date_start, date_end) = if partition_time_level == PartitionTimeLevel::Daily { - ( - offset_time.format("%Y/%m/%d/00").to_string(), - offset_time.format("%Y/%m/%d/23").to_string(), - ) - } else { - ( - offset_time.format("%Y/%m/%d/%H").to_string(), - offset_time.format("%Y/%m/%d/%H").to_string(), - ) - }; - // What this hour of files turns into. Decided once here — the workers and - // 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, - &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() { - i64::MAX - } else { - infra_file_list::merge_max_original_size() - }; - let files = file_list::query_for_merge( - org_id, - stream_type, - stream_name, - &date_start, - &date_end, - max_original_size, - ) - .await - .map_err(|e| anyhow::anyhow!("query file list failed: {e}"))?; - - log::debug!( - "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] date range: [{date_start},{date_end}], files: {}", - files.len(), - ); - - if files.is_empty() { - // update job status - if let Err(e) = infra_file_list::set_job_done(&[job_id]).await { - log::error!("[COMPACTOR] set_job_done failed: {e}"); - } - return Ok(()); - } - - // do partition by partition key - let mut partition_files_with_size: HashMap> = HashMap::default(); - for file in files { - let file_name = file.key.clone(); - let prefix = file_name[..file_name.rfind('/').unwrap()].to_string(); - let partition = partition_files_with_size.entry(prefix).or_default(); - partition.push(file.to_owned()); - } - - // use multiple threads to merge - let semaphore = std::sync::Arc::new(Semaphore::new(cfg.limit.file_merge_thread_num)); - let mut tasks = Vec::with_capacity(partition_files_with_size.len()); - for (prefix, mut files_with_size) in partition_files_with_size.into_iter() { - let org_id = org_id.to_string(); - let stream_name = stream_name.to_string(); - let mode = mode.clone(); - let permit = semaphore.clone().acquire_owned().await.unwrap(); - let worker_tx = worker_tx.clone(); - let task: JoinHandle, anyhow::Error>> = tokio::task::spawn(async move { - let cfg = get_config(); - // sort by file size - let job_strategy = MergeStrategy::from(&cfg.compact.strategy); - match job_strategy { - MergeStrategy::FileSize => { - files_with_size.sort_by_key(|k| k.meta.original_size); - } - MergeStrategy::FileTime => { - files_with_size.sort_by_key(|k| k.meta.min_ts); - } - MergeStrategy::TimeRange => { - files_with_size = sort_by_time_range(files_with_size); - } - } - - 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(); - if mode.merges_whole_batch() { - batch_groups.push(MergeBatch { - batch_id: 0, - org_id: org_id.clone(), - stream_type, - stream_name: stream_name.clone(), - prefix: prefix.clone(), - files: files_with_size.clone(), - mode: mode.clone(), - }); - } else { - let mut new_file_list = Vec::new(); - let mut new_file_size = 0; - for file in files_with_size.iter() { - if new_file_size + file.meta.original_size > cfg.compact.max_file_size as i64 - || (cfg.compact.max_group_files > 0 - && new_file_list.len() >= cfg.compact.max_group_files) - { - if new_file_list.len() <= 1 { - if job_strategy == MergeStrategy::FileSize { - break; - } - new_file_list.clear(); - new_file_size = file.meta.original_size; - new_file_list.push(file.clone()); - continue; // replace previous file with current file - } - batch_groups.push(MergeBatch { - batch_id: batch_groups.len(), - org_id: org_id.clone(), - stream_type, - stream_name: stream_name.clone(), - prefix: prefix.clone(), - files: new_file_list.clone(), - mode: mode.clone(), - }); - new_file_size = 0; - new_file_list.clear(); - } - new_file_size += file.meta.original_size; - new_file_list.push(file.clone()); - } - // The trailing batch is always below max_file_size (the loop flushes a group - // only when adding the next file would exceed it). In incremental mode we do - // NOT seal this remainder: more files will arrive in the still-open hour, and - // sealing now would force re-merging it later (write amplification). Carry it - // to the next round; the scheduled hour-end pass seals whatever is left. - if new_file_list.len() > 1 && !is_incremental { - batch_groups.push(MergeBatch { - batch_id: batch_groups.len(), - org_id: org_id.clone(), - stream_type, - stream_name: stream_name.clone(), - prefix: prefix.clone(), - files: new_file_list.clone(), - mode: mode.clone(), - }); - } - - if batch_groups.is_empty() { - return Ok(vec![]); // no files need to merge - } - } - - // send to worker - let batch_group_len = batch_groups.len(); - let (inner_tx, mut inner_rx) = mpsc::channel(batch_group_len); - for batch in batch_groups.iter() { - if let Err(e) = worker_tx.send((inner_tx.clone(), batch.clone())).await { - log::error!("[COMPACTOR] send batch to worker failed: {e}"); - return Err(anyhow::Error::msg("send batch to worker failed")); - } - } - let mut worker_results = Vec::with_capacity(batch_group_len); - for _ in 0..batch_group_len { - let result = inner_rx.recv().await.unwrap(); - worker_results.push(result); - } - - let mut last_error = None; - let mut check_guard = HashSet::with_capacity(batch_groups.len()); - let mut orphan_blooms = Vec::new(); - for ret in worker_results { - let (batch_id, new_files) = match ret { - Ok(v) => v, - Err(e) => { - log::error!("[COMPACTOR] merge files failed: {e}"); - last_error = Some(e); - continue; - } - }; - - if check_guard.contains(&batch_id) { - log::warn!( - "[COMPACTOR] merge files for stream: [{org_id}/{stream_type}/{stream_name}] found error files, batch_id: {batch_id} duplicate" - ); - continue; - } - check_guard.insert(batch_id); - - // delete small files keys & write big files keys, use transaction - let delete_file_list = batch_groups.get(batch_id).unwrap().files.as_slice(); - let mut events = Vec::with_capacity(new_files.len() + delete_file_list.len()); - for new_file in new_files { - if !new_file.key.is_empty() { - events.push(new_file); - } - } - - for file in delete_file_list { - events.push(FileKey { - deleted: true, - selection: None, - row_group_size: None, - ..file.clone() - }); - } - events.sort_by(|a, b| a.key.cmp(&b.key)); - - // write file list to storage - if let Err(e) = write_file_list(&org_id, stream_type, &events).await { - log::error!("[COMPACTOR] write file list failed: {e}"); - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; - continue; - } - - // collect orphan blooms after writing file list successfully - for file in delete_file_list { - if file.meta.bloom_ver > 0 { - orphan_blooms.push(file.meta.bloom_ver); - } - } - } - drop(permit); - if let Some(e) = last_error { - return Err(e); - } - Ok(orphan_blooms) - }); - tasks.push(task); - } - - // collect bloom files which need to be clean - let mut orphan_blooms = Vec::new(); - for task in tasks { - orphan_blooms.extend(task.await??); - } - - // Build bloom filters for the current hour. Failures are non-fatal: files - // left at bloom_ver = 0 fall back to the regular Tantivy search path. - let build_start = std::time::Instant::now(); - match crate::bloom::compact::build_for_stream( - org_id, - stream_type, - stream_name, - &date_start, - is_incremental, - orphan_blooms, - ) - .await - { - Ok(false) => {} - Ok(true) => { - let build_time = build_start.elapsed().as_millis(); - log::info!( - "[COMPACTOR] bloom build for {org_id}/{stream_type}/{stream_name}/{date_start} took: {build_time} ms" - ); - } - Err(e) => { - log::warn!( - "[COMPACTOR] bloom build for {org_id}/{stream_type}/{stream_name}/{date_start} failed: {e}" - ); - } - } - - // update job status - if let Err(e) = infra_file_list::set_job_done(&[job_id]).await { - log::error!("[COMPACTOR] set_job_done failed: {e}"); - } - - // metrics - let time = start.elapsed().as_secs_f64(); - metrics::COMPACT_USED_TIME - .with_label_values(&[org_id, stream_type.as_str()]) - .inc_by(time); - - Ok(()) -} - -// merge small files into big file, upload to storage, returns the big file key and merged files -// params: -// - thread_id: the id of the thread -// - org_id: the id of the organization -// - stream_type: the type of the stream -// - stream_name: the name of the stream -// - prefix: the prefix of the files -// - files_with_size: the files to merge -// - mode: what the merge produces (decided by the scheduler) -// returns: -// - new_files: the files that are merged -// - retain_file_list: the files that are not merged -pub async fn merge_files( - thread_id: usize, - org_id: &str, - stream_type: StreamType, - stream_name: &str, - prefix: &str, - files_with_size: &[FileKey], - mode: &MergeMode, -) -> Result<(Vec, Vec), anyhow::Error> { - let start = std::time::Instant::now(); - // a whole-batch mode (downsampling) merges everything it is given, even a - // single file; otherwise 0/1 files means nothing to do - let merge_whole_batch = mode.merges_whole_batch(); - if files_with_size.len() <= 1 && !merge_whole_batch { - return Ok((Vec::new(), Vec::new())); - } - - let mut new_file_size = 0; - let mut new_compressed_file_size = 0; - let mut new_file_list = Vec::new(); - let cfg = get_config(); - for file in files_with_size.iter() { - if (new_file_size + file.meta.original_size > cfg.compact.max_file_size as i64 - || new_compressed_file_size + file.meta.compressed_size - > cfg.compact.max_file_size as i64) - && !merge_whole_batch - { - break; - } - new_file_size += file.meta.original_size; - new_compressed_file_size += file.meta.compressed_size; - new_file_list.push(file.clone()); - // metrics - metrics::COMPACT_MERGED_FILES - .with_label_values(&[org_id, stream_type.as_str()]) - .inc(); - metrics::COMPACT_MERGED_BYTES - .with_label_values(&[org_id, stream_type.as_str()]) - .inc_by(file.meta.original_size as u64); - } - // no files need to merge - if new_file_list.len() <= 1 && !merge_whole_batch { - return Ok((Vec::new(), Vec::new())); - } - - let retain_file_list = new_file_list.clone(); - - // cache parquet files - let deleted_files = cache_remote_files(&new_file_list).await?; - log::info!( - "[COMPACTOR:WORKER:{thread_id}] download {} parquet files, took: {} ms", - new_file_list.len(), - start.elapsed().as_millis() - ); - if !deleted_files.is_empty() { - new_file_list.retain(|f| !deleted_files.contains(&f.key)); - } - if new_file_list.len() <= 1 && !merge_whole_batch { - return Ok((Vec::new(), retain_file_list)); - } - - // get time range and stats for these files in a single iteration - let (min_ts, max_ts, total_records, new_file_size) = new_file_list.iter().fold( - (i64::MAX, i64::MIN, 0, 0), - |(min_ts, max_ts, records, size), file| { - ( - min_ts.min(file.meta.min_ts), - max_ts.max(file.meta.max_ts), - records + file.meta.records, - size + file.meta.original_size, - ) - }, - ); - let min_ts = if min_ts == i64::MAX { 0 } else { min_ts }; - let max_ts = if max_ts == i64::MIN { 0 } else { max_ts }; - let new_file_meta = FileMeta { - min_ts, - max_ts, - records: total_records, - original_size: new_file_size, - compressed_size: 0, - flattened: false, - index_size: 0, - bloom_ver: 0, - }; - if new_file_meta.records == 0 { - return Err(anyhow::anyhow!("merge_files error: records is 0")); - } - - // get latest version of schema - let latest_schema = infra::schema::get(org_id, stream_name, stream_type).await?; - let stream_settings = infra::schema::unwrap_stream_settings(&latest_schema); - let bloom_filter_fields = get_stream_setting_bloom_filter_fields(&stream_settings); - let full_text_search_fields = get_stream_setting_fts_fields(&stream_settings); - let index_fields = get_stream_setting_index_fields(&stream_settings); - let (defined_schema_fields, need_original, index_original_data, index_all_values, storage_type) = - match stream_settings { - Some(s) => ( - s.defined_schema_fields, - s.store_original_data, - s.index_original_data, - s.index_all_values, - s.storage_type, - ), - None => (Vec::new(), false, false, false, StorageType::Normal), - }; - let latest_schema = if !defined_schema_fields.is_empty() { - let latest_schema = SchemaCache::new(latest_schema); - let latest_schema = generate_schema_for_defined_schema_fields( - stream_type, - &latest_schema, - &defined_schema_fields, - need_original, - index_original_data, - index_all_values, - ); - latest_schema.schema().clone() - } else { - Arc::new(latest_schema) - }; - - // read schema from parquet file and group files by schema - let mut schemas = HashMap::new(); - let files = new_file_list.clone(); - let mut fi = 0; - for file in new_file_list.iter() { - fi += 1; - log::info!( - "[COMPACTOR:WORKER:{thread_id}:{fi}] merge small file: {}", - file.key - ); - let buf = file_data::get(&file.account, &file.key, None).await?; - let file_format = FileFormat::from_extension(&file.key) - .ok_or_else(|| anyhow::anyhow!("invalid file format: {}", file.key))?; - let schema = match read_schema_from_bytes(file_format, &buf).await { - Ok(schema) => schema, - Err(e) => { - log::error!( - "[COMPACTOR:WORKER:{thread_id}:{fi}] read schema error for file: {}, err: {e}", - file.key - ); - return Err(e); - } - }; - let schema = schema.as_ref().clone().with_metadata(Default::default()); - let schema_key = schema.hash_key(); - if !schemas.contains_key(&schema_key) { - schemas.insert(schema_key.clone(), schema); - } - } - - // generate the parquet schema - let all_fields = schemas - .values() - .flat_map(|s| s.fields().iter().map(|f| f.name().to_string())) - .collect::>(); - let schema = Arc::new(latest_schema.retain(all_fields)); - - // generate datafusion tables - let trace_id = ider::generate(); - let session = config::meta::search::Session { - id: trace_id.to_string(), - storage_type: config::meta::search::StorageType::Memory, - work_group: None, - 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(input_sort_order) - .build(session, files.clone(), schema.clone()) - .await - { - Ok(tables) => tables, - Err(e) => { - log::error!("create_parquet_table err: {e}, files: {files:?}, schema: {schema:?}"); - return Err(DataFusionError::Plan(format!("create_parquet_table err: {e}")).into()); - } - }; - - let merge_result = { - let mode = mode.clone(); - DATAFUSION_RUNTIME - .spawn(async move { - merge::merge_parquet_files( - schema, - tables, - &bloom_filter_fields, - new_file_meta, - &mode, - MergeOutput::for_compactor(stream_type), - ) - .await - }) - .await? - }; - - // clear session data - search::datafusion::storage::file_list::clear(&trace_id); - - let files = new_file_list.into_iter().map(|f| f.key).collect::>(); - let buf = match merge_result { - Ok(v) => v, - Err(e) => { - log::error!("merge_parquet_files err: {e}, files: {files:?}"); - return Err(DataFusionError::Plan(format!("merge_parquet_files err: {e}")).into()); - } - }; - - let latest_schema_fields = latest_schema - .fields() - .iter() - .map(|f| f.name()) - .collect::>(); - let need_index = full_text_search_fields - .iter() - .chain(index_fields.iter()) - .any(|f| latest_schema_fields.contains(f)); - if !need_index { - log::debug!("skip index generation for stream: {org_id}/{stream_type}/{stream_name}"); - } - - let MergeParquetResult { - files: merged_files, - file_format, - } = buf; - // an empty result would delete the source files without a replacement - if merged_files.is_empty() { - return Err(anyhow::anyhow!( - "merge_parquet_files error: produced no files" - )); - } - let mut new_files = Vec::with_capacity(merged_files.len()); - for MergedFile { - buf, - meta: mut new_file_meta, - layout, - } in merged_files - { - new_file_meta.compressed_size = buf.len() as i64; - if new_file_meta.compressed_size == 0 { - return Err(anyhow::anyhow!( - "merge_parquet_files error: compressed_size is 0" - )); - } - - let id = ider::generate_file_name(); - let new_file_key = format!("{prefix}/{}", layout.file_name(&id, file_format)); - - // upload file to storage - let buf = Bytes::from(buf); - if cfg.cache_latest_files.enabled - && cfg.cache_latest_files.cache_parquet - && cfg.cache_latest_files.download_from_node - { - infra::cache::file_data::disk::set(&new_file_key, buf.clone()).await?; - log::debug!("merge_files {new_file_key} file_data::disk::set success"); - } - - // TODO: check how compliance will interact with org storage - let account = storage::get_account(org_id, &new_file_key).unwrap_or_default(); - if cfg.s3.feature_force_infrequent_access && storage_type.is_compliance() { - storage::put_with_compliance(&account, &new_file_key, buf.clone()).await?; - } else { - storage::put(&account, &new_file_key, buf.clone()).await?; - } - - if cfg.search.inverted_index_enabled && stream_type.support_index() && need_index { - generate_inverted_index( - org_id, - &new_file_key, - &full_text_search_fields, - &index_fields, - &retain_file_list, - &mut new_file_meta, - latest_schema.clone(), - buf, - ) - .await?; - } - new_files.push(FileKey::new(0, account, new_file_key, new_file_meta, false)); - } - log::info!( - "[COMPACTOR:WORKER:{thread_id}] merged {} files into {} new file(s): {:?}, original_size: {}, compressed_size: {}, took: {} ms", - retain_file_list.len(), - new_files.len(), - new_files.iter().map(|f| f.key.as_str()).collect::>(), - new_files.iter().map(|f| f.meta.original_size).sum::(), - new_files - .iter() - .map(|f| f.meta.compressed_size) - .sum::(), - start.elapsed().as_millis(), - ); - - Ok((new_files, retain_file_list)) -} - -#[allow(clippy::too_many_arguments)] -async fn generate_inverted_index( - org_id: &str, - new_file_key: &str, - fts_fields: &[String], - index_fields: &[String], - retain_file_list: &[FileKey], - new_file_meta: &mut FileMeta, - latest_schema: Arc, - buf: Bytes, -) -> Result<(), anyhow::Error> { - let index_size = create_tantivy_index( - "COMPACTOR", - org_id, - new_file_key, - fts_fields, - index_fields, - latest_schema, // Use stream schema to include all configured fields - buf, - ) - .await - .map_err(|e| { - anyhow::anyhow!( - "create_tantivy_index_on_compactor for file: {new_file_key}, error: {e}, need delete files: {retain_file_list:?}", - ) - })?; - new_file_meta.index_size = index_size as i64; - - Ok(()) -} - -async fn write_file_list( - org_id: &str, - stream_type: StreamType, - events: &[FileKey], -) -> Result<(), anyhow::Error> { - if events.is_empty() { - return Ok(()); - } - - let del_items = events - .iter() - .filter(|v| v.deleted) - .map(|v| FileListDeleted { - id: 0, - account: v.account.clone(), - file: v.key.clone(), - index_file: v.meta.index_size > 0, - flattened: v.meta.flattened, - }) - .collect::>(); - - // set to db - // retry 5 times - let cfg = get_config(); - let mut success = false; - let mut mark_deleted_done = false; - let created_at = config::utils::time::now_micros(); - for _ in 0..5 { - if !mark_deleted_done && let Err(e) = infra::file_list::batch_process(events).await { - log::error!("[COMPACTOR] batch_process to db failed, retrying: {e}"); - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; - continue; - } - mark_deleted_done = true; - if !del_items.is_empty() - && let Err(e) = infra_file_list::batch_add_deleted(org_id, created_at, &del_items).await - { - log::error!("[COMPACTOR] batch_add_deleted to db failed, retrying: {e}"); - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; - continue; - } - success = true; - break; - } - - // handle dump_stats for file_list type streams - if success && stream_type == StreamType::Filelist && cfg.compact.file_list_dump_enabled { - let (deleted_files, new_files): (Vec<_>, Vec<_>) = events.iter().partition(|e| e.deleted); - super::dump::handle_dump_stats_on_merge(&deleted_files, &new_files).await; - } - - if success { - // send broadcast to other nodes - if cfg.cache_latest_files.enabled { - // get id for all the new files - let file_ids = infra_file_list::query_ids_by_files(events).await?; - let mut events = events.to_vec(); - for event in events.iter_mut() { - if let Some(id) = file_ids.get(&event.key) { - event.id = *id; - } - } - if let Err(e) = db::file_list::broadcast::send(&events).await { - log::error!("[COMPACTOR] send broadcast for file_list failed: {e}"); - } - } - } else { - return Err(anyhow::anyhow!("batch_write to db failed")); - } - - Ok(()) -} - -async fn cache_remote_files(files: &[FileKey]) -> Result, anyhow::Error> { - let cfg = get_config(); - let scan_size = files.iter().map(|f| f.meta.compressed_size).sum::(); - if is_local_disk_storage() - || !cfg.disk_cache.enabled - || scan_size >= cfg.disk_cache.skip_size as i64 - { - return Ok(Vec::new()); - }; - - let mut tasks = Vec::with_capacity(files.len()); - let semaphore = std::sync::Arc::new(Semaphore::new(cfg.limit.cpu_num)); - for file in files.iter() { - let file_account = file.account.to_string(); - let file_name = file.key.to_string(); - let file_size = file.meta.compressed_size as usize; - let permit = semaphore.clone().acquire_owned().await.unwrap(); - let task: tokio::task::JoinHandle> = tokio::task::spawn(async move { - let ret = if !file_data::disk::exist(&file_name).await { - file_data::disk::download(&file_account, &file_name, Some(file_size)).await - } else { - Ok(0) - }; - // In case where the parquet file is not found or has no data, we assume that it - // must have been deleted by some external entity, and hence we - // should remove the entry from file_list table. - let file_name = match ret { - Ok(data_len) => { - if data_len > 0 && data_len != file_size { - log::warn!( - "[COMPACT] download file {file_name} found size mismatch, expected: {file_size}, actual: {data_len}, will skip it", - ); - // skip this file for compact - Some(file_name) - } else { - None - } - } - Err(e) => { - if e.to_string().to_lowercase().contains("not found") - || e.to_string().to_lowercase().contains("data size is zero") - { - // delete file from file list - log::error!("[COMPACT] found invalid file: {file_name}, will delete it"); - if let Err(e) = - infra_file_list::delete_parquet_file(&file_account, &file_name, true) - .await - { - log::error!("[COMPACT] delete from file_list err: {e}"); - } - Some(file_name) - } else { - log::error!("[COMPACT] download file to cache err: {e}"); - // remove downloaded file - let _ = file_data::disk::remove(&file_name).await; - None - } - } - }; - drop(permit); - file_name - }); - tasks.push(task); - } - - let mut delete_files = Vec::new(); - for task in tasks { - match task.await { - Ok(file) => { - if let Some(file) = file { - delete_files.push(file); - } - } - Err(e) => { - log::error!("[COMPACTOR] load file task err: {e}"); - } - } - } - - Ok(delete_files) -} - -/// sort by time range without overlapping -fn sort_by_time_range(mut file_list: Vec) -> Vec { - let files_num = file_list.len(); - file_list.sort_by_key(|f| f.meta.min_ts); - let mut groups: Vec> = Vec::with_capacity(files_num); - for file in file_list { - let mut inserted = None; - for (i, group) in groups.iter().enumerate() { - if group - .last() - .is_some_and(|f| file.meta.min_ts >= f.meta.max_ts) - { - inserted = Some(i); - break; - } - } - if let Some(i) = inserted { - groups[i].push(file); - } else { - groups.push(vec![file]); - } - } - let mut files = Vec::with_capacity(files_num); - for group in groups { - files.extend(group); - } - files -} - -#[cfg(test)] -mod tests { - use config::meta::stream::{FileKey, FileMeta}; - - use super::*; - - #[test] - fn test_job_range_end_hourly_and_daily() { - // 2026-08-18 10:10:00 UTC - let offset = Utc - .with_ymd_and_hms(2026, 8, 18, 10, 10, 0) - .unwrap() - .timestamp_micros(); - assert_eq!( - job_range_end(offset, PartitionTimeLevel::Hourly), - Utc.with_ymd_and_hms(2026, 8, 18, 11, 0, 0) - .unwrap() - .timestamp_micros() - - 1 - ); - assert_eq!( - job_range_end(offset, PartitionTimeLevel::Daily), - Utc.with_ymd_and_hms(2026, 8, 19, 0, 0, 0) - .unwrap() - .timestamp_micros() - - 1 - ); - } - - // Helper function to create test FileKey - fn create_file_key(key: &str, min_ts: i64, max_ts: i64, original_size: i64) -> FileKey { - FileKey { - id: 0, - account: "test_account".to_string(), - key: key.to_string(), - meta: FileMeta { - min_ts, - max_ts, - records: 100, - original_size, - compressed_size: original_size / 2, // assume 50% compression - index_size: 0, - flattened: false, - bloom_ver: 0, - }, - deleted: false, - selection: None, - row_group_size: None, - } - } - - #[test] - fn test_sort_by_time_range_edge_case_adjacent_files() { - let files = vec![ - create_file_key("file1.parquet", 1000, 2000, 1024), - create_file_key("file2.parquet", 2000, 3000, 1024), // exactly adjacent - create_file_key("file3.parquet", 3000, 4000, 1024), // exactly adjacent - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 3); - - // Adjacent files should be able to be in the same group - assert_eq!(result[0].key, "file1.parquet"); - assert_eq!(result[1].key, "file2.parquet"); - assert_eq!(result[2].key, "file3.parquet"); - } - - // Test helper function creation - #[test] - fn test_create_file_key_helper() { - let file_key = create_file_key("test.parquet", 1000, 2000, 1024); - assert_eq!(file_key.key, "test.parquet"); - assert_eq!(file_key.meta.min_ts, 1000); - assert_eq!(file_key.meta.max_ts, 2000); - assert_eq!(file_key.meta.original_size, 1024); - assert_eq!(file_key.meta.compressed_size, 512); // 50% compression - assert_eq!(file_key.meta.records, 100); - assert!(!file_key.meta.flattened); - assert_eq!(file_key.id, 0); - assert_eq!(file_key.account, "test_account"); - assert!(!file_key.deleted); - assert!(file_key.selection.is_none()); - } - - // Boundary tests for sort_by_time_range - #[test] - fn test_sort_by_time_range_negative_timestamps() { - let files = vec![ - create_file_key("file1.parquet", -2000, -1000, 1024), - create_file_key("file2.parquet", -1000, 0, 1024), - create_file_key("file3.parquet", 0, 1000, 1024), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 3); - assert_eq!(result[0].key, "file1.parquet"); - assert_eq!(result[1].key, "file2.parquet"); - assert_eq!(result[2].key, "file3.parquet"); - } - - #[test] - fn test_sort_by_time_range_large_timestamps() { - let files = vec![ - create_file_key("file1.parquet", i64::MAX - 2000, i64::MAX - 1000, 1024), - create_file_key("file2.parquet", i64::MAX - 1000, i64::MAX, 1024), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 2); - assert_eq!(result[0].key, "file1.parquet"); - assert_eq!(result[1].key, "file2.parquet"); - } - - // Edge case where min_ts equals max_ts (point in time) - #[test] - fn test_sort_by_time_range_point_in_time() { - let files = vec![ - create_file_key("file1.parquet", 1000, 1000, 1024), // Point in time - create_file_key("file2.parquet", 1000, 2000, 1024), // Overlaps with file1 - create_file_key("file3.parquet", 2000, 2000, 1024), // Point in time, adjacent - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 3); - - // Verify all files are present - let keys: Vec<&String> = result.iter().map(|f| &f.key).collect(); - assert!(keys.contains(&&"file1.parquet".to_string())); - assert!(keys.contains(&&"file2.parquet".to_string())); - assert!(keys.contains(&&"file3.parquet".to_string())); - } - - #[test] - fn test_sort_by_time_range_many_files_random_order() { - let files = vec![ - create_file_key("file_f.parquet", 6000, 7000, 1024), - create_file_key("file_b.parquet", 2000, 3000, 1024), - create_file_key("file_d.parquet", 4000, 5000, 1024), - create_file_key("file_a.parquet", 1000, 2000, 1024), - create_file_key("file_c.parquet", 3000, 4000, 1024), - create_file_key("file_e.parquet", 5000, 6000, 1024), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 6); - - // Should be sorted by min_ts (all adjacent files) - assert_eq!(result[0].key, "file_a.parquet"); - assert_eq!(result[1].key, "file_b.parquet"); - assert_eq!(result[2].key, "file_c.parquet"); - assert_eq!(result[3].key, "file_d.parquet"); - assert_eq!(result[4].key, "file_e.parquet"); - assert_eq!(result[5].key, "file_f.parquet"); - } - - #[test] - fn test_sort_by_time_range_gaps_between_files() { - let files = vec![ - create_file_key("file1.parquet", 1000, 2000, 1024), - create_file_key("file2.parquet", 5000, 6000, 1024), // gap after file1 - create_file_key("file3.parquet", 3000, 4000, 1024), // fits in gap - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 3); - - // Should be sorted by min_ts - assert_eq!(result[0].key, "file1.parquet"); - assert_eq!(result[1].key, "file3.parquet"); - assert_eq!(result[2].key, "file2.parquet"); - } - - #[test] - fn test_sort_by_time_range_empty_list() { - let files = vec![]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 0); - } - - #[test] - fn test_sort_by_time_range_single_file() { - let files = vec![create_file_key("file1.parquet", 1000, 2000, 1024)]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 1); - assert_eq!(result[0].key, "file1.parquet"); - assert_eq!(result[0].meta.min_ts, 1000); - assert_eq!(result[0].meta.max_ts, 2000); - } - - #[test] - fn test_sort_by_time_range_already_sorted_non_overlapping() { - let files = vec![ - create_file_key("file1.parquet", 1000, 2000, 1024), - create_file_key("file2.parquet", 2000, 3000, 1024), - create_file_key("file3.parquet", 3000, 4000, 1024), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 3); - assert_eq!(result[0].key, "file1.parquet"); - assert_eq!(result[1].key, "file2.parquet"); - assert_eq!(result[2].key, "file3.parquet"); - } - - #[test] - fn test_sort_by_time_range_unsorted_non_overlapping() { - let files = vec![ - create_file_key("file3.parquet", 3000, 4000, 1024), - create_file_key("file1.parquet", 1000, 2000, 1024), - create_file_key("file2.parquet", 2000, 3000, 1024), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 3); - // Should be sorted by min_ts - assert_eq!(result[0].key, "file1.parquet"); - assert_eq!(result[1].key, "file2.parquet"); - assert_eq!(result[2].key, "file3.parquet"); - } - - #[test] - fn test_sort_by_time_range_overlapping_files() { - let files = vec![ - create_file_key("file1.parquet", 1000, 2500, 1024), // overlaps with file2 - create_file_key("file2.parquet", 2000, 3000, 1024), // overlaps with file1 - create_file_key("file3.parquet", 3500, 4000, 1024), // non-overlapping - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 3); - - // First file should be file1 (min_ts = 1000) - assert_eq!(result[0].key, "file1.parquet"); - - // Due to overlapping, file3 should come next (can fit in same group as file1) - // file2 would be in a separate group since it overlaps with file1 - let mut found_file2 = false; - let mut found_file3 = false; - for file in &result { - if file.key == "file2.parquet" { - found_file2 = true; - } - if file.key == "file3.parquet" { - found_file3 = true; - } - } - assert!(found_file2); - assert!(found_file3); - } - - #[test] - fn test_sort_by_time_range_complex_overlapping() { - let files = vec![ - create_file_key("file1.parquet", 1000, 1500, 1024), - create_file_key("file2.parquet", 1200, 1800, 1024), // overlaps with file1 - create_file_key("file3.parquet", 1600, 2000, 1024), // overlaps with file2 - create_file_key("file4.parquet", 2000, 2500, 1024), // adjacent to file3 - create_file_key("file5.parquet", 3000, 3500, 1024), // separate group - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 5); - - // Verify all files are present - let keys: Vec<&String> = result.iter().map(|f| &f.key).collect(); - assert!(keys.contains(&&"file1.parquet".to_string())); - assert!(keys.contains(&&"file2.parquet".to_string())); - assert!(keys.contains(&&"file3.parquet".to_string())); - assert!(keys.contains(&&"file4.parquet".to_string())); - assert!(keys.contains(&&"file5.parquet".to_string())); - } - - #[test] - fn test_sort_by_time_range_identical_timestamps() { - let files = vec![ - create_file_key("file1.parquet", 1000, 2000, 1024), - create_file_key("file2.parquet", 1000, 2000, 512), - create_file_key("file3.parquet", 1000, 2000, 2048), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 3); - - // All files have same timestamp, so they should all be in separate groups - // due to overlap, but ordering should be maintained based on original order after sorting - let keys: Vec<&String> = result.iter().map(|f| &f.key).collect(); - assert_eq!(keys.len(), 3); - assert!(keys.contains(&&"file1.parquet".to_string())); - assert!(keys.contains(&&"file2.parquet".to_string())); - assert!(keys.contains(&&"file3.parquet".to_string())); - } - - // ── Additional sort_by_time_range edge-case tests ──────────────────────── - - /// Two files share the same min_ts but have different max_ts values. - /// The file with the smaller max_ts ends first; the second file starts at - /// the same time and thus overlaps → they must land in different groups. - #[test] - fn test_sort_by_time_range_same_min_ts_different_max_ts() { - let files = vec![ - create_file_key("file_a.parquet", 1000, 3000, 1024), - create_file_key("file_b.parquet", 1000, 2000, 1024), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 2); - let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); - assert!(keys.contains(&"file_a.parquet")); - assert!(keys.contains(&"file_b.parquet")); - // Both share min_ts=1000, so they overlap: each ends up in its own group. - // Consecutive elements in the output must not have a later file whose - // min_ts < the predecessor's max_ts within the same group-chain. - // We just verify the invariant: result preserves all files. - assert_eq!(result.len(), 2); - } - - /// All files are completely overlapping (same time range). - /// Every file goes into a separate group; total count must be preserved. - #[test] - fn test_sort_by_time_range_all_identical_ranges() { - let files = vec![ - create_file_key("file1.parquet", 500, 1500, 1024), - create_file_key("file2.parquet", 500, 1500, 512), - create_file_key("file3.parquet", 500, 1500, 2048), - create_file_key("file4.parquet", 500, 1500, 768), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 4); - let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); - assert!(keys.contains(&"file1.parquet")); - assert!(keys.contains(&"file2.parquet")); - assert!(keys.contains(&"file3.parquet")); - assert!(keys.contains(&"file4.parquet")); - } - - /// Files whose min_ts == max_ts == 0 (zero timestamps). - #[test] - fn test_sort_by_time_range_zero_timestamps() { - let files = vec![ - create_file_key("file1.parquet", 0, 0, 1024), - create_file_key("file2.parquet", 0, 1000, 1024), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 2); - // First file is whichever has min_ts=0; both should survive. - let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); - assert!(keys.contains(&"file1.parquet")); - assert!(keys.contains(&"file2.parquet")); - } - - /// Two non-overlapping groups: group A has files [1000,2000] and [2000,3000]; - /// group B has files [1500,2500] and [2500,3500]. The algorithm must - /// produce exactly 4 files in the output. - #[test] - fn test_sort_by_time_range_two_interleaved_groups() { - // After sort by min_ts: - // A1: [1000,2000], B1: [1500,2500], A2: [2000,3000], B2: [2500,3500] - // A1 opens group-0. B1 overlaps A1 (1500 < 2000) → opens group-1. - // A2: min_ts=2000 >= group-0's last max=2000 → fits group-0. - // B2: min_ts=2500 >= group-1's last max=2500 → fits group-1. - let files = vec![ - create_file_key("A1.parquet", 1000, 2000, 1024), - create_file_key("B1.parquet", 1500, 2500, 1024), - create_file_key("A2.parquet", 2000, 3000, 1024), - create_file_key("B2.parquet", 2500, 3500, 1024), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 4); - let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); - assert!(keys.contains(&"A1.parquet")); - assert!(keys.contains(&"A2.parquet")); - assert!(keys.contains(&"B1.parquet")); - assert!(keys.contains(&"B2.parquet")); - // Group-0 chain: A1 then A2 (adjacent). Verify they appear consecutively. - let a1_pos = result.iter().position(|f| f.key == "A1.parquet").unwrap(); - let a2_pos = result.iter().position(|f| f.key == "A2.parquet").unwrap(); - assert_eq!( - a2_pos, - a1_pos + 1, - "A1 and A2 should be in the same group (consecutive)" - ); - } - - /// A chain of files where each one's min_ts equals the previous file's max_ts - /// (strictly adjacent, no gap). All should end up in one group. - #[test] - fn test_sort_by_time_range_strictly_adjacent_chain() { - let files = vec![ - create_file_key("f1.parquet", 100, 200, 512), - create_file_key("f3.parquet", 300, 400, 512), - create_file_key("f2.parquet", 200, 300, 512), - create_file_key("f4.parquet", 400, 500, 512), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 4); - // All are adjacent (min_ts == prev max_ts) so they chain into one group. - assert_eq!(result[0].key, "f1.parquet"); - assert_eq!(result[1].key, "f2.parquet"); - assert_eq!(result[2].key, "f3.parquet"); - assert_eq!(result[3].key, "f4.parquet"); - } - - /// A single file where min_ts == max_ts (point-in-time, zero-width range). - #[test] - fn test_sort_by_time_range_single_point_file() { - let files = vec![create_file_key("point.parquet", 42, 42, 256)]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 1); - assert_eq!(result[0].key, "point.parquet"); - assert_eq!(result[0].meta.min_ts, 42); - assert_eq!(result[0].meta.max_ts, 42); - } - - // ── sort_by_time_range: additional branch-coverage tests ───────────────── - - /// Two files that don't overlap and have a gap between them. The second - /// file's min_ts is strictly greater than the first file's max_ts, so both - /// land in the same chain (group-0) and appear in order. - #[test] - fn test_sort_by_time_range_two_non_overlapping_with_gap() { - let files = vec![ - create_file_key("early.parquet", 1000, 2000, 512), - create_file_key("late.parquet", 5000, 6000, 512), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 2); - assert_eq!(result[0].key, "early.parquet"); - assert_eq!(result[1].key, "late.parquet"); - } - - /// Files presented in strictly descending min_ts order. After the internal - /// sort by min_ts the algorithm must produce the same output as if they had - /// been given in ascending order. - #[test] - fn test_sort_by_time_range_descending_input_order() { - let files = vec![ - create_file_key("f4.parquet", 4000, 5000, 512), - create_file_key("f3.parquet", 3000, 4000, 512), - create_file_key("f2.parquet", 2000, 3000, 512), - create_file_key("f1.parquet", 1000, 2000, 512), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 4); - // Adjacent chain: all files should appear sorted by min_ts. - assert_eq!(result[0].key, "f1.parquet"); - assert_eq!(result[1].key, "f2.parquet"); - assert_eq!(result[2].key, "f3.parquet"); - assert_eq!(result[3].key, "f4.parquet"); - } - - /// File where min_ts == max_ts == i64::MIN (boundary value). - #[test] - fn test_sort_by_time_range_min_i64_timestamps() { - let files = vec![ - create_file_key("a.parquet", i64::MIN, i64::MIN, 256), - create_file_key("b.parquet", i64::MIN, 0, 256), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 2); - let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); - assert!(keys.contains(&"a.parquet")); - assert!(keys.contains(&"b.parquet")); - } - - /// A file whose min_ts is greater than its max_ts (malformed / inverted range). - /// The algorithm must still return all files without panicking. - #[test] - fn test_sort_by_time_range_inverted_range_no_panic() { - let files = vec![ - create_file_key("good.parquet", 1000, 3000, 512), - create_file_key("bad.parquet", 5000, 2000, 512), // inverted: min > max - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 2); - let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); - assert!(keys.contains(&"good.parquet")); - assert!(keys.contains(&"bad.parquet")); - } - - /// Many fully-overlapping files create as many groups as there are files. - /// Verify that the total count is always preserved regardless of input size. - #[test] - fn test_sort_by_time_range_ten_fully_overlapping_files() { - let files: Vec = (0..10) - .map(|i| create_file_key(&format!("f{i}.parquet"), 0, 1000, 256)) - .collect(); - let result = sort_by_time_range(files); - assert_eq!(result.len(), 10, "All files must be preserved"); - } - - /// Non-overlapping files interleaved with overlapping ones. - /// Ensures multiple groups can be built simultaneously and - /// files correctly land in the first group that can accept them. - #[test] - fn test_sort_by_time_range_mixed_overlap_and_gaps() { - // After sort by min_ts: - // A: [100, 200], B: [150, 300], C: [200, 400], D: [300, 500], E: [600, 700] - // Group-0 starts with A. - // B (min=150 < A.max=200) → new group-1. - // C (min=200 >= A.max=200) → fits group-0 (A chain extends to max=400). - // D (min=300 < C.max=400) in group-0 fails; min=300 >= B.max=300 → fits group-1. - // E (min=600 >= C.max=400) → fits group-0. - let files = vec![ - create_file_key("A.parquet", 100, 200, 512), - create_file_key("B.parquet", 150, 300, 512), - create_file_key("C.parquet", 200, 400, 512), - create_file_key("D.parquet", 300, 500, 512), - create_file_key("E.parquet", 600, 700, 512), - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 5); - let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); - assert!(keys.contains(&"A.parquet")); - assert!(keys.contains(&"B.parquet")); - assert!(keys.contains(&"C.parquet")); - assert!(keys.contains(&"D.parquet")); - assert!(keys.contains(&"E.parquet")); - } - - /// Verify `sort_by_time_range` with exactly two files where the second - /// file's min_ts equals the first file's max_ts — the boundary condition - /// `min_ts >= max_ts` in the group-fit predicate. - #[test] - fn test_sort_by_time_range_exact_boundary_two_files() { - let files = vec![ - create_file_key("first.parquet", 1000, 2000, 512), - create_file_key("second.parquet", 2000, 3000, 512), // min == prev max - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 2); - // min_ts(second) == max_ts(first), so `file.meta.min_ts >= f.meta.max_ts` - // evaluates to true → second file joins first file's group. - assert_eq!(result[0].key, "first.parquet"); - assert_eq!(result[1].key, "second.parquet"); - } - - /// A file whose min_ts is one less than the previous group's max_ts must - /// NOT join that group (it overlaps by 1 microsecond). - #[test] - fn test_sort_by_time_range_one_unit_overlap() { - let files = vec![ - create_file_key("first.parquet", 1000, 2000, 512), - create_file_key("overlap.parquet", 1999, 3000, 512), // 1999 < 2000 - ]; - let result = sort_by_time_range(files); - assert_eq!(result.len(), 2); - // overlap.parquet must be in a different group from first.parquet. - // The first element should be first.parquet (lower min_ts). - assert_eq!(result[0].key, "first.parquet"); - // Both files must appear; overlap.parquet goes to a new group. - let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); - assert!(keys.contains(&"overlap.parquet")); - } -} +pub use file::merge_files; +pub use job::{ + generate_downsampling_job_by_stream_and_rule, generate_job_by_stream, + generate_old_data_job_by_stream, +}; +pub use stream::merge_by_stream; diff --git a/src/compaction/src/merge/stream.rs b/src/compaction/src/merge/stream.rs new file mode 100644 index 0000000000..e81cfae55c --- /dev/null +++ b/src/compaction/src/merge/stream.rs @@ -0,0 +1,1012 @@ +// 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 ::datafusion::arrow::datatypes::Schema; +use chrono::{DateTime, TimeZone, Utc}; +use config::{ + get_config, + meta::{ + promql::tsid_layout::MetricsFileLayout, + stream::{FileKey, FileListDeleted, MergeStrategy, PartitionTimeLevel, StreamType}, + }, + metrics, + utils::time::hour_micros, +}; +use hashbrown::{HashMap, HashSet}; +use infra::{file_list as infra_file_list, schema::get_partition_time_level}; +use search::datafusion::merge::MergeMode; +use search_service::file_list; +use tokio::{ + sync::{Semaphore, mpsc}, + task::JoinHandle, +}; + +use super::{ + job::job_range_end, + tsid_major::{TsidMajorMergeScope, tsid_major_merge_scope}, +}; +use crate::worker::{MergeBatch, MergeSender}; + +/// compactor run steps on a stream: +/// 3. get a cluster lock for compactor stream +/// 4. read last compacted offset: year/month/day/hour +/// 5. read current hour all files +/// 6. compact small files to big files -> COMPACTOR_MAX_FILE_SIZE +/// 7. write to storage +/// 8. delete small files keys & write big files keys, use transaction +/// 9. delete small files from storage +/// 10. update last compacted offset +/// 11. release cluster lock +pub async fn merge_by_stream( + worker_tx: mpsc::Sender<(MergeSender, MergeBatch)>, + org_id: &str, + stream_type: StreamType, + stream_name: &str, + job_id: i64, + offset: i64, +) -> Result<(), anyhow::Error> { + let cfg = get_config(); + let start = std::time::Instant::now(); + + // get schema + let schema = infra::schema::get(org_id, stream_name, stream_type).await?; + if schema == Schema::empty() { + // the stream was deleted, mark the job as done + if let Err(e) = infra_file_list::set_job_done(&[job_id]).await { + log::error!("[COMPACTOR] set_job_done failed: {e}"); + } + return Ok(()); + } + + log::debug!( + "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] offset: {offset}" + ); + + // A job whose offset hour has not yet settled (hour end + 3 * max_file_retention_time, + // see is_past_hour) is an incremental round on a hour that can still receive files + // (enqueued by the ingester, see service::compaction::incremental): only seal full-size + // groups and carry the remainder, so each file is merged into a sealed output exactly + // once. The scheduled hour-end pass seals whatever is left. + let offset = offset - offset % hour_micros(1); + let is_incremental = !crate::is_past_hour(offset); + + // check offset + let partition_time_level = get_partition_time_level(stream_type); + let offset_time: DateTime = Utc.timestamp_nanos(offset * 1000); + let (date_start, date_end) = if partition_time_level == PartitionTimeLevel::Daily { + ( + offset_time.format("%Y/%m/%d/00").to_string(), + offset_time.format("%Y/%m/%d/23").to_string(), + ) + } else { + ( + offset_time.format("%Y/%m/%d/%H").to_string(), + offset_time.format("%Y/%m/%d/%H").to_string(), + ) + }; + // What this hour of files turns into. Decided once here — the workers and + // 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, + &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() { + i64::MAX + } else { + infra_file_list::merge_max_original_size() + }; + let files = file_list::query_for_merge( + org_id, + stream_type, + stream_name, + &date_start, + &date_end, + max_original_size, + ) + .await + .map_err(|e| anyhow::anyhow!("query file list failed: {e}"))?; + + log::debug!( + "[COMPACTOR] merge_by_stream [{org_id}/{stream_type}/{stream_name}] date range: [{date_start},{date_end}], files: {}", + files.len(), + ); + + if files.is_empty() { + // update job status + if let Err(e) = infra_file_list::set_job_done(&[job_id]).await { + log::error!("[COMPACTOR] set_job_done failed: {e}"); + } + return Ok(()); + } + + // do partition by partition key + let mut partition_files_with_size: HashMap> = HashMap::default(); + for file in files { + let file_name = file.key.clone(); + let prefix = file_name[..file_name.rfind('/').unwrap()].to_string(); + let partition = partition_files_with_size.entry(prefix).or_default(); + partition.push(file.to_owned()); + } + + // use multiple threads to merge + let semaphore = std::sync::Arc::new(Semaphore::new(cfg.limit.file_merge_thread_num)); + let mut tasks = Vec::with_capacity(partition_files_with_size.len()); + for (prefix, mut files_with_size) in partition_files_with_size.into_iter() { + let org_id = org_id.to_string(); + let stream_name = stream_name.to_string(); + let mode = mode.clone(); + let permit = semaphore.clone().acquire_owned().await.unwrap(); + let worker_tx = worker_tx.clone(); + let task: JoinHandle, anyhow::Error>> = tokio::task::spawn(async move { + let cfg = get_config(); + // sort by file size + let job_strategy = MergeStrategy::from(&cfg.compact.strategy); + match job_strategy { + MergeStrategy::FileSize => { + files_with_size.sort_by_key(|k| k.meta.original_size); + } + MergeStrategy::FileTime => { + files_with_size.sort_by_key(|k| k.meta.min_ts); + } + MergeStrategy::TimeRange => { + files_with_size = sort_by_time_range(files_with_size); + } + } + + 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(); + if mode.merges_whole_batch() { + batch_groups.push(MergeBatch { + batch_id: 0, + org_id: org_id.clone(), + stream_type, + stream_name: stream_name.clone(), + prefix: prefix.clone(), + files: files_with_size.clone(), + mode: mode.clone(), + }); + } else { + let mut new_file_list = Vec::new(); + let mut new_file_size = 0; + for file in files_with_size.iter() { + if new_file_size + file.meta.original_size > cfg.compact.max_file_size as i64 + || (cfg.compact.max_group_files > 0 + && new_file_list.len() >= cfg.compact.max_group_files) + { + if new_file_list.len() <= 1 { + if job_strategy == MergeStrategy::FileSize { + break; + } + new_file_list.clear(); + new_file_size = file.meta.original_size; + new_file_list.push(file.clone()); + continue; // replace previous file with current file + } + batch_groups.push(MergeBatch { + batch_id: batch_groups.len(), + org_id: org_id.clone(), + stream_type, + stream_name: stream_name.clone(), + prefix: prefix.clone(), + files: new_file_list.clone(), + mode: mode.clone(), + }); + new_file_size = 0; + new_file_list.clear(); + } + new_file_size += file.meta.original_size; + new_file_list.push(file.clone()); + } + // The trailing batch is always below max_file_size (the loop flushes a group + // only when adding the next file would exceed it). In incremental mode we do + // NOT seal this remainder: more files will arrive in the still-open hour, and + // sealing now would force re-merging it later (write amplification). Carry it + // to the next round; the scheduled hour-end pass seals whatever is left. + if new_file_list.len() > 1 && !is_incremental { + batch_groups.push(MergeBatch { + batch_id: batch_groups.len(), + org_id: org_id.clone(), + stream_type, + stream_name: stream_name.clone(), + prefix: prefix.clone(), + files: new_file_list.clone(), + mode: mode.clone(), + }); + } + + if batch_groups.is_empty() { + return Ok(vec![]); // no files need to merge + } + } + + // send to worker + let batch_group_len = batch_groups.len(); + let (inner_tx, mut inner_rx) = mpsc::channel(batch_group_len); + for batch in batch_groups.iter() { + if let Err(e) = worker_tx.send((inner_tx.clone(), batch.clone())).await { + log::error!("[COMPACTOR] send batch to worker failed: {e}"); + return Err(anyhow::Error::msg("send batch to worker failed")); + } + } + let mut worker_results = Vec::with_capacity(batch_group_len); + for _ in 0..batch_group_len { + let result = inner_rx.recv().await.unwrap(); + worker_results.push(result); + } + + let mut last_error = None; + let mut check_guard = HashSet::with_capacity(batch_groups.len()); + let mut orphan_blooms = Vec::new(); + for ret in worker_results { + let (batch_id, new_files) = match ret { + Ok(v) => v, + Err(e) => { + log::error!("[COMPACTOR] merge files failed: {e}"); + last_error = Some(e); + continue; + } + }; + + if check_guard.contains(&batch_id) { + log::warn!( + "[COMPACTOR] merge files for stream: [{org_id}/{stream_type}/{stream_name}] found error files, batch_id: {batch_id} duplicate" + ); + continue; + } + check_guard.insert(batch_id); + + // delete small files keys & write big files keys, use transaction + let delete_file_list = batch_groups.get(batch_id).unwrap().files.as_slice(); + let mut events = Vec::with_capacity(new_files.len() + delete_file_list.len()); + for new_file in new_files { + if !new_file.key.is_empty() { + events.push(new_file); + } + } + + for file in delete_file_list { + events.push(FileKey { + deleted: true, + selection: None, + row_group_size: None, + ..file.clone() + }); + } + events.sort_by(|a, b| a.key.cmp(&b.key)); + + // write file list to storage + if let Err(e) = write_file_list(&org_id, stream_type, &events).await { + log::error!("[COMPACTOR] write file list failed: {e}"); + tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + continue; + } + + // collect orphan blooms after writing file list successfully + for file in delete_file_list { + if file.meta.bloom_ver > 0 { + orphan_blooms.push(file.meta.bloom_ver); + } + } + } + drop(permit); + if let Some(e) = last_error { + return Err(e); + } + Ok(orphan_blooms) + }); + tasks.push(task); + } + + // collect bloom files which need to be clean + let mut orphan_blooms = Vec::new(); + for task in tasks { + orphan_blooms.extend(task.await??); + } + + // Build bloom filters for the current hour. Failures are non-fatal: files + // left at bloom_ver = 0 fall back to the regular Tantivy search path. + let build_start = std::time::Instant::now(); + match crate::bloom::compact::build_for_stream( + org_id, + stream_type, + stream_name, + &date_start, + is_incremental, + orphan_blooms, + ) + .await + { + Ok(false) => {} + Ok(true) => { + let build_time = build_start.elapsed().as_millis(); + log::info!( + "[COMPACTOR] bloom build for {org_id}/{stream_type}/{stream_name}/{date_start} took: {build_time} ms" + ); + } + Err(e) => { + log::warn!( + "[COMPACTOR] bloom build for {org_id}/{stream_type}/{stream_name}/{date_start} failed: {e}" + ); + } + } + + // update job status + if let Err(e) = infra_file_list::set_job_done(&[job_id]).await { + log::error!("[COMPACTOR] set_job_done failed: {e}"); + } + + // metrics + let time = start.elapsed().as_secs_f64(); + metrics::COMPACT_USED_TIME + .with_label_values(&[org_id, stream_type.as_str()]) + .inc_by(time); + + Ok(()) +} + +async fn write_file_list( + org_id: &str, + stream_type: StreamType, + events: &[FileKey], +) -> Result<(), anyhow::Error> { + if events.is_empty() { + return Ok(()); + } + + let del_items = events + .iter() + .filter(|v| v.deleted) + .map(|v| FileListDeleted { + id: 0, + account: v.account.clone(), + file: v.key.clone(), + index_file: v.meta.index_size > 0, + flattened: v.meta.flattened, + }) + .collect::>(); + + // set to db + // retry 5 times + let cfg = get_config(); + let mut success = false; + let mut mark_deleted_done = false; + let created_at = config::utils::time::now_micros(); + for _ in 0..5 { + if !mark_deleted_done && let Err(e) = infra::file_list::batch_process(events).await { + log::error!("[COMPACTOR] batch_process to db failed, retrying: {e}"); + tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + continue; + } + mark_deleted_done = true; + if !del_items.is_empty() + && let Err(e) = infra_file_list::batch_add_deleted(org_id, created_at, &del_items).await + { + log::error!("[COMPACTOR] batch_add_deleted to db failed, retrying: {e}"); + tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + continue; + } + success = true; + break; + } + + // handle dump_stats for file_list type streams + if success && stream_type == StreamType::Filelist && cfg.compact.file_list_dump_enabled { + let (deleted_files, new_files): (Vec<_>, Vec<_>) = events.iter().partition(|e| e.deleted); + crate::dump::handle_dump_stats_on_merge(&deleted_files, &new_files).await; + } + + if success { + // send broadcast to other nodes + if cfg.cache_latest_files.enabled { + // get id for all the new files + let file_ids = infra_file_list::query_ids_by_files(events).await?; + let mut events = events.to_vec(); + for event in events.iter_mut() { + if let Some(id) = file_ids.get(&event.key) { + event.id = *id; + } + } + if let Err(e) = db::file_list::broadcast::send(&events).await { + log::error!("[COMPACTOR] send broadcast for file_list failed: {e}"); + } + } + } else { + return Err(anyhow::anyhow!("batch_write to db failed")); + } + + Ok(()) +} + +/// sort by time range without overlapping +fn sort_by_time_range(mut file_list: Vec) -> Vec { + let files_num = file_list.len(); + file_list.sort_by_key(|f| f.meta.min_ts); + let mut groups: Vec> = Vec::with_capacity(files_num); + for file in file_list { + let mut inserted = None; + for (i, group) in groups.iter().enumerate() { + if group + .last() + .is_some_and(|f| file.meta.min_ts >= f.meta.max_ts) + { + inserted = Some(i); + break; + } + } + if let Some(i) = inserted { + groups[i].push(file); + } else { + groups.push(vec![file]); + } + } + let mut files = Vec::with_capacity(files_num); + for group in groups { + files.extend(group); + } + files +} + +#[cfg(test)] +mod tests { + use config::meta::stream::{FileKey, FileMeta}; + + use super::*; + + // Helper function to create test FileKey + fn create_file_key(key: &str, min_ts: i64, max_ts: i64, original_size: i64) -> FileKey { + FileKey { + id: 0, + account: "test_account".to_string(), + key: key.to_string(), + meta: FileMeta { + min_ts, + max_ts, + records: 100, + original_size, + compressed_size: original_size / 2, // assume 50% compression + index_size: 0, + flattened: false, + bloom_ver: 0, + }, + deleted: false, + selection: None, + row_group_size: None, + } + } + + #[test] + fn test_sort_by_time_range_edge_case_adjacent_files() { + let files = vec![ + create_file_key("file1.parquet", 1000, 2000, 1024), + create_file_key("file2.parquet", 2000, 3000, 1024), // exactly adjacent + create_file_key("file3.parquet", 3000, 4000, 1024), // exactly adjacent + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 3); + + // Adjacent files should be able to be in the same group + assert_eq!(result[0].key, "file1.parquet"); + assert_eq!(result[1].key, "file2.parquet"); + assert_eq!(result[2].key, "file3.parquet"); + } + + // Test helper function creation + #[test] + fn test_create_file_key_helper() { + let file_key = create_file_key("test.parquet", 1000, 2000, 1024); + assert_eq!(file_key.key, "test.parquet"); + assert_eq!(file_key.meta.min_ts, 1000); + assert_eq!(file_key.meta.max_ts, 2000); + assert_eq!(file_key.meta.original_size, 1024); + assert_eq!(file_key.meta.compressed_size, 512); // 50% compression + assert_eq!(file_key.meta.records, 100); + assert!(!file_key.meta.flattened); + assert_eq!(file_key.id, 0); + assert_eq!(file_key.account, "test_account"); + assert!(!file_key.deleted); + assert!(file_key.selection.is_none()); + } + + // Boundary tests for sort_by_time_range + #[test] + fn test_sort_by_time_range_negative_timestamps() { + let files = vec![ + create_file_key("file1.parquet", -2000, -1000, 1024), + create_file_key("file2.parquet", -1000, 0, 1024), + create_file_key("file3.parquet", 0, 1000, 1024), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 3); + assert_eq!(result[0].key, "file1.parquet"); + assert_eq!(result[1].key, "file2.parquet"); + assert_eq!(result[2].key, "file3.parquet"); + } + + #[test] + fn test_sort_by_time_range_large_timestamps() { + let files = vec![ + create_file_key("file1.parquet", i64::MAX - 2000, i64::MAX - 1000, 1024), + create_file_key("file2.parquet", i64::MAX - 1000, i64::MAX, 1024), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 2); + assert_eq!(result[0].key, "file1.parquet"); + assert_eq!(result[1].key, "file2.parquet"); + } + + // Edge case where min_ts equals max_ts (point in time) + #[test] + fn test_sort_by_time_range_point_in_time() { + let files = vec![ + create_file_key("file1.parquet", 1000, 1000, 1024), // Point in time + create_file_key("file2.parquet", 1000, 2000, 1024), // Overlaps with file1 + create_file_key("file3.parquet", 2000, 2000, 1024), // Point in time, adjacent + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 3); + + // Verify all files are present + let keys: Vec<&String> = result.iter().map(|f| &f.key).collect(); + assert!(keys.contains(&&"file1.parquet".to_string())); + assert!(keys.contains(&&"file2.parquet".to_string())); + assert!(keys.contains(&&"file3.parquet".to_string())); + } + + #[test] + fn test_sort_by_time_range_many_files_random_order() { + let files = vec![ + create_file_key("file_f.parquet", 6000, 7000, 1024), + create_file_key("file_b.parquet", 2000, 3000, 1024), + create_file_key("file_d.parquet", 4000, 5000, 1024), + create_file_key("file_a.parquet", 1000, 2000, 1024), + create_file_key("file_c.parquet", 3000, 4000, 1024), + create_file_key("file_e.parquet", 5000, 6000, 1024), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 6); + + // Should be sorted by min_ts (all adjacent files) + assert_eq!(result[0].key, "file_a.parquet"); + assert_eq!(result[1].key, "file_b.parquet"); + assert_eq!(result[2].key, "file_c.parquet"); + assert_eq!(result[3].key, "file_d.parquet"); + assert_eq!(result[4].key, "file_e.parquet"); + assert_eq!(result[5].key, "file_f.parquet"); + } + + #[test] + fn test_sort_by_time_range_gaps_between_files() { + let files = vec![ + create_file_key("file1.parquet", 1000, 2000, 1024), + create_file_key("file2.parquet", 5000, 6000, 1024), // gap after file1 + create_file_key("file3.parquet", 3000, 4000, 1024), // fits in gap + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 3); + + // Should be sorted by min_ts + assert_eq!(result[0].key, "file1.parquet"); + assert_eq!(result[1].key, "file3.parquet"); + assert_eq!(result[2].key, "file2.parquet"); + } + + #[test] + fn test_sort_by_time_range_empty_list() { + let files = vec![]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 0); + } + + #[test] + fn test_sort_by_time_range_single_file() { + let files = vec![create_file_key("file1.parquet", 1000, 2000, 1024)]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 1); + assert_eq!(result[0].key, "file1.parquet"); + assert_eq!(result[0].meta.min_ts, 1000); + assert_eq!(result[0].meta.max_ts, 2000); + } + + #[test] + fn test_sort_by_time_range_already_sorted_non_overlapping() { + let files = vec![ + create_file_key("file1.parquet", 1000, 2000, 1024), + create_file_key("file2.parquet", 2000, 3000, 1024), + create_file_key("file3.parquet", 3000, 4000, 1024), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 3); + assert_eq!(result[0].key, "file1.parquet"); + assert_eq!(result[1].key, "file2.parquet"); + assert_eq!(result[2].key, "file3.parquet"); + } + + #[test] + fn test_sort_by_time_range_unsorted_non_overlapping() { + let files = vec![ + create_file_key("file3.parquet", 3000, 4000, 1024), + create_file_key("file1.parquet", 1000, 2000, 1024), + create_file_key("file2.parquet", 2000, 3000, 1024), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 3); + // Should be sorted by min_ts + assert_eq!(result[0].key, "file1.parquet"); + assert_eq!(result[1].key, "file2.parquet"); + assert_eq!(result[2].key, "file3.parquet"); + } + + #[test] + fn test_sort_by_time_range_overlapping_files() { + let files = vec![ + create_file_key("file1.parquet", 1000, 2500, 1024), // overlaps with file2 + create_file_key("file2.parquet", 2000, 3000, 1024), // overlaps with file1 + create_file_key("file3.parquet", 3500, 4000, 1024), // non-overlapping + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 3); + + // First file should be file1 (min_ts = 1000) + assert_eq!(result[0].key, "file1.parquet"); + + // Due to overlapping, file3 should come next (can fit in same group as file1) + // file2 would be in a separate group since it overlaps with file1 + let mut found_file2 = false; + let mut found_file3 = false; + for file in &result { + if file.key == "file2.parquet" { + found_file2 = true; + } + if file.key == "file3.parquet" { + found_file3 = true; + } + } + assert!(found_file2); + assert!(found_file3); + } + + #[test] + fn test_sort_by_time_range_complex_overlapping() { + let files = vec![ + create_file_key("file1.parquet", 1000, 1500, 1024), + create_file_key("file2.parquet", 1200, 1800, 1024), // overlaps with file1 + create_file_key("file3.parquet", 1600, 2000, 1024), // overlaps with file2 + create_file_key("file4.parquet", 2000, 2500, 1024), // adjacent to file3 + create_file_key("file5.parquet", 3000, 3500, 1024), // separate group + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 5); + + // Verify all files are present + let keys: Vec<&String> = result.iter().map(|f| &f.key).collect(); + assert!(keys.contains(&&"file1.parquet".to_string())); + assert!(keys.contains(&&"file2.parquet".to_string())); + assert!(keys.contains(&&"file3.parquet".to_string())); + assert!(keys.contains(&&"file4.parquet".to_string())); + assert!(keys.contains(&&"file5.parquet".to_string())); + } + + #[test] + fn test_sort_by_time_range_identical_timestamps() { + let files = vec![ + create_file_key("file1.parquet", 1000, 2000, 1024), + create_file_key("file2.parquet", 1000, 2000, 512), + create_file_key("file3.parquet", 1000, 2000, 2048), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 3); + + // All files have same timestamp, so they should all be in separate groups + // due to overlap, but ordering should be maintained based on original order after sorting + let keys: Vec<&String> = result.iter().map(|f| &f.key).collect(); + assert_eq!(keys.len(), 3); + assert!(keys.contains(&&"file1.parquet".to_string())); + assert!(keys.contains(&&"file2.parquet".to_string())); + assert!(keys.contains(&&"file3.parquet".to_string())); + } + + // ── Additional sort_by_time_range edge-case tests ──────────────────────── + + /// Two files share the same min_ts but have different max_ts values. + /// The file with the smaller max_ts ends first; the second file starts at + /// the same time and thus overlaps → they must land in different groups. + #[test] + fn test_sort_by_time_range_same_min_ts_different_max_ts() { + let files = vec![ + create_file_key("file_a.parquet", 1000, 3000, 1024), + create_file_key("file_b.parquet", 1000, 2000, 1024), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 2); + let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); + assert!(keys.contains(&"file_a.parquet")); + assert!(keys.contains(&"file_b.parquet")); + // Both share min_ts=1000, so they overlap: each ends up in its own group. + // Consecutive elements in the output must not have a later file whose + // min_ts < the predecessor's max_ts within the same group-chain. + // We just verify the invariant: result preserves all files. + assert_eq!(result.len(), 2); + } + + /// All files are completely overlapping (same time range). + /// Every file goes into a separate group; total count must be preserved. + #[test] + fn test_sort_by_time_range_all_identical_ranges() { + let files = vec![ + create_file_key("file1.parquet", 500, 1500, 1024), + create_file_key("file2.parquet", 500, 1500, 512), + create_file_key("file3.parquet", 500, 1500, 2048), + create_file_key("file4.parquet", 500, 1500, 768), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 4); + let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); + assert!(keys.contains(&"file1.parquet")); + assert!(keys.contains(&"file2.parquet")); + assert!(keys.contains(&"file3.parquet")); + assert!(keys.contains(&"file4.parquet")); + } + + /// Files whose min_ts == max_ts == 0 (zero timestamps). + #[test] + fn test_sort_by_time_range_zero_timestamps() { + let files = vec![ + create_file_key("file1.parquet", 0, 0, 1024), + create_file_key("file2.parquet", 0, 1000, 1024), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 2); + // First file is whichever has min_ts=0; both should survive. + let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); + assert!(keys.contains(&"file1.parquet")); + assert!(keys.contains(&"file2.parquet")); + } + + /// Two non-overlapping groups: group A has files [1000,2000] and [2000,3000]; + /// group B has files [1500,2500] and [2500,3500]. The algorithm must + /// produce exactly 4 files in the output. + #[test] + fn test_sort_by_time_range_two_interleaved_groups() { + // After sort by min_ts: + // A1: [1000,2000], B1: [1500,2500], A2: [2000,3000], B2: [2500,3500] + // A1 opens group-0. B1 overlaps A1 (1500 < 2000) → opens group-1. + // A2: min_ts=2000 >= group-0's last max=2000 → fits group-0. + // B2: min_ts=2500 >= group-1's last max=2500 → fits group-1. + let files = vec![ + create_file_key("A1.parquet", 1000, 2000, 1024), + create_file_key("B1.parquet", 1500, 2500, 1024), + create_file_key("A2.parquet", 2000, 3000, 1024), + create_file_key("B2.parquet", 2500, 3500, 1024), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 4); + let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); + assert!(keys.contains(&"A1.parquet")); + assert!(keys.contains(&"A2.parquet")); + assert!(keys.contains(&"B1.parquet")); + assert!(keys.contains(&"B2.parquet")); + // Group-0 chain: A1 then A2 (adjacent). Verify they appear consecutively. + let a1_pos = result.iter().position(|f| f.key == "A1.parquet").unwrap(); + let a2_pos = result.iter().position(|f| f.key == "A2.parquet").unwrap(); + assert_eq!( + a2_pos, + a1_pos + 1, + "A1 and A2 should be in the same group (consecutive)" + ); + } + + /// A chain of files where each one's min_ts equals the previous file's max_ts + /// (strictly adjacent, no gap). All should end up in one group. + #[test] + fn test_sort_by_time_range_strictly_adjacent_chain() { + let files = vec![ + create_file_key("f1.parquet", 100, 200, 512), + create_file_key("f3.parquet", 300, 400, 512), + create_file_key("f2.parquet", 200, 300, 512), + create_file_key("f4.parquet", 400, 500, 512), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 4); + // All are adjacent (min_ts == prev max_ts) so they chain into one group. + assert_eq!(result[0].key, "f1.parquet"); + assert_eq!(result[1].key, "f2.parquet"); + assert_eq!(result[2].key, "f3.parquet"); + assert_eq!(result[3].key, "f4.parquet"); + } + + /// A single file where min_ts == max_ts (point-in-time, zero-width range). + #[test] + fn test_sort_by_time_range_single_point_file() { + let files = vec![create_file_key("point.parquet", 42, 42, 256)]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 1); + assert_eq!(result[0].key, "point.parquet"); + assert_eq!(result[0].meta.min_ts, 42); + assert_eq!(result[0].meta.max_ts, 42); + } + + // ── sort_by_time_range: additional branch-coverage tests ───────────────── + + /// Two files that don't overlap and have a gap between them. The second + /// file's min_ts is strictly greater than the first file's max_ts, so both + /// land in the same chain (group-0) and appear in order. + #[test] + fn test_sort_by_time_range_two_non_overlapping_with_gap() { + let files = vec![ + create_file_key("early.parquet", 1000, 2000, 512), + create_file_key("late.parquet", 5000, 6000, 512), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 2); + assert_eq!(result[0].key, "early.parquet"); + assert_eq!(result[1].key, "late.parquet"); + } + + /// Files presented in strictly descending min_ts order. After the internal + /// sort by min_ts the algorithm must produce the same output as if they had + /// been given in ascending order. + #[test] + fn test_sort_by_time_range_descending_input_order() { + let files = vec![ + create_file_key("f4.parquet", 4000, 5000, 512), + create_file_key("f3.parquet", 3000, 4000, 512), + create_file_key("f2.parquet", 2000, 3000, 512), + create_file_key("f1.parquet", 1000, 2000, 512), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 4); + // Adjacent chain: all files should appear sorted by min_ts. + assert_eq!(result[0].key, "f1.parquet"); + assert_eq!(result[1].key, "f2.parquet"); + assert_eq!(result[2].key, "f3.parquet"); + assert_eq!(result[3].key, "f4.parquet"); + } + + /// File where min_ts == max_ts == i64::MIN (boundary value). + #[test] + fn test_sort_by_time_range_min_i64_timestamps() { + let files = vec![ + create_file_key("a.parquet", i64::MIN, i64::MIN, 256), + create_file_key("b.parquet", i64::MIN, 0, 256), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 2); + let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); + assert!(keys.contains(&"a.parquet")); + assert!(keys.contains(&"b.parquet")); + } + + /// A file whose min_ts is greater than its max_ts (malformed / inverted range). + /// The algorithm must still return all files without panicking. + #[test] + fn test_sort_by_time_range_inverted_range_no_panic() { + let files = vec![ + create_file_key("good.parquet", 1000, 3000, 512), + create_file_key("bad.parquet", 5000, 2000, 512), // inverted: min > max + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 2); + let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); + assert!(keys.contains(&"good.parquet")); + assert!(keys.contains(&"bad.parquet")); + } + + /// Many fully-overlapping files create as many groups as there are files. + /// Verify that the total count is always preserved regardless of input size. + #[test] + fn test_sort_by_time_range_ten_fully_overlapping_files() { + let files: Vec = (0..10) + .map(|i| create_file_key(&format!("f{i}.parquet"), 0, 1000, 256)) + .collect(); + let result = sort_by_time_range(files); + assert_eq!(result.len(), 10, "All files must be preserved"); + } + + /// Non-overlapping files interleaved with overlapping ones. + /// Ensures multiple groups can be built simultaneously and + /// files correctly land in the first group that can accept them. + #[test] + fn test_sort_by_time_range_mixed_overlap_and_gaps() { + // After sort by min_ts: + // A: [100, 200], B: [150, 300], C: [200, 400], D: [300, 500], E: [600, 700] + // Group-0 starts with A. + // B (min=150 < A.max=200) → new group-1. + // C (min=200 >= A.max=200) → fits group-0 (A chain extends to max=400). + // D (min=300 < C.max=400) in group-0 fails; min=300 >= B.max=300 → fits group-1. + // E (min=600 >= C.max=400) → fits group-0. + let files = vec![ + create_file_key("A.parquet", 100, 200, 512), + create_file_key("B.parquet", 150, 300, 512), + create_file_key("C.parquet", 200, 400, 512), + create_file_key("D.parquet", 300, 500, 512), + create_file_key("E.parquet", 600, 700, 512), + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 5); + let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); + assert!(keys.contains(&"A.parquet")); + assert!(keys.contains(&"B.parquet")); + assert!(keys.contains(&"C.parquet")); + assert!(keys.contains(&"D.parquet")); + assert!(keys.contains(&"E.parquet")); + } + + /// Verify `sort_by_time_range` with exactly two files where the second + /// file's min_ts equals the first file's max_ts — the boundary condition + /// `min_ts >= max_ts` in the group-fit predicate. + #[test] + fn test_sort_by_time_range_exact_boundary_two_files() { + let files = vec![ + create_file_key("first.parquet", 1000, 2000, 512), + create_file_key("second.parquet", 2000, 3000, 512), // min == prev max + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 2); + // min_ts(second) == max_ts(first), so `file.meta.min_ts >= f.meta.max_ts` + // evaluates to true → second file joins first file's group. + assert_eq!(result[0].key, "first.parquet"); + assert_eq!(result[1].key, "second.parquet"); + } + + /// A file whose min_ts is one less than the previous group's max_ts must + /// NOT join that group (it overlaps by 1 microsecond). + #[test] + fn test_sort_by_time_range_one_unit_overlap() { + let files = vec![ + create_file_key("first.parquet", 1000, 2000, 512), + create_file_key("overlap.parquet", 1999, 3000, 512), // 1999 < 2000 + ]; + let result = sort_by_time_range(files); + assert_eq!(result.len(), 2); + // overlap.parquet must be in a different group from first.parquet. + // The first element should be first.parquet (lower min_ts). + assert_eq!(result[0].key, "first.parquet"); + // Both files must appear; overlap.parquet goes to a new group. + let keys: Vec<&str> = result.iter().map(|f| f.key.as_str()).collect(); + assert!(keys.contains(&"overlap.parquet")); + } +}