diff --git a/src/cli/basic/test.rs b/src/cli/basic/test.rs index 75f72b5481..829e9a1fac 100644 --- a/src/cli/basic/test.rs +++ b/src/cli/basic/test.rs @@ -32,9 +32,14 @@ pub async fn file_list( println!("strategy: {mode}"); println!("group size: {group_size}gb"); - let file_list = - crate::service::file_list::query_by_date(org, stream_name, stream_type.into(), hour, hour) - .await?; + let file_list = crate::service::file_list::query_for_merge( + org, + stream_name, + stream_type.into(), + hour, + hour, + ) + .await?; println!("get files: {}", file_list.len()); let group_size = group_size.parse::().unwrap_or(5) * 1024 * 1024 * 1024; diff --git a/src/config/src/config.rs b/src/config/src/config.rs index e2c92cf39a..d6f433557e 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -1073,6 +1073,14 @@ pub struct Common { pub min_auto_refresh_interval: u32, #[env_config(name = "ZO_ADDITIONAL_REPORTING_ORGS", default = "")] pub additional_reporting_orgs: String, + #[env_config(name = "ZO_FILE_LIST_DUMP_ENABLED", default = false)] + pub file_list_dump_enabled: bool, + #[env_config(name = "ZO_FILE_LIST_DUMP_DUAL_WRITE", default = true)] + pub file_list_dump_dual_write: bool, + #[env_config(name = "ZO_FILE_LIST_DUMP_MIN_HOUR", default = 2)] + pub file_list_dump_min_hour: usize, + #[env_config(name = "ZO_FILE_LIST_DUMP_DEBUG_CHECK", default = true)] + pub file_list_dump_debug_check: bool, } #[derive(EnvConfig)] @@ -2143,6 +2151,11 @@ fn check_common_config(cfg: &mut Config) -> Result<(), anyhow::Error> { cfg.common.feature_join_right_side_max_rows = 50_000; } + // debug check is useful only when dual write is enabled. Otherwise it will raise error + // incorrectly each time + cfg.common.file_list_dump_debug_check = + cfg.common.file_list_dump_dual_write && cfg.common.file_list_dump_debug_check; + Ok(()) } diff --git a/src/config/src/meta/stream.rs b/src/config/src/meta/stream.rs index f275502973..9979023141 100644 --- a/src/config/src/meta/stream.rs +++ b/src/config/src/meta/stream.rs @@ -419,6 +419,27 @@ impl std::ops::Sub<&StreamStats> for &StreamStats { } } +impl std::ops::Add<&StreamStats> for &StreamStats { + type Output = StreamStats; + + fn add(self, rhs: &StreamStats) -> Self::Output { + let mut ret = StreamStats { + created_at: self.created_at, + file_num: self.file_num + rhs.file_num, + doc_num: self.doc_num + rhs.doc_num, + doc_time_min: self.doc_time_min.min(rhs.doc_time_min), + doc_time_max: self.doc_time_max.max(rhs.doc_time_max), + storage_size: self.storage_size + rhs.storage_size, + compressed_size: self.compressed_size + rhs.compressed_size, + index_size: self.index_size + rhs.index_size, + }; + if ret.doc_time_min == 0 { + ret.doc_time_min = rhs.doc_time_min; + } + ret + } +} + impl From<&FileMeta> for cluster_rpc::FileMeta { fn from(req: &FileMeta) -> Self { cluster_rpc::FileMeta { diff --git a/src/infra/src/file_list/mod.rs b/src/infra/src/file_list/mod.rs index dc6ad62d5e..266d3e030b 100644 --- a/src/infra/src/file_list/mod.rs +++ b/src/infra/src/file_list/mod.rs @@ -58,6 +58,7 @@ pub trait FileList: Sync + Send + 'static { async fn batch_add(&self, files: &[FileKey]) -> Result<()>; async fn batch_add_with_id(&self, files: &[(i64, &FileKey)]) -> Result<()>; async fn batch_add_history(&self, files: &[FileKey]) -> Result<()>; + async fn update_dump_records(&self, dump_file: &FileKey, dumped_ids: &[i64]) -> Result<()>; async fn batch_process(&self, files: &[FileKey]) -> Result<()>; async fn batch_add_deleted( &self, @@ -80,7 +81,7 @@ pub trait FileList: Sync + Send + 'static { time_range: Option<(i64, i64)>, flattened: Option, ) -> Result>; - async fn query_by_date( + async fn query_for_merge( &self, org_id: &str, stream_type: StreamType, @@ -170,6 +171,16 @@ pub trait FileList: Sync + Send + 'static { async fn update_running_jobs(&self, ids: &[i64]) -> Result<()>; async fn check_running_jobs(&self, before_date: i64) -> Result<()>; async fn clean_done_jobs(&self, before_date: i64) -> Result<()>; + async fn get_entries_in_range( + &self, + org: &str, + stream: Option<&str>, + start_time: i64, + end_time: i64, + min_id: Option, + ) -> Result>; + async fn get_pending_dump_jobs(&self) -> Result>; + async fn set_job_dumped_status(&self, id: i64, dumped: bool) -> Result<()>; } pub async fn create_table() -> Result<()> { @@ -210,6 +221,11 @@ pub async fn batch_process(files: &[FileKey]) -> Result<()> { CLIENT.batch_process(files).await } +#[inline] +pub async fn update_dump_records(dump_file: &FileKey, dumped_ids: &[i64]) -> Result<()> { + CLIENT.update_dump_records(dump_file, dumped_ids).await +} + #[inline] pub async fn batch_add_deleted( org_id: &str, @@ -273,15 +289,15 @@ pub async fn query( } #[inline] -#[tracing::instrument(name = "infra:file_list:db:query_by_date")] -pub async fn query_by_date( +#[tracing::instrument(name = "infra:file_list:db:query_for_merge")] +pub async fn query_for_merge( org_id: &str, stream_type: StreamType, stream_name: &str, date_range: Option<(String, String)>, ) -> Result> { CLIENT - .query_by_date(org_id, stream_type, stream_name, date_range) + .query_for_merge(org_id, stream_type, stream_name, date_range) .await } @@ -464,6 +480,29 @@ pub async fn clean_done_jobs(before_date: i64) -> Result<()> { CLIENT.clean_done_jobs(before_date).await } +#[inline] +pub async fn get_entries_in_range( + org: &str, + stream: Option<&str>, + start_time: i64, + end_time: i64, + min_id: Option, +) -> Result> { + CLIENT + .get_entries_in_range(org, stream, start_time, end_time, min_id) + .await +} + +#[inline] +pub async fn get_pending_dump_jobs() -> Result> { + CLIENT.get_pending_dump_jobs().await +} + +#[inline] +pub async fn set_job_dumped_status(id: i64, dumped: bool) -> Result<()> { + CLIENT.set_job_dumped_status(id, dumped).await +} + pub async fn local_cache_gc() -> Result<()> { tokio::task::spawn(async move { let cfg = config::get_config(); @@ -499,7 +538,7 @@ fn validate_time_range(time_range: Option<(i64, i64)>) -> Result<()> { Ok(()) } -fn calculate_max_ts_upper_bound(time_end: i64, stream_type: StreamType) -> i64 { +pub fn calculate_max_ts_upper_bound(time_end: i64, stream_type: StreamType) -> i64 { let ts = super::schema::unwrap_partition_time_level(None, stream_type).duration(); if ts > 0 { time_end + second_micros(ts) @@ -513,6 +552,8 @@ pub struct FileRecord { #[sqlx(default)] pub id: i64, #[sqlx(default)] + pub org: String, + #[sqlx(default)] pub stream: String, pub date: String, pub file: String, diff --git a/src/infra/src/file_list/mysql.rs b/src/infra/src/file_list/mysql.rs index 90e9e70b5a..184f8dcaad 100644 --- a/src/infra/src/file_list/mysql.rs +++ b/src/infra/src/file_list/mysql.rs @@ -105,6 +105,73 @@ impl super::FileList for MysqlFileList { self.inner_batch_process("file_list", files).await } + async fn update_dump_records(&self, dump_file: &FileKey, dumped_ids: &[i64]) -> Result<()> { + let pool = CLIENT.clone(); + let mut tx = pool.begin().await?; + + let (stream_key, date_key, file_name) = + parse_file_key_columns(&dump_file.key).map_err(|e| Error::Message(e.to_string()))?; + let org_id = stream_key[..stream_key.find('/').unwrap()].to_string(); + let meta = &dump_file.meta; + DB_QUERY_NUMS + .with_label_values(&["insert", "file_list", ""]) + .inc(); + if let Err(e) = sqlx::query(r#"INSERT IGNORE INTO file_list (org, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);"#) + .bind(org_id) + .bind(stream_key) + .bind(date_key) + .bind(file_name) + .bind(false) + .bind(meta.min_ts) + .bind(meta.max_ts) + .bind(meta.records) + .bind(meta.original_size) + .bind(meta.compressed_size) + .bind(meta.index_size) + .bind(meta.flattened) + .execute(&pool) + .await { + if let Err(e) = tx.rollback().await { + log::error!("[MYSQL] rollback file_list update file dump error: {e}",); + } + return Err(e.into()); + } + + for chunk in dumped_ids.chunks(get_config().limit.file_list_id_batch_size) { + if chunk.is_empty() { + continue; + } + let ids = chunk + .iter() + .map(|id| id.to_string()) + .collect::>() + .join(","); + let query_str = format!("DELETE FROM file_list WHERE id IN ({ids})"); + DB_QUERY_NUMS + .with_label_values(&["delete_by_ids", "file_list"]) + .inc(); + let start = std::time::Instant::now(); + let res = sqlx::query(&query_str).execute(&mut *tx).await; + let time = start.elapsed().as_secs_f64(); + DB_QUERY_TIME + .with_label_values(&["delete_by_ids", "file_list"]) + .observe(time); + if let Err(e) = res { + if let Err(e) = tx.rollback().await { + log::error!("[MYSQL] rollback file_list update file dump error: {e}",); + } + return Err(e.into()); + } + } + + if let Err(e) = tx.commit().await { + log::error!("[MYSQL] commit file_list update dump file error: {e}"); + return Err(e.into()); + } + Ok(()) + } + async fn batch_add_deleted( &self, org_id: &str, @@ -365,7 +432,7 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp .collect()) } - async fn query_by_date( + async fn query_for_merge( &self, org_id: &str, stream_type: StreamType, @@ -382,7 +449,7 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp let pool = CLIENT_RO.clone(); DB_QUERY_NUMS - .with_label_values(&["query_by_date", "file_list", ""]) + .with_label_values(&["query_for_merge", "file_list", ""]) .inc(); let start = std::time::Instant::now(); let (date_start, date_end) = date_range.unwrap_or(("".to_string(), "".to_string())); @@ -400,7 +467,7 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp .await; let time = start.elapsed().as_secs_f64(); DB_QUERY_TIME - .with_label_values(&["query_by_date", "file_list"]) + .with_label_values(&["query_for_merge", "file_list"]) .observe(time); Ok(ret? .iter() @@ -1217,20 +1284,24 @@ SELECT stream, max(id) as id, CAST(COUNT(*) AS SIGNED) AS num } async fn set_job_done(&self, ids: &[i64]) -> Result<()> { + let config = get_config(); let pool = CLIENT.clone(); DB_QUERY_NUMS .with_label_values(&["update", "file_list_jobs", ""]) .inc(); let sql = format!( - "UPDATE file_list_jobs SET status = ?, updated_at = ? WHERE id IN ({});", + "UPDATE file_list_jobs SET status = ?, updated_at = ?, dumped = ? WHERE id IN ({});", ids.iter() .map(|id| id.to_string()) .collect::>() .join(",") ); + // if dump enabled we set dumped to false, so dump job can work + // id dump disabled, we set it to true, so cleanup can remvoe the jobs sqlx::query(&sql) .bind(super::FileListJobStatus::Done) .bind(config::utils::time::now_micros()) + .bind(!config.common.file_list_dump_enabled) .execute(&pool) .await?; Ok(()) @@ -1279,11 +1350,14 @@ SELECT stream, max(id) as id, CAST(COUNT(*) AS SIGNED) AS num DB_QUERY_NUMS .with_label_values(&["delete", "file_list_jobs", ""]) .inc(); - let ret = sqlx::query(r#"DELETE FROM file_list_jobs WHERE status = ? AND updated_at < ?;"#) - .bind(super::FileListJobStatus::Done) - .bind(before_date) - .execute(&pool) - .await?; + let ret = sqlx::query( + r#"DELETE FROM file_list_jobs WHERE status = ? AND updated_at < ? AND dumped = ?;"#, + ) + .bind(super::FileListJobStatus::Done) + .bind(before_date) + .bind(true) + .execute(&pool) + .await?; if ret.rows_affected() > 0 { log::warn!("[MYSQL] clean done jobs"); } @@ -1326,6 +1400,110 @@ SELECT stream, max(id) as id, CAST(COUNT(*) AS SIGNED) AS num } Ok(job_status) } + + async fn get_entries_in_range( + &self, + org: &str, + stream: Option<&str>, + time_start: i64, + time_end: i64, + min_id: Option, + ) -> Result> { + if time_start == 0 && time_end == 0 { + return Ok(Vec::new()); + } + + let day_partitions = if time_end - time_start <= DAY_MICRO_SECS + || time_end - time_start > DAY_MICRO_SECS * 30 + || !get_config().limit.file_list_multi_thread + { + vec![(time_start, time_end)] + } else { + let mut partitions = Vec::new(); + let mut start = time_start; + while start < time_end { + let end_of_day = std::cmp::min(end_of_the_day(start), time_end); + partitions.push((start, end_of_day)); + start = end_of_day + 1; // next day, use end_of_day + 1 microsecond + } + partitions + }; + + let mut tasks = Vec::with_capacity(day_partitions.len()); + + for (time_start, time_end) in day_partitions { + let o = org.to_string(); + let sql = "SELECT * FROM file_list WHERE max_ts >= ? AND min_ts <= ? AND org = ? AND deleted = ?"; + let sql = match stream { + Some(stream) => format!("{sql} AND stream = '{stream}'"), + None => sql.to_string(), + }; + let sql = match min_id { + Some(id) => format!("{sql} AND id >= {id}"), + None => sql, + }; + tasks.push(tokio::task::spawn(async move { + let pool = CLIENT.clone(); + sqlx::query_as::<_, super::FileRecord>(&sql) + .bind(time_start) + .bind(time_end) + .bind(o) + .bind(false) + .fetch_all(&pool) + .await + })); + } + + let mut rets = Vec::new(); + for task in tasks { + match task.await { + Ok(Ok(r)) => rets.extend(r.into_iter().filter(|r| !r.deleted)), + Ok(Err(e)) => { + return Err(e.into()); + } + Err(e) => { + return Err(Error::Message(e.to_string())); + } + }; + } + + Ok(rets) + } + + async fn get_pending_dump_jobs(&self) -> Result> { + let pool = CLIENT_RO.clone(); + + DB_QUERY_NUMS + .with_label_values(&["select", "file_list_jobs"]) + .inc(); + let ret = sqlx::query_as::<_, (i64,String, String, i64)>( + r#"SELECT id, org, stream, offsets FROM file_list_jobs WHERE status = ? AND dumped = ? limit 1000"#, + ) + .bind(super::FileListJobStatus::Done) + .bind(false) + .fetch_all(&pool) + .await?; + + let mut pending: Vec<(i64, String, String, i64)> = Vec::new(); + + for (id, org, stream, offset) in ret.iter() { + pending.push((*id, org.to_string(), stream.to_string(), *offset)); + } + + Ok(pending) + } + async fn set_job_dumped_status(&self, id: i64, dumped: bool) -> Result<()> { + let pool = CLIENT.clone(); + DB_QUERY_NUMS + .with_label_values(&["update", "file_list_jobs"]) + .inc(); + sqlx::query(r#"UPDATE file_list_jobs SET dumped = ? WHERE id = ?;"#) + .bind(dumped) + .bind(id) + .execute(&pool) + .await?; + Ok(()) + } } impl MysqlFileList { @@ -1616,6 +1794,9 @@ CREATE TABLE IF NOT EXISTS stream_stats let data_type = "BOOLEAN default false not null"; add_column("file_list_deleted", column, data_type).await?; + // create col dumped for file_list_jobs for version <=0.14.0 + add_column("file_list_jobs", "dumped", "BOOLEAN default false not null").await?; + Ok(()) } @@ -1634,6 +1815,11 @@ pub async fn create_table_index() -> Result<()> { "file_list", &["stream", "date"], ), + ( + "file_list_org_deleted_stream_idx", + "file_list", + &["org", "deleted", "stream"], + ), ("file_list_history_org_idx", "file_list_history", &["org"]), ( "file_list_history_stream_ts_idx", diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs index 97df2ddcc5..fde5a7ac2e 100644 --- a/src/infra/src/file_list/postgres.rs +++ b/src/infra/src/file_list/postgres.rs @@ -106,6 +106,76 @@ impl super::FileList for PostgresFileList { self.inner_batch_process("file_list", files).await } + async fn update_dump_records(&self, dump_file: &FileKey, dumped_ids: &[i64]) -> Result<()> { + let pool = CLIENT.clone(); + let mut tx = pool.begin().await?; + + let (stream_key, date_key, file_name) = + parse_file_key_columns(&dump_file.key).map_err(|e| Error::Message(e.to_string()))?; + let org_id = stream_key[..stream_key.find('/').unwrap()].to_string(); + let meta = &dump_file.meta; + DB_QUERY_NUMS + .with_label_values(&["insert", "file_list", ""]) + .inc(); + if let Err(e) = sqlx::query(r#"INSERT INTO file_list (org, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT DO NOTHING;"# + ) + .bind(org_id) + .bind(stream_key) + .bind(date_key) + .bind(file_name) + .bind(false) + .bind(meta.min_ts) + .bind(meta.max_ts) + .bind(meta.records) + .bind(meta.original_size) + .bind(meta.compressed_size) + .bind(meta.index_size) + .bind(meta.flattened) + .execute(&mut *tx).await + { + if let Err(e) = tx.rollback().await { + log::error!( + "[POSTGRES] rollback file_list update dump error: {e}", + ); + } + return Err(e.into()); + } + + for chunk in dumped_ids.chunks(get_config().limit.file_list_id_batch_size) { + if chunk.is_empty() { + continue; + } + let ids = chunk + .iter() + .map(|id| id.to_string()) + .collect::>() + .join(","); + let query_str = format!("delete FROM file_list WHERE id IN ({ids})"); + DB_QUERY_NUMS + .with_label_values(&["delete_by_ids", "file_list"]) + .inc(); + let start = std::time::Instant::now(); + let res = sqlx::query(&query_str).execute(&mut *tx).await; + let time = start.elapsed().as_secs_f64(); + DB_QUERY_TIME + .with_label_values(&["delete_by_ids", "file_list"]) + .observe(time); + if let Err(e) = res { + if let Err(e) = tx.rollback().await { + log::error!("[POSTGRES] rollback file_list update dump error: {e}",); + } + return Err(e.into()); + } + } + + if let Err(e) = tx.commit().await { + log::error!("[POSTGRES] commit file_list dump update error: {e}",); + return Err(e.into()); + } + Ok(()) + } + async fn batch_add_deleted( &self, org_id: &str, @@ -373,7 +443,7 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp .collect()) } - async fn query_by_date( + async fn query_for_merge( &self, org_id: &str, stream_type: StreamType, @@ -390,7 +460,7 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp let pool = CLIENT_RO.clone(); DB_QUERY_NUMS - .with_label_values(&["query_by_date", "file_list", ""]) + .with_label_values(&["query_for_merge", "file_list", ""]) .inc(); let start = std::time::Instant::now(); @@ -409,7 +479,7 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp .await; let time = start.elapsed().as_secs_f64(); DB_QUERY_TIME - .with_label_values(&["query_by_date", "file_list"]) + .with_label_values(&["query_for_merge", "file_list"]) .observe(time); Ok(ret? .iter() @@ -1158,20 +1228,24 @@ SELECT stream, max(id) as id, COUNT(*)::BIGINT AS num } async fn set_job_done(&self, ids: &[i64]) -> Result<()> { + let config = get_config(); let pool = CLIENT.clone(); DB_QUERY_NUMS .with_label_values(&["update", "file_list_jobs", ""]) .inc(); let sql = format!( - "UPDATE file_list_jobs SET status = $1, updated_at = $2 WHERE id IN ({});", + "UPDATE file_list_jobs SET status = $1, updated_at = $2, dumped = $3 WHERE id IN ({});", ids.iter() .map(|id| id.to_string()) .collect::>() .join(",") ); + // if dump enabled we set dumped to false, so dump job can work + // id dump disabled, we set it to true, so cleanup can remvoe the jobs sqlx::query(&sql) .bind(super::FileListJobStatus::Done) .bind(config::utils::time::now_micros()) + .bind(!config.common.file_list_dump_enabled) .execute(&pool) .await?; Ok(()) @@ -1220,12 +1294,14 @@ SELECT stream, max(id) as id, COUNT(*)::BIGINT AS num DB_QUERY_NUMS .with_label_values(&["delete", "file_list_jobs", ""]) .inc(); - let ret = - sqlx::query(r#"DELETE FROM file_list_jobs WHERE status = $1 AND updated_at < $2;"#) - .bind(super::FileListJobStatus::Done) - .bind(before_date) - .execute(&pool) - .await?; + let ret = sqlx::query( + r#"DELETE FROM file_list_jobs WHERE status = $1 AND updated_at < $2 AND dumped = $3;"#, + ) + .bind(super::FileListJobStatus::Done) + .bind(before_date) + .bind(true) + .execute(&pool) + .await?; if ret.rows_affected() > 0 { log::warn!("[POSTGRES] clean done jobs"); } @@ -1271,6 +1347,110 @@ SELECT stream, max(id) as id, COUNT(*)::BIGINT AS num } Ok(job_status) } + + async fn get_entries_in_range( + &self, + org: &str, + stream: Option<&str>, + time_start: i64, + time_end: i64, + min_id: Option, + ) -> Result> { + if time_start == 0 && time_end == 0 { + return Ok(Vec::new()); + } + + let day_partitions = if time_end - time_start <= DAY_MICRO_SECS + || time_end - time_start > DAY_MICRO_SECS * 30 + || !get_config().limit.file_list_multi_thread + { + vec![(time_start, time_end)] + } else { + let mut partitions = Vec::new(); + let mut start = time_start; + while start < time_end { + let end_of_day = std::cmp::min(end_of_the_day(start), time_end); + partitions.push((start, end_of_day)); + start = end_of_day + 1; // next day, use end_of_day + 1 microsecond + } + partitions + }; + + let mut tasks = Vec::with_capacity(day_partitions.len()); + + for (time_start, time_end) in day_partitions { + let o = org.to_string(); + let sql = "SELECT * FROM file_list WHERE max_ts >= $1 AND min_ts <= $2 AND org = $3 AND deleted = $4"; + let sql = match stream { + Some(stream) => format!("{sql} AND stream = '{stream}'"), + None => sql.to_string(), + }; + let sql = match min_id { + Some(id) => format!("{sql} AND id >= {id}"), + None => sql, + }; + tasks.push(tokio::task::spawn(async move { + let pool = CLIENT.clone(); + sqlx::query_as::<_, super::FileRecord>(&sql) + .bind(time_start) + .bind(time_end) + .bind(o) + .bind(false) + .fetch_all(&pool) + .await + })); + } + + let mut rets = Vec::new(); + for task in tasks { + match task.await { + Ok(Ok(r)) => rets.extend(r.into_iter().filter(|r| !r.deleted)), + Ok(Err(e)) => { + return Err(e.into()); + } + Err(e) => { + return Err(Error::Message(e.to_string())); + } + }; + } + Ok(rets) + } + + async fn get_pending_dump_jobs(&self) -> Result> { + let pool = CLIENT_RO.clone(); + + DB_QUERY_NUMS + .with_label_values(&["select", "file_list_jobs"]) + .inc(); + let ret = sqlx::query_as::<_, (i64,String, String, i64)>( + r#"SELECT id, org, stream, offsets FROM file_list_jobs WHERE status = $1 AND dumped = $2 limit 1000"#, + ) + .bind(super::FileListJobStatus::Done) + .bind(false) + .fetch_all(&pool) + .await?; + + let mut pending: Vec<(i64, String, String, i64)> = Vec::new(); + + for (id, org, stream, offset) in ret.iter() { + pending.push((*id, org.to_string(), stream.to_string(), *offset)); + } + + Ok(pending) + } + + async fn set_job_dumped_status(&self, id: i64, dumped: bool) -> Result<()> { + let pool = CLIENT.clone(); + DB_QUERY_NUMS + .with_label_values(&["update", "file_list_jobs"]) + .inc(); + sqlx::query("UPDATE file_list_jobs SET dumped = $1 WHERE id = $2;") + .bind(dumped) + .bind(id) + .execute(&pool) + .await?; + Ok(()) + } } impl PostgresFileList { @@ -1566,6 +1746,9 @@ CREATE TABLE IF NOT EXISTS stream_stats let data_type = "BOOLEAN default false not null"; add_column("file_list_deleted", column, data_type).await?; + // create col dumped for file_list_jobs for version <=0.14.0 + add_column("file_list_jobs", "dumped", "BOOLEAN default false not null").await?; + Ok(()) } @@ -1584,6 +1767,11 @@ pub async fn create_table_index() -> Result<()> { "file_list", &["stream", "date"], ), + ( + "file_list_org_deleted_stream_idx", + "file_list", + &["org", "deleted", "stream"], + ), ("file_list_history_org_idx", "file_list_history", &["org"]), ( "file_list_history_stream_ts_idx", diff --git a/src/infra/src/file_list/sqlite.rs b/src/infra/src/file_list/sqlite.rs index dd3b3a7064..2a5c6f1f09 100644 --- a/src/infra/src/file_list/sqlite.rs +++ b/src/infra/src/file_list/sqlite.rs @@ -107,6 +107,64 @@ impl super::FileList for SqliteFileList { self.inner_batch_process("file_list", files).await } + async fn update_dump_records(&self, dump_file: &FileKey, dumped_ids: &[i64]) -> Result<()> { + let client = CLIENT_RW.clone(); + let client = client.lock().await; + + let mut tx = client.begin().await?; + + let (stream_key, date_key, file_name) = + parse_file_key_columns(&dump_file.key).map_err(|e| Error::Message(e.to_string()))?; + let org_id = stream_key[..stream_key.find('/').unwrap()].to_string(); + let meta = &dump_file.meta; + + if let Err(e) = sqlx::query(r#" INSERT INTO file_list (org, stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, index_size, flattened) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12);"#) + .bind(org_id) + .bind(stream_key) + .bind(date_key) + .bind(file_name) + .bind(false) + .bind(meta.min_ts) + .bind(meta.max_ts) + .bind(meta.records) + .bind(meta.original_size) + .bind(meta.compressed_size) + .bind(meta.index_size) + .bind(meta.flattened) + .execute(&mut *tx) + .await{ + if let Err(e) = tx.rollback().await { + log::error!("[SQLITE] rollback file_list dump file update error: {e}",); + } + return Err(e.into()); + } + + for chunk in dumped_ids.chunks(get_config().limit.file_list_id_batch_size) { + if chunk.is_empty() { + continue; + } + let ids = chunk + .iter() + .map(|id| id.to_string()) + .collect::>() + .join(","); + let query_str = format!("DELETE FROM file_list WHERE id IN ({ids})"); + if let Err(e) = sqlx::query(&query_str).execute(&mut *tx).await { + if let Err(e) = tx.rollback().await { + log::error!("[SQLITE] rollback file_list dump file update error: {e}",); + } + return Err(e.into()); + } + } + + if let Err(e) = tx.commit().await { + log::error!("[SQLITE] transaction commit error for dump file update {e}"); + return Err(e.into()); + } + Ok(()) + } + async fn batch_add_deleted( &self, org_id: &str, @@ -348,7 +406,7 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp .collect()) } - async fn query_by_date( + async fn query_for_merge( &self, org_id: &str, stream_type: StreamType, @@ -1020,18 +1078,22 @@ SELECT stream, max(id) as id, COUNT(*) AS num } async fn set_job_done(&self, ids: &[i64]) -> Result<()> { + let config = get_config(); let client = CLIENT_RW.clone(); let client = client.lock().await; let sql = format!( - "UPDATE file_list_jobs SET status = $1, updated_at = $2 WHERE id IN ({});", + "UPDATE file_list_jobs SET status = $1, updated_at = $2, dumped = $3 WHERE id IN ({});", ids.iter() .map(|id| id.to_string()) .collect::>() .join(",") ); + // if dump enabled we set dumped to false, so dump job can work + // id dump disabled, we set it to true, so cleanup can remvoe the jobs sqlx::query(&sql) .bind(super::FileListJobStatus::Done) .bind(config::utils::time::now_micros()) + .bind(!config.common.file_list_dump_enabled) .execute(&*client) .await?; Ok(()) @@ -1074,12 +1136,14 @@ SELECT stream, max(id) as id, COUNT(*) AS num async fn clean_done_jobs(&self, before_date: i64) -> Result<()> { let client = CLIENT_RW.clone(); let client = client.lock().await; - let ret = - sqlx::query(r#"DELETE FROM file_list_jobs WHERE status = $1 AND updated_at < $2;"#) - .bind(super::FileListJobStatus::Done) - .bind(before_date) - .execute(&*client) - .await?; + let ret = sqlx::query( + r#"DELETE FROM file_list_jobs WHERE status = $1 AND updated_at < $2 AND dumped = $3;"#, + ) + .bind(super::FileListJobStatus::Done) + .bind(before_date) + .bind(true) + .execute(&*client) + .await?; if ret.rows_affected() > 0 { log::warn!("[SQLITE] clean done jobs"); } @@ -1119,6 +1183,104 @@ SELECT stream, max(id) as id, COUNT(*) AS num } Ok(job_status) } + + async fn get_entries_in_range( + &self, + org: &str, + stream: Option<&str>, + time_start: i64, + time_end: i64, + min_id: Option, + ) -> Result> { + if time_start == 0 && time_end == 0 { + return Ok(Vec::new()); + } + + let day_partitions = if time_end - time_start <= DAY_MICRO_SECS + || time_end - time_start > DAY_MICRO_SECS * 30 + || !get_config().limit.file_list_multi_thread + { + vec![(time_start, time_end)] + } else { + let mut partitions = Vec::new(); + let mut start = time_start; + while start < time_end { + let end_of_day = std::cmp::min(end_of_the_day(start), time_end); + partitions.push((start, end_of_day)); + start = end_of_day + 1; // next day, use end_of_day + 1 microsecond + } + partitions + }; + + let mut tasks = Vec::with_capacity(day_partitions.len()); + + for (time_start, time_end) in day_partitions { + let o = org.to_string(); + let sql = "SELECT * FROM file_list WHERE max_ts >= $1 AND min_ts <= $2 AND org = $3 AND deleted = $4"; + let sql = match stream { + Some(stream) => format!("{sql} AND stream = '{stream}'"), + None => sql.to_string(), + }; + let sql = match min_id { + Some(id) => format!("{sql} AND id >= {id}"), + None => sql, + }; + tasks.push(tokio::task::spawn(async move { + let pool = CLIENT_RO.clone(); + sqlx::query_as::<_, super::FileRecord>(&sql) + .bind(time_start) + .bind(time_end) + .bind(o) + .bind(false) + .fetch_all(&pool) + .await + })); + } + + let mut rets = Vec::new(); + for task in tasks { + match task.await { + Ok(Ok(r)) => rets.extend(r.into_iter().filter(|r| !r.deleted)), + Ok(Err(e)) => { + return Err(e.into()); + } + Err(e) => { + return Err(Error::Message(e.to_string())); + } + }; + } + Ok(rets) + } + + async fn get_pending_dump_jobs(&self) -> Result> { + let pool = CLIENT_RO.clone(); + + let ret = sqlx::query_as::<_, (i64,String, String, i64)>( + r#"SELECT id, org, stream, offsets FROM file_list_jobs WHERE status = $1 AND dumped = $2 limit 1000"#, + ) + .bind(super::FileListJobStatus::Done) + .bind(false) + .fetch_all(&pool) + .await?; + + let mut pending: Vec<(i64, String, String, i64)> = Vec::new(); + + for (id, org, stream, offset) in ret.iter() { + pending.push((*id, org.to_string(), stream.to_string(), *offset)); + } + + Ok(pending) + } + async fn set_job_dumped_status(&self, id: i64, dumped: bool) -> Result<()> { + let client = CLIENT_RW.clone(); + let client = client.lock().await; + sqlx::query("UPDATE file_list_jobs SET dumped = $1 WHERE id = $2;") + .bind(dumped) + .bind(id) + .execute(&*client) + .await?; + Ok(()) + } } impl SqliteFileList { @@ -1414,6 +1576,15 @@ CREATE TABLE IF NOT EXISTS stream_stats let data_type = "BOOLEAN default false not null"; add_column(&client, "file_list_deleted", column, data_type).await?; + // create col dumped for file_list_jobs for version <=0.14.0 + add_column( + &client, + "file_list_jobs", + "dumped", + "BOOLEAN default false not null", + ) + .await?; + Ok(()) } @@ -1430,6 +1601,11 @@ pub async fn create_table_index() -> Result<()> { "file_list", &["stream", "date"], ), + ( + "file_list_org_deleted_stream_idx", + "file_list", + &["org", "deleted", "stream"], + ), ("file_list_history_org_idx", "file_list_history", &["org"]), ( "file_list_history_stream_ts_idx", diff --git a/src/job/file_list_dump.rs b/src/job/file_list_dump.rs new file mode 100644 index 0000000000..fe57c64ed7 --- /dev/null +++ b/src/job/file_list_dump.rs @@ -0,0 +1,450 @@ +// Copyright 2025 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use std::sync::Arc; + +use arrow::{ + array::{BooleanBuilder, Int64Builder, StringBuilder}, + datatypes::Field, + record_batch::RecordBatch, +}; +use arrow_schema::Schema; +use chrono::{Datelike, Timelike, Utc}; +use config::{ + PARQUET_BATCH_SIZE, PARQUET_MAX_ROW_GROUP_SIZE, + cluster::{self, LOCAL_NODE}, + get_config, get_parquet_compression, + meta::stream::{FileKey, FileMeta, StreamType}, + utils::time::hour_micros, +}; +use hashbrown::HashSet; +use infra::{ + dist_lock, + file_list::FileRecord, + schema::{STREAM_SCHEMAS_LATEST, SchemaCache}, +}; +use itertools::Itertools; +use once_cell::sync::Lazy; +use parquet::{arrow::AsyncArrowWriter, file::properties::WriterProperties}; +use tokio::sync::{ + Mutex, RwLock, + mpsc::{Receiver, Sender}, +}; + +use crate::{common::infra::cluster::get_node_by_uuid, service::db}; + +pub static FILE_LIST_SCHEMA: Lazy> = Lazy::new(|| { + Arc::new(Schema::new(vec![ + Field::new("id", arrow_schema::DataType::Int64, false), + Field::new("org", arrow_schema::DataType::Utf8, false), + Field::new("stream", arrow_schema::DataType::Utf8, false), + Field::new("date", arrow_schema::DataType::Utf8, false), + Field::new("file", arrow_schema::DataType::Utf8, false), + Field::new("deleted", arrow_schema::DataType::Boolean, false), + Field::new("flattened", arrow_schema::DataType::Boolean, false), + Field::new("min_ts", arrow_schema::DataType::Int64, false), + Field::new("max_ts", arrow_schema::DataType::Int64, false), + Field::new("records", arrow_schema::DataType::Int64, false), + Field::new("original_size", arrow_schema::DataType::Int64, false), + Field::new("compressed_size", arrow_schema::DataType::Int64, false), + Field::new("index_size", arrow_schema::DataType::Int64, true), + ])) +}); + +const DUMP_JOB_MIN_INTERVAL: i64 = 30; + +// these correspond to job_id, org_id, stream and offset +// job id is the id from db for that job row +type JobInfo = (i64, String, String, i64); + +struct DownloadQueue { + sender: Sender, + receiver: Arc>>, +} + +impl DownloadQueue { + fn new(sender: Sender, receiver: Arc>>) -> Self { + Self { sender, receiver } + } +} + +// The reason we set it to 1 here is there is no special benefit for having larger queue in this +// case the sender can simply wait on the channel if there is no worker available. +static FILE_LIST_DUMP_CHANNEL: Lazy = Lazy::new(|| { + let (tx, rx) = tokio::sync::mpsc::channel::(1); + DownloadQueue::new(tx, Arc::new(Mutex::new(rx))) +}); + +// This stores the job ids that are currently in the queue for dumping +// normally the queue should be big enough and dumping should be fast enough +// that before the dump job runs again, everyhting from queue is dumped. But in extream cases +// it is possible that there are still jobs in the queue witing to be dumped, and the +// dump job (enqueing new jobs) runs again. In this case, it will again pick up the jobs which are +// currently in queue, because they are not marked as done in the db. In such case, it is possible +// that files in that time range get dumped before they are compacted. So instead we maintain this +// and before enqueing, check if the job_id is already waiting in queue or not. +static ONGOING_JOB_IDS: Lazy>> = Lazy::new(|| RwLock::new(HashSet::new())); + +fn get_dump_stream_key(org: &str, stream: &str) -> String { + format!( + "{}/{}/{}", + org, + StreamType::Filelist, + stream.replace('/', "_") + ) +} + +async fn lock_stream(org_id: &str, stream_type: StreamType, stream_name: &str) -> Option<()> { + let (_, 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 None; // 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 = match dist_lock::lock(&lock_key, 0).await { + Ok(v) => v, + Err(_) => return None, + }; + // 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() + { + // we cannot do anything even if unlock fails, so ignore + let _ = dist_lock::unlock(&locker).await; + return None; // 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; + let _ = dist_lock::unlock(&locker).await; + drop(locker); + match ret { + Ok(_) => {} + Err(_) => return None, + } + } + Some(()) +} + +pub async fn run() -> Result<(), anyhow::Error> { + // cache generation is only done on compactor + if !LOCAL_NODE.is_compactor() { + return Ok(()); + } + let config = get_config(); + + if !config.common.file_list_dump_enabled { + return Ok(()); + } + + // spawn threads which will do the actual dumping + for _ in 0..config.limit.file_merge_thread_num { + let rx = FILE_LIST_DUMP_CHANNEL.receiver.clone(); + tokio::spawn(async move { + loop { + let ret = rx.lock().await.recv().await; + match ret { + None => { + log::debug!("[FILE_LIST_DUMP:JOB] Receiving channel is closed"); + break; + } + Some((job_id, org, stream, offset)) => { + if let Err(e) = dump(job_id, &org, &stream, offset).await { + log::error!("error in dumping files {stream} offset {offset} : {e}"); + } + let mut ongoing = ONGOING_JOB_IDS.write().await; + ongoing.remove(&job_id); + drop(ongoing); + } + } + } + }); + } + + let interval: u64 = std::cmp::max(config.compact.interval as i64, DUMP_JOB_MIN_INTERVAL) + .try_into() + .unwrap(); + + // loop and keep checking on file_list_jobs for next dump jobs + loop { + if cluster::is_offline() { + break; + } + + // sleep + tokio::time::sleep(tokio::time::Duration::from_secs(interval)).await; + + let pending = infra::file_list::get_pending_dump_jobs().await?; + let threshold_hour = Utc::now().timestamp_micros() + - (config.common.file_list_dump_min_hour as i64 * hour_micros(1)); + + let ongoing = ONGOING_JOB_IDS.read().await; + let pending = pending + .into_iter() + .filter(|(id, _, _, offset)| { + let end = offset + hour_micros(1); + if end >= threshold_hour { + return false; + } + // if we already have that job in queue, we should not re-queue it + !ongoing.contains(id) + }) + .collect::>(); + // it is important to drop this here, as the worker threads will + // also write lock this in order to remove successful jobs + // if we don't drop, we can deadlock if the queue gets full, because + // worker will block on this lock, queue is full, and we block on queue to have space + // but no worker free to remove from queue. + drop(ongoing); + for (job_id, org, stream, offset) in pending { + if let Err(e) = FILE_LIST_DUMP_CHANNEL + .sender + .send((job_id, org, stream, offset)) + .await + { + log::error!("error in sending dump job to worker thread for {job_id} : {e}"); + } else { + let mut ongoing = ONGOING_JOB_IDS.write().await; + ongoing.insert(job_id); + drop(ongoing); + } + } + } + log::info!("[COMPACTOR:JOB] job::files::file_list_dump is stopped"); + Ok(()) +} + +async fn dump(job_id: i64, org: &str, stream: &str, offset: i64) -> Result<(), anyhow::Error> { + let start = offset; + let end = offset + hour_micros(1); + + let columns = stream.split('/').collect::>(); + if columns.len() != 3 { + log::warn!("invalid stream name format for {job_id} : {stream}"); + infra::file_list::set_job_dumped_status(job_id, true).await?; + return Ok(()); + } + let stream_type = StreamType::from(columns[1]); + if stream_type == StreamType::Filelist { + // we do not want to dump the dumped files + infra::file_list::set_job_dumped_status(job_id, true).await?; + return Ok(()); + } + + if lock_stream(org, stream_type, stream).await.is_none() { + // someone else is processing this. + return Ok(()); + } + let files = infra::file_list::get_entries_in_range(org, Some(stream), start, end, None).await?; + if files.is_empty() { + if let Err(e) = infra::file_list::set_job_dumped_status(job_id, true).await { + log::error!("error in setting dumped = true for job with id {job_id}, error : {e}"); + } + return Ok(()); + } + if let Err(e) = generate_dump(org, stream, (start, end), files, job_id).await { + log::error!("[COMPACTOR:JOB] file_list dump file generation error : {e}"); + } else { + let stream_key = get_dump_stream_key(org, stream); + let schema_exists = STREAM_SCHEMAS_LATEST + .read() + .await + .get(&stream_key) + .is_some(); + if !schema_exists { + if let Err(e) = super::db::schema::merge( + org, + &stream.replace('/', "_"), + StreamType::Filelist, + &FILE_LIST_SCHEMA, + Some(Utc::now().timestamp_micros()), + ) + .await + { + log::error!("erroring in saving file list schema for {stream_key} to db : {e}"); + } + + let cache = SchemaCache::new(FILE_LIST_SCHEMA.as_ref().to_owned()); + let mut w = STREAM_SCHEMAS_LATEST.write().await; + w.insert(stream_key.clone(), cache); + } + } + Ok(()) +} + +fn get_writer( + schema: Arc, + buf: &mut Vec, +) -> Result>, anyhow::Error> { + let cfg = get_config(); + let writer_props = WriterProperties::builder() + .set_write_batch_size(PARQUET_BATCH_SIZE) // in bytes + .set_max_row_group_size(PARQUET_MAX_ROW_GROUP_SIZE) // maximum number of rows in a row group + .set_compression(get_parquet_compression(&cfg.common.parquet_compression)); + + let writer_props = writer_props.build(); + let writer = AsyncArrowWriter::try_new(buf, schema.clone(), Some(writer_props))?; + Ok(writer) +} + +fn create_record_batch(files: Vec) -> Result { + let schema = FILE_LIST_SCHEMA.clone(); + if files.is_empty() { + return Ok(RecordBatch::new_empty(schema)); + } + let batch_size = files.len(); + + let mut field_id = Int64Builder::with_capacity(batch_size); + let mut field_org = StringBuilder::with_capacity(batch_size, batch_size * 128); + let mut field_stream = StringBuilder::with_capacity(batch_size, batch_size * 256); + let mut field_date = StringBuilder::with_capacity(batch_size, batch_size * 10); + let mut field_file = StringBuilder::with_capacity(batch_size, batch_size * 20); + let mut field_deleted = BooleanBuilder::with_capacity(batch_size); + let mut field_min_ts = Int64Builder::with_capacity(batch_size); + let mut field_max_ts = Int64Builder::with_capacity(batch_size); + let mut field_records = Int64Builder::with_capacity(batch_size); + let mut field_original_size = Int64Builder::with_capacity(batch_size); + let mut field_compressed_size = Int64Builder::with_capacity(batch_size); + let mut field_index_size = Int64Builder::with_capacity(batch_size); + let mut field_flattened = BooleanBuilder::with_capacity(batch_size); + + for file in files { + field_id.append_value(file.id); + field_org.append_value(file.org); + field_stream.append_value(file.stream); + field_date.append_value(file.date); + field_file.append_value(file.file); + field_deleted.append_value(file.deleted); + field_min_ts.append_value(file.min_ts); + field_max_ts.append_value(file.max_ts); + field_records.append_value(file.records); + field_original_size.append_value(file.original_size); + field_compressed_size.append_value(file.compressed_size); + field_index_size.append_value(file.index_size); + field_flattened.append_value(file.flattened); + } + + let batch = RecordBatch::try_new( + schema.clone(), + vec![ + Arc::new(field_id.finish()), + Arc::new(field_org.finish()), + Arc::new(field_stream.finish()), + Arc::new(field_date.finish()), + Arc::new(field_file.finish()), + Arc::new(field_deleted.finish()), + Arc::new(field_flattened.finish()), + Arc::new(field_min_ts.finish()), + Arc::new(field_max_ts.finish()), + Arc::new(field_records.finish()), + Arc::new(field_original_size.finish()), + Arc::new(field_compressed_size.finish()), + Arc::new(field_index_size.finish()), + ], + )?; + Ok(batch) +} + +async fn generate_dump( + org: &str, + stream: &str, + range: (i64, i64), + files: Vec, + job_id: i64, +) -> Result<(), anyhow::Error> { + // if there are no files to dump, no point in making a dump file + if files.is_empty() { + return Ok(()); + } + + let config = get_config(); + let batch_size = files.len(); + + // if dual write is not enabled, we will remove the records + // else keep them + let ids: Vec = if !config.common.file_list_dump_dual_write { + files.iter().map(|r| r.id).collect() + } else { + vec![] + }; + + let mut buf = Vec::new(); + let mut writer = get_writer(FILE_LIST_SCHEMA.clone(), &mut buf)?; + // we split the file list into vec of vecs, with each sub-vec having PARQUET_BATCH_SIZE elements + // at most (last may be shorter) doing this the following way means we don't have to clone + // anything, we simply shift the ownership via iterator + let chunks: Vec> = files + .into_iter() + .chunks(PARQUET_BATCH_SIZE) + .into_iter() + .map(|c| c.collect()) + .collect(); + for chunk in chunks { + let batch = create_record_batch(chunk)?; + writer.write(&batch).await?; + } + writer.close().await?; + + let ts = chrono::DateTime::from_timestamp_micros(range.0).unwrap(); + + let file_name = format!( + "{:04}/{:02}/{:02}/{:02}/{}.parquet", + ts.year(), + ts.month(), + ts.day(), + ts.hour(), + config::ider::generate() + ); + + let stream_key = get_dump_stream_key(org, stream); + let file_key = format!("files/{stream_key}/{file_name}"); + + let meta = FileMeta { + min_ts: range.0, + max_ts: range.1, + records: batch_size as i64, + original_size: buf.len() as i64, + compressed_size: buf.len() as i64, + index_size: 0, + flattened: false, + }; + + let dump_file = FileKey { + key: file_key.clone(), + meta: meta.clone(), + deleted: false, + segment_ids: None, + }; + + // first store the file in storage + // then update the entries in db, + // and if both pass only then set the job as dumped=true + + infra::storage::put(&file_key, buf.into()).await?; + infra::file_list::update_dump_records(&dump_file, &ids).await?; + infra::file_list::set_job_dumped_status(job_id, true).await?; + log::info!( + "successfully dumped file list {stream} offset {} count {batch_size}", + range.0 + ); + Ok(()) +} diff --git a/src/job/mod.rs b/src/job/mod.rs index b1531d3e06..e2de3fd3ed 100644 --- a/src/job/mod.rs +++ b/src/job/mod.rs @@ -32,6 +32,7 @@ mod alert_manager; mod cipher; mod compactor; mod file_downloader; +mod file_list_dump; pub(crate) mod files; mod flatten_compactor; pub mod metrics; @@ -43,6 +44,7 @@ pub(crate) mod syslog_server; mod telemetry; pub use file_downloader::queue_background_download; +pub use file_list_dump::FILE_LIST_SCHEMA; pub use mmdb_downloader::MMDB_INIT_NOTIFIER; pub async fn init() -> Result<(), anyhow::Error> { @@ -214,6 +216,10 @@ pub async fn init() -> Result<(), anyhow::Error> { tokio::task::spawn(async move { alert_manager::run().await }); tokio::task::spawn(async move { file_downloader::run().await }); + if LOCAL_NODE.is_compactor() { + tokio::task::spawn(async move { file_list_dump::run().await }); + } + // load metrics disk cache tokio::task::spawn(async move { crate::service::promql::search::init().await }); // start pipeline data retention diff --git a/src/service/compact/merge.rs b/src/service/compact/merge.rs index 6d8c918639..a87e401e42 100644 --- a/src/service/compact/merge.rs +++ b/src/service/compact/merge.rs @@ -462,9 +462,10 @@ pub async fn merge_by_stream( offset_time.format("%Y/%m/%d/%H").to_string(), ) }; - let files = file_list::query_by_date(org_id, stream_name, stream_type, &date_start, &date_end) - .await - .map_err(|e| anyhow::anyhow!("query file list failed: {}", e))?; + let files = + file_list::query_for_merge(org_id, stream_name, stream_type, &date_start, &date_end) + .await + .map_err(|e| anyhow::anyhow!("query file list failed: {}", e))?; log::debug!( "[COMPACTOR] merge_by_stream [{}/{}/{}] date range: [{},{}], files: {}", diff --git a/src/service/compact/retention.rs b/src/service/compact/retention.rs index 21d19ca523..50240f3b4a 100644 --- a/src/service/compact/retention.rs +++ b/src/service/compact/retention.rs @@ -277,6 +277,7 @@ pub async fn delete_all( // delete from file list delete_from_file_list(org_id, stream_type, stream_name, (start_time, end_time)).await?; + super::super::file_list_dump::delete_all_for_stream(org_id, stream_type, stream_name).await?; log::info!( "deleted file list for: {}/{}/{}/all", org_id, @@ -372,6 +373,14 @@ pub async fn delete_by_date( // delete from file list delete_from_file_list(org_id, stream_type, stream_name, time_range).await?; + super::super::file_list_dump::delete_in_time_range( + org_id, + stream_type, + stream_name, + (date_start.timestamp_micros(), date_end.timestamp_micros()), + ) + .await?; + // archive old schema versions let mut schema_versions = infra::schema::get_versions(org_id, stream_name, stream_type, Some(time_range)).await?; @@ -420,7 +429,15 @@ async fn delete_from_file_list( stream_name: &str, time_range: (i64, i64), ) -> Result<(), anyhow::Error> { + let task_id = tokio::task::try_id() + .map(|id| id.to_string()) + .unwrap_or_else(|| rand::random::().to_string()); + let fake_trace_id = format!( + "delete_from_file_list-{}-{}-{}", + task_id, time_range.0, time_range.1 + ); let files = file_list::query( + &fake_trace_id, org_id, stream_name, stream_type, diff --git a/src/service/compact/stats.rs b/src/service/compact/stats.rs index 20029e3523..158b9da6ff 100644 --- a/src/service/compact/stats.rs +++ b/src/service/compact/stats.rs @@ -17,7 +17,10 @@ use config::{cluster::LOCAL_NODE, meta::stream::StreamStats}; use hashbrown::HashMap; use infra::{dist_lock, file_list as infra_file_list}; -use crate::{common::infra::cluster::get_node_by_uuid, service::db}; +use crate::{ + common::infra::cluster::get_node_by_uuid, + service::{db, file_list_dump}, +}; pub async fn update_stats_from_file_list() -> Result, anyhow::Error> { // get last offset @@ -49,13 +52,21 @@ pub async fn update_stats_from_file_list() -> Result, anyhow: let add_stream_stats = infra_file_list::stats(&org_id, None, None, pk_value, false) .await .unwrap_or_default(); + let dumped_add_stats = file_list_dump::stats(&org_id, None, None, pk_value) + .await + .unwrap_or_default(); let del_stream_stats = infra_file_list::stats(&org_id, None, None, pk_value, true) .await .unwrap_or_default(); + // dump never store deleted files, so we do not have to consider deleted here let mut stream_stats = HashMap::new(); for (stream, stats) in add_stream_stats { stream_stats.insert(stream, stats); } + for (stream, stats) in dumped_add_stats { + let entry = stream_stats.entry(stream).or_insert(StreamStats::default()); + *entry = &*entry + &stats; + } for (stream, stats) in del_stream_stats { let entry = stream_stats.entry(stream).or_insert(StreamStats::default()); *entry = &*entry - &stats; diff --git a/src/service/db/file_list/mod.rs b/src/service/db/file_list/mod.rs index 918fdb33c7..d9c9eb1c4b 100644 --- a/src/service/db/file_list/mod.rs +++ b/src/service/db/file_list/mod.rs @@ -27,7 +27,6 @@ use o2_enterprise::enterprise::{ super_cluster::stream::client::super_cluster_cache_stats, }; use once_cell::sync::Lazy; - pub mod broadcast; pub mod local; @@ -134,9 +133,10 @@ async fn single_cache_stats() -> Result<()> { for org_id in orgs { let ret = infra::file_list::get_stream_stats(&org_id, None, None).await; if ret.is_err() { - log::error!("Load stream stats error: {}", ret.err().unwrap()); + log::error!("Load stream stats from db error: {}", ret.err().unwrap()); continue; } + for (stream, stats) in ret.unwrap() { let columns = stream.split('/').collect::>(); let org_id = columns[0]; diff --git a/src/service/file_list.rs b/src/service/file_list.rs index 373b5aa862..91d5bea739 100644 --- a/src/service/file_list.rs +++ b/src/service/file_list.rs @@ -13,12 +13,13 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use chrono::Utc; use config::{ cluster::LOCAL_NODE, get_config, meta::{ search::ScanStats, - stream::{FileKey, PartitionTimeLevel, StreamType}, + stream::{FileKey, FileMeta, PartitionTimeLevel, StreamType}, }, metrics::{FILE_LIST_CACHE_HIT_COUNT, FILE_LIST_ID_SELECT_COUNT}, utils::file::get_file_meta as util_get_file_meta, @@ -27,7 +28,10 @@ use hashbrown::HashSet; use infra::{errors::Result, file_list, storage}; use rayon::slice::ParallelSliceMut; -use crate::service::search::inspector::{SearchInspectorFieldsBuilder, search_inspector_fields}; +use crate::service::{ + file_list_dump, + search::inspector::{SearchInspectorFieldsBuilder, search_inspector_fields}, +}; #[tracing::instrument( name = "service::file_list::query", @@ -35,6 +39,7 @@ use crate::service::search::inspector::{SearchInspectorFieldsBuilder, search_ins fields(org_id = org_id, stream_name = stream_name) )] pub async fn query( + trace_id: &str, org_id: &str, stream_name: &str, stream_type: StreamType, @@ -42,6 +47,7 @@ pub async fn query( time_min: i64, time_max: i64, ) -> Result> { + let cfg = get_config(); let files = file_list::query( org_id, stream_type, @@ -51,6 +57,36 @@ pub async fn query( None, ) .await?; + let dumped_files = file_list_dump::query( + trace_id, + org_id, + stream_name, + stream_type, + (time_min, time_max), + None, + ) + .await?; + if cfg.common.file_list_dump_debug_check { + let dumped_file_names = dumped_files + .iter() + .map(|f| "files/".to_string() + &f.stream + "/" + &f.date + "/" + &f.file) + .collect::>(); + let missing_files: usize = files + .iter() + .map(|(name, _)| { + if dumped_file_names.contains(name) { + 0 + } else { + 1 + } + }) + .sum(); + if missing_files > 0 { + log::info!( + "[trace_id : {trace_id}] dump was missing {missing_files} files present in db" + ); + } + } let mut file_keys = Vec::with_capacity(files.len()); for file in files { file_keys.push(FileKey { @@ -60,22 +96,50 @@ pub async fn query( segment_ids: None, }); } + if !cfg.common.file_list_dump_dual_write { + // we only consider these files in case of dual write disabled, + // because with dual write there are some edge cases which cannot be sovled even with id + // de-dup so the data gets counted twice. + for file in dumped_files { + file_keys.push(FileKey { + key: "files/".to_string() + &file.stream + "/" + &file.date + "/" + &file.file, + meta: FileMeta { + min_ts: file.min_ts, + max_ts: file.max_ts, + records: file.records, + original_size: file.original_size, + compressed_size: file.compressed_size, + index_size: file.index_size, + flattened: file.flattened, + }, + deleted: file.deleted, + segment_ids: None, + }) + } + } + + file_keys.par_sort_unstable_by(|a, b| a.key.cmp(&b.key)); + file_keys.dedup_by(|a, b| a.key == b.key); Ok(file_keys) } +/// NOTE: This will not query the files from file_dump. If you also want files from the dump, use +/// query function instead. Currently this is used only when compacting on stream, and we do not +/// support re-compaction of already dumped files, so this function completely ignores the files +/// from dump #[tracing::instrument( - name = "service::file_list::query_by_date", + name = "service::file_list::query_for_merge", skip_all, fields(org_id = org_id, stream_name = stream_name) )] -pub async fn query_by_date( +pub async fn query_for_merge( org_id: &str, stream_name: &str, stream_type: StreamType, date_start: &str, date_end: &str, ) -> Result> { - let files = file_list::query_by_date( + let files = file_list::query_for_merge( org_id, stream_type, stream_name, @@ -91,11 +155,21 @@ pub async fn query_by_date( segment_ids: None, }); } + // we don't need to query from dump here, because + // we expected all dumped files to be already compacted, + // and the compaction marks old files as deleted, which is not possible with dump Ok(file_keys) } #[tracing::instrument(name = "service::file_list::query_by_ids", skip_all)] -pub async fn query_by_ids(trace_id: &str, ids: &[i64]) -> Result> { +pub async fn query_by_ids( + trace_id: &str, + ids: &[i64], + org: &str, + stream_type: StreamType, + stream: &str, + time_range: Option<(i64, i64)>, +) -> Result> { let cfg = get_config(); FILE_LIST_ID_SELECT_COUNT .with_label_values(&[]) @@ -196,12 +270,82 @@ pub async fn query_by_ids(trace_id: &str, ids: &[i64]) -> Result> { ) ); + // query from file_list_dump + // we use the min(id) as id hint because that automatically filter outs any dump files + // which cannot have the ids we are looking for + let dumped_files = file_list_dump::query( + trace_id, + org, + stream, + stream_type, + time_range.unwrap_or((0, Utc::now().timestamp_micros())), + ids.iter().min().copied(), + ) + .await?; + + let dumped_files: Vec<_> = dumped_files + .into_iter() + .filter(|r| ids.contains(&r.id)) + .map(|r| { + ( + r.id, + FileKey { + key: "files/".to_string() + &r.stream + "/" + &r.date + "/" + &r.file, + meta: FileMeta { + min_ts: r.min_ts, + max_ts: r.max_ts, + records: r.records, + original_size: r.original_size, + compressed_size: r.compressed_size, + index_size: r.index_size, + flattened: r.flattened, + }, + deleted: r.deleted, + segment_ids: None, + }, + ) + }) + .collect(); + + if cfg.common.file_list_dump_debug_check { + let dump_ids = dumped_files + .iter() + .map(|entry| entry.0) + .collect::>(); + let db_ids = db_files.iter().map(|entry| entry.0).collect::>(); + let missing_files: usize = db_ids + .iter() + .map(|id| if dump_ids.contains(id) { 0 } else { 1 }) + .sum(); + if missing_files > 0 { + log::info!( + "[trace_id : {trace_id}] dump was missing {missing_files} files present in db" + ); + } + } + // 3. set the local cache if !cfg.common.local_mode { let start = std::time::Instant::now(); let db_files: Vec<_> = db_files.iter().map(|(id, f)| (*id, f)).collect(); if let Err(e) = file_list::LOCAL_CACHE.batch_add_with_id(&db_files).await { - log::error!("[trace_id {trace_id}] file_list set cache failed: {:?}", e); + log::error!( + "[trace_id {trace_id}] file_list set cache failed for db files: {:?}", + e + ); + } + + if !cfg.common.file_list_dump_dual_write { + let dumped_files: Vec<_> = dumped_files.iter().map(|(id, f)| (*id, f)).collect(); + if let Err(e) = file_list::LOCAL_CACHE + .batch_add_with_id(&dumped_files) + .await + { + log::error!( + "[trace_id {trace_id}] file_list set cache failed for dumped files: {:?}", + e + ); + } } log::info!( @@ -225,7 +369,11 @@ pub async fn query_by_ids(trace_id: &str, ids: &[i64]) -> Result> { // 4. merge the results files.extend(db_files.into_iter().map(|(_, f)| f)); - + if !cfg.common.file_list_dump_dual_write { + files.extend(dumped_files.into_iter().map(|(_, f)| f)); + } + files.par_sort_unstable_by(|a, b| a.key.cmp(&b.key)); + files.dedup_by(|a, b| a.key == b.key); Ok(files) } @@ -235,12 +383,40 @@ pub async fn query_by_ids(trace_id: &str, ids: &[i64]) -> Result> { fields(org_id = org_id, stream_name = stream_name) )] pub async fn query_ids( + trace_id: &str, org_id: &str, stream_type: StreamType, stream_name: &str, time_range: Option<(i64, i64)>, ) -> Result> { + let cfg = get_config(); let mut files = file_list::query_ids(org_id, stream_type, stream_name, time_range).await?; + let dumped_files = super::file_list_dump::get_ids_in_range( + trace_id, + org_id, + stream_name, + stream_type, + time_range + .ok_or_else(|| infra::errors::Error::Message("time_range is required".to_string()))?, + ) + .await + .unwrap(); + if cfg.common.file_list_dump_debug_check { + let dump_ids = dumped_files.iter().map(|f| f.id).collect::>(); + let db_ids = files.iter().map(|f| f.id).collect::>(); + let missing_files: usize = db_ids + .iter() + .map(|id| if dump_ids.contains(id) { 0 } else { 1 }) + .sum(); + if missing_files > 0 { + log::info!( + "[trace_id : {trace_id}] dump was missing {missing_files} files present in db" + ); + } + } + if !cfg.common.file_list_dump_dual_write { + files.extend(dumped_files.into_iter()); + } files.par_sort_unstable_by(|a, b| a.id.cmp(&b.id)); files.dedup_by(|a, b| a.id == b.id); Ok(files) diff --git a/src/service/file_list_dump.rs b/src/service/file_list_dump.rs new file mode 100644 index 0000000000..2f9dc96026 --- /dev/null +++ b/src/service/file_list_dump.rs @@ -0,0 +1,487 @@ +// Copyright 2025 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 arrow::array::{BooleanArray, Int64Array, RecordBatch, StringArray}; +use chrono::Utc; +use config::{ + get_config, + meta::stream::{FileKey, FileListDeleted, StreamStats, StreamType}, + utils::time::hour_micros, +}; +use hashbrown::HashMap; +use infra::{ + errors, + file_list::{FileId, FileRecord, calculate_max_ts_upper_bound}, +}; +use rayon::slice::ParallelSliceMut; + +use super::search::datafusion::exec::prepare_datafusion_context; +use crate::service::search::datafusion::exec::create_parquet_table; + +macro_rules! get_col { + ($var:ident, $name:literal, $typ:ty, $rbatch:ident) => { + let $var = $rbatch + .column_by_name($name) + .unwrap() + .as_any() + .downcast_ref::<$typ>() + .unwrap(); + }; +} + +#[inline] +fn round_down_to_hour(v: i64) -> i64 { + v - (v % hour_micros(1)) +} + +async fn get_dump_files_in_range( + org: &str, + stream: Option<&str>, + range: (i64, i64), + min_id: Option, +) -> Result, errors::Error> { + let start = round_down_to_hour(range.0); + let end = round_down_to_hour(range.1) + hour_micros(1); + + let list = infra::file_list::get_entries_in_range(org, stream, start, end, min_id).await?; + let list = list + .into_iter() + .filter(|f| { + let columns = f.stream.split('/').collect::>(); + if columns.len() != 3 { + return false; + } + let stream_type = StreamType::from(columns[1]); + stream_type == StreamType::Filelist + }) + .collect(); + + Ok(list) +} + +fn record_batch_to_file_record(rb: RecordBatch) -> Vec { + get_col!(id_col, "id", Int64Array, rb); + get_col!(org_col, "org", StringArray, rb); + get_col!(stream_col, "stream", StringArray, rb); + get_col!(date_col, "date", StringArray, rb); + get_col!(file_col, "file", StringArray, rb); + get_col!(deleted_col, "deleted", BooleanArray, rb); + get_col!(flattened_col, "flattened", BooleanArray, rb); + get_col!(min_ts_col, "min_ts", Int64Array, rb); + get_col!(max_ts_col, "max_ts", Int64Array, rb); + get_col!(records_col, "records", Int64Array, rb); + get_col!(original_size_col, "original_size", Int64Array, rb); + get_col!(compressed_size_col, "compressed_size", Int64Array, rb); + get_col!(index_size_col, "index_size", Int64Array, rb); + let mut ret = Vec::with_capacity(rb.num_rows()); + for idx in 0..rb.num_rows() { + let t = FileRecord { + id: id_col.value(idx), + org: org_col.value(idx).to_string(), + stream: stream_col.value(idx).to_string(), + date: date_col.value(idx).to_string(), + file: file_col.value(idx).to_string(), + deleted: deleted_col.value(idx), + flattened: flattened_col.value(idx), + min_ts: min_ts_col.value(idx), + max_ts: max_ts_col.value(idx), + records: records_col.value(idx), + original_size: original_size_col.value(idx), + compressed_size: compressed_size_col.value(idx), + index_size: index_size_col.value(idx), + }; + ret.push(t); + } + ret.par_sort_unstable_by_key(|f| f.id); + ret.dedup_by_key(|f| f.id); + ret +} + +fn record_batch_to_file_id(rb: RecordBatch) -> Vec { + get_col!(id_col, "id", Int64Array, rb); + get_col!(records_col, "records", Int64Array, rb); + get_col!(original_size_col, "original_size", Int64Array, rb); + + let mut ret = Vec::with_capacity(rb.num_rows()); + for idx in 0..rb.num_rows() { + let t = FileId { + id: id_col.value(idx), + records: records_col.value(idx), + original_size: original_size_col.value(idx), + deleted: false, + }; + ret.push(t); + } + ret.par_sort_unstable_by_key(|f| f.id); + ret.dedup_by_key(|f| f.id); + ret +} + +fn record_batch_to_stats(rb: RecordBatch) -> Vec<(String, StreamStats)> { + get_col!(stream_col, "stream", StringArray, rb); + get_col!(min_ts_col, "min_ts", Int64Array, rb); + get_col!(max_ts_col, "max_ts", Int64Array, rb); + get_col!(file_num_col, "file_num", Int64Array, rb); + get_col!(records_col, "records", Int64Array, rb); + get_col!(original_size_col, "original_size", Int64Array, rb); + get_col!(compressed_size_col, "compressed_size", Int64Array, rb); + get_col!(index_size_col, "index_size", Int64Array, rb); + + let mut ret = Vec::with_capacity(rb.num_rows()); + for idx in 0..rb.num_rows() { + let t = StreamStats { + created_at: 0, + doc_time_min: min_ts_col.value(idx), + doc_time_max: max_ts_col.value(idx), + doc_num: records_col.value(idx), + file_num: file_num_col.value(idx), + storage_size: original_size_col.value(idx) as f64, + compressed_size: compressed_size_col.value(idx) as f64, + index_size: index_size_col.value(idx) as f64, + }; + let stream = stream_col.value(idx).to_string(); + ret.push((stream, t)); + } + ret +} + +async fn inner_exec( + trace_id: &str, + partitions: usize, + dump_files: &[FileKey], + query: &str, +) -> Result, errors::Error> { + let schema = super::super::job::FILE_LIST_SCHEMA.clone(); + + let session = config::meta::search::Session { + id: trace_id.to_string(), + storage_type: config::meta::search::StorageType::Memory, + work_group: None, + target_partitions: partitions, + }; + let tbl = create_parquet_table( + &session, + schema.clone(), + dump_files, + HashMap::new(), + false, + None, + None, + vec![], + false, + ) + .await?; + let ctx = prepare_datafusion_context(None, vec![], vec![], false, partitions).await?; + ctx.register_table("file_list", tbl)?; + let df = ctx.sql(query).await?; + let ret = df.collect().await?; + Ok(ret) +} + +async fn exec( + trace_id: &str, + partitions: usize, + dump_files: &[FileKey], + query: &str, +) -> Result, errors::Error> { + let trace_id = format!("{trace_id}-file-list-dump"); + let ret = inner_exec(&trace_id, partitions, dump_files, query).await; + // we always have to clear the files loaded + super::search::datafusion::storage::file_list::clear(&trace_id); + ret +} + +pub async fn query( + trace_id: &str, + org: &str, + stream: &str, + stream_type: StreamType, + range: (i64, i64), + id_hint: Option, +) -> Result, errors::Error> { + let cfg = get_config(); + let stream_key = format!( + "{org}/{}/{org}_{stream_type}_{stream}", + StreamType::Filelist + ); + let db_start = std::time::Instant::now(); + let dump_files = get_dump_files_in_range(org, Some(&stream_key), range, id_hint).await?; + if dump_files.is_empty() { + return Ok(vec![]); + } + let db_time = db_start.elapsed().as_millis(); + + let process_start = std::time::Instant::now(); + let dump_files: Vec<_> = dump_files + .into_iter() + .map(|f| FileKey { + key: format!("files/{}/{}/{}", stream_key, f.date, f.file), + meta: (&f).into(), + deleted: false, + segment_ids: None, + }) + .collect(); + let max_ts_upper_bound = calculate_max_ts_upper_bound(range.1, stream_type); + + let stream_key = format!("{org}/{stream_type}/{stream}"); + let query = format!( + "SELECT * FROM file_list WHERE org= '{org}' AND stream = '{stream_key}' AND max_ts >= {} AND max_ts <= {} AND min_ts <= {};", + range.0, max_ts_upper_bound, range.1 + ); + + let t = exec(trace_id, cfg.limit.cpu_num, &dump_files, &query).await?; + + let ret = t + .into_iter() + .flat_map(record_batch_to_file_record) + .collect(); + let process_time = process_start.elapsed().as_millis(); + + log::info!( + "[FILE_LIST_DUMP: {trace_id}] : getting dump files from db took {db_time} ms, searching for entries took {process_time} ms" + ); + + Ok(ret) +} + +pub async fn get_ids_in_range( + trace_id: &str, + org: &str, + stream: &str, + stream_type: StreamType, + range: (i64, i64), +) -> Result, errors::Error> { + let cfg = get_config(); + let stream_key = format!( + "{org}/{}/{org}_{stream_type}_{stream}", + StreamType::Filelist + ); + let db_start = std::time::Instant::now(); + let dump_files = get_dump_files_in_range(org, Some(&stream_key), range, None).await?; + if dump_files.is_empty() { + return Ok(vec![]); + } + let db_time = db_start.elapsed().as_millis(); + + let process_start = std::time::Instant::now(); + let dump_files: Vec<_> = dump_files + .into_iter() + .map(|f| FileKey { + key: format!("files/{}/{}/{}", stream_key, f.date, f.file), + meta: (&f).into(), + deleted: false, + segment_ids: None, + }) + .collect(); + let max_ts_upper_bound = calculate_max_ts_upper_bound(range.1, stream_type); + + let stream_key = format!("{org}/{stream_type}/{stream}"); + + let query = format!( + "SELECT id,records,original_size FROM file_list WHERE org= '{org}' AND stream = '{stream_key}' AND max_ts >= {} AND max_ts <= {} AND min_ts <= {};", + range.0, max_ts_upper_bound, range.1 + ); + + let t = exec(trace_id, cfg.limit.cpu_num, &dump_files, &query).await?; + + let ret = t.into_iter().flat_map(record_batch_to_file_id).collect(); + let process_time = process_start.elapsed().as_millis(); + + log::info!( + "[FILE_LIST_DUMP: {trace_id}] : getting dump files from db took {db_time} ms, searching for entries took {process_time} ms" + ); + + Ok(ret) +} + +async fn move_and_delete( + org: &str, + stream_type: StreamType, + stream: &str, + range: (i64, i64), +) -> Result<(), errors::Error> { + let stream_key = format!( + "{org}/{}/{org}_{stream_type}_{stream}", + StreamType::Filelist + ); + let list = + infra::file_list::get_entries_in_range(org, Some(&stream_key), range.0, range.1, None) + .await?; + + // for deleting dump files themselves, we explicitly make sure that + // the min and max ts both are in the range given, so to not accidentally delete + // dump files which may have entries outside the range + let dump_files: Vec<_> = get_dump_files_in_range(org, Some(stream), range, None) + .await? + .into_iter() + .filter(|f| f.min_ts >= range.0 && f.max_ts <= range.1) + .collect(); + let del_items: Vec<_> = list + .iter() + .chain(dump_files.iter()) + .map(|f| FileListDeleted { + file: format!("files/{}/{}/{}", stream_key, f.date, f.file), + index_file: false, + flattened: false, + }) + .collect(); + + let mut inserted_into_deleted = false; + + for _ in 0..5 { + if !inserted_into_deleted { + if let Err(e) = + infra::file_list::batch_add_deleted(org, Utc::now().timestamp_micros(), &del_items) + .await + { + log::error!( + "[FILE_LIST_DUMP] batch_add_deleted to db failed, retrying: {}", + e + ); + tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + continue; + } + } + inserted_into_deleted = true; + let items: Vec<_> = list + .iter() + .chain(dump_files.iter()) + .map(|f| FileKey { + key: format!("files/{}/{}/{}", f.stream, f.date, f.file), + meta: f.into(), + deleted: true, + segment_ids: None, + }) + .collect(); + if let Err(e) = infra::file_list::batch_process(&items).await { + log::error!( + "[FILE_LIST_DUMP] batch_delete to db failed, retrying: {}", + e + ); + tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + continue; + } + break; + } + + Ok(()) +} + +pub async fn delete_all_for_stream( + org: &str, + stream_type: StreamType, + stream: &str, +) -> Result<(), errors::Error> { + move_and_delete(org, stream_type, stream, (0, Utc::now().timestamp_micros())).await +} + +pub async fn delete_in_time_range( + org: &str, + stream_type: StreamType, + stream: &str, + range: (i64, i64), +) -> Result<(), errors::Error> { + move_and_delete(org, stream_type, stream, range).await +} + +// we never store deleted file in dump, so we never have to consider deleted in this +pub async fn stats( + org_id: &str, + stream_type: Option, + stream_name: Option<&str>, + pk_value: Option<(i64, i64)>, +) -> Result, errors::Error> { + let cfg = get_config(); + + // in case of dual write, we have no way of de-duping with ids for stats. + // so we simply do not consider dumped files when dual-write is enabled. + if cfg.common.file_list_dump_dual_write { + return Ok(vec![]); + } + + // we can be sure that file with id x will always be dumped in a dump file with id > x + // because file dump is taken after original file entry, and id always increases + let min_id = match pk_value { + Some((min, _)) => Some(min), + _ => None, + }; + + let (field, value, dump_files) = match (stream_type, stream_name) { + (Some(stype), Some(sname)) => { + let stream_key = format!("{org_id}/{}/{org_id}_{stype}_{sname}", StreamType::Filelist); + let dump_files = get_dump_files_in_range( + org_id, + Some(&stream_key), + (0, Utc::now().timestamp_micros()), + min_id, + ) + .await?; + ( + "stream", + format!( + "{}/{}/{}", + org_id, + stream_type.unwrap(), + stream_name.unwrap() + ), + dump_files, + ) + } + _ => { + let dump_files = + get_dump_files_in_range(org_id, None, (0, Utc::now().timestamp_micros()), min_id) + .await?; + ("org", org_id.to_string(), dump_files) + } + }; + + if dump_files.is_empty() { + return Ok(vec![]); + } + + let dump_files: Vec<_> = dump_files + .into_iter() + .map(|f| FileKey { + key: format!("files/{}/{}/{}", f.stream, f.date, f.file), + meta: (&f).into(), + deleted: false, + segment_ids: None, + }) + .collect(); + + let sql = format!( + r#" +SELECT stream, MIN(min_ts) AS min_ts, MAX(max_ts) AS max_ts, COUNT(*)::BIGINT AS file_num, +SUM(records)::BIGINT AS records, SUM(original_size)::BIGINT AS original_size, SUM(compressed_size)::BIGINT AS compressed_size, SUM(index_size)::BIGINT AS index_size +FROM file_list +WHERE {field} = '{value}' + "# + ); + let sql = match pk_value { + None => format!("{} GROUP BY stream", sql), + Some((0, 0)) => format!("{} GROUP BY stream", sql), + Some((min, max)) => { + format!("{} AND id > {} AND id <= {} GROUP BY stream", sql, min, max) + } + }; + + let task_id = tokio::task::try_id() + .map(|id| id.to_string()) + .unwrap_or_else(|| rand::random::().to_string()); + let fake_trace_id = format!("stats_on_dump-{}", task_id); + let t = exec(&fake_trace_id, cfg.limit.cpu_num, &dump_files, &sql).await?; + let ret = t.into_iter().flat_map(record_batch_to_stats).collect(); + Ok(ret) +} diff --git a/src/service/mod.rs b/src/service/mod.rs index 81d02b4a56..04cd786071 100644 --- a/src/service/mod.rs +++ b/src/service/mod.rs @@ -25,6 +25,7 @@ pub mod enrichment; pub mod enrichment_table; pub mod exporter; pub mod file_list; +pub mod file_list_dump; pub mod folders; pub mod functions; pub mod grpc; diff --git a/src/service/promql/search/grpc/mod.rs b/src/service/promql/search/grpc/mod.rs index ef54dffc0b..132d97f655 100644 --- a/src/service/promql/search/grpc/mod.rs +++ b/src/service/promql/search/grpc/mod.rs @@ -114,7 +114,7 @@ pub async fn search( } else { // 1. get max records stream let start_time = std::time::Instant::now(); - let file_list = match get_max_file_list(org_id, &query.query, start, end).await { + let file_list = match get_max_file_list(&trace_id, org_id, &query.query, start, end).await { Ok(v) => v, Err(e) => { log::error!( @@ -242,6 +242,7 @@ pub async fn search_inner( } async fn get_max_file_list( + trace_id: &str, org_id: &str, query: &str, start: i64, @@ -258,6 +259,7 @@ async fn get_max_file_list( let mut max_records = 0; for stream_name in metrics_name { let stream_file_list = crate::service::file_list::query( + trace_id, org_id, &stream_name, StreamType::Metrics, diff --git a/src/service/promql/search/grpc/storage.rs b/src/service/promql/search/grpc/storage.rs index 4289b53088..1617d13a23 100644 --- a/src/service/promql/search/grpc/storage.rs +++ b/src/service/promql/search/grpc/storage.rs @@ -219,6 +219,7 @@ async fn get_file_list( ) -> Result> { let (time_min, time_max) = time_range; let results = match file_list::query( + trace_id, org_id, stream_name, StreamType::Metrics, diff --git a/src/service/search/cluster/flight.rs b/src/service/search/cluster/flight.rs index 4372b92d47..cbe1680e50 100644 --- a/src/service/search/cluster/flight.rs +++ b/src/service/search/cluster/flight.rs @@ -107,6 +107,7 @@ pub async fn search( // 1. get file id list let file_id_list = get_file_id_lists( + trace_id, &sql.org_id, sql.stream_type, &sql.stream_names, @@ -873,6 +874,7 @@ pub async fn register_table(ctx: &SessionContext, sql: &Sql) -> Result<()> { #[tracing::instrument(name = "service:search:cluster:flight:get_file_id_lists", skip_all)] pub async fn get_file_id_lists( + trace_id: &str, org_id: &str, stream_type: StreamType, stream_names: &[TableReference], @@ -893,7 +895,8 @@ pub async fn get_file_id_lists( } // get file list let file_id_list = - crate::service::file_list::query_ids(org_id, stream_type, &name, time_range).await?; + crate::service::file_list::query_ids(trace_id, org_id, stream_type, &name, time_range) + .await?; file_lists.insert(stream.clone(), file_id_list); } Ok(file_lists) diff --git a/src/service/search/datafusion/exec.rs b/src/service/search/datafusion/exec.rs index 1b7d099e24..f86940689c 100644 --- a/src/service/search/datafusion/exec.rs +++ b/src/service/search/datafusion/exec.rs @@ -134,6 +134,9 @@ pub async fn merge_parquet_files( "SELECT MIN({}) AS {}, SUM(count) as count, {} FROM tbl GROUP BY {} ORDER BY {} DESC", TIMESTAMP_COL_NAME, TIMESTAMP_COL_NAME, fields_str, fields_str, TIMESTAMP_COL_NAME ) + } else if stream_type == StreamType::Filelist { + // for file list we do not have timestamp, so we instead sort by min ts of entries + "SELECT * FROM tbl ORDER BY min_ts DESC".to_string() } else { format!("SELECT * FROM tbl ORDER BY {} DESC", TIMESTAMP_COL_NAME) }; diff --git a/src/service/search/grpc/flight.rs b/src/service/search/grpc/flight.rs index 86b6a2f576..6cb6edde6d 100644 --- a/src/service/search/grpc/flight.rs +++ b/src/service/search/grpc/flight.rs @@ -396,7 +396,15 @@ async fn get_file_list_by_ids( idx_file_list: &[cluster_rpc::IdxFileName], ) -> Result<(Vec, usize), Error> { let start = std::time::Instant::now(); - let file_list = crate::service::file_list::query_by_ids(trace_id, ids).await?; + let file_list = crate::service::file_list::query_by_ids( + trace_id, + ids, + org_id, + stream_type, + stream_name, + time_range, + ) + .await?; // if there are any files in idx_files_list, use them to filter the files we got from ids, // otherwise use all the files we got from ids let file_list = if idx_file_list.is_empty() { diff --git a/src/service/search/mod.rs b/src/service/search/mod.rs index 9285155e31..5c07f0ca5a 100644 --- a/src/service/search/mod.rs +++ b/src/service/search/mod.rs @@ -638,6 +638,7 @@ pub async fn search_partition( if !skip_get_file_list && !use_stream_stats_for_partition { let stream_files = crate::service::file_list::query_ids( + trace_id, &sql.org_id, stream_type, &stream_name, diff --git a/src/service/search/super_cluster/follower.rs b/src/service/search/super_cluster/follower.rs index 79e3632f3e..c172da0052 100644 --- a/src/service/search/super_cluster/follower.rs +++ b/src/service/search/super_cluster/follower.rs @@ -127,7 +127,8 @@ pub async fn search( let stream_type = stream.get_stream_type(req.stream_type); // 1. get file id list - let file_id_list = get_file_id_lists(&req.org_id, stream_type, &stream, req.time_range).await?; + let file_id_list = + get_file_id_lists(&trace_id, &req.org_id, stream_type, &stream, req.time_range).await?; let file_id_list_vec = file_id_list.iter().collect::>(); let file_id_list_took = start.elapsed().as_millis() as usize; @@ -301,6 +302,7 @@ pub async fn search( skip_all )] pub async fn get_file_id_lists( + trace_id: &str, org_id: &str, stream_type: StreamType, stream: &TableReference, @@ -316,8 +318,14 @@ pub async fn get_file_id_lists( time_range = Some((start, end)); } } - let file_id_list = - crate::service::file_list::query_ids(org_id, stream_type, &stream_name, time_range).await?; + let file_id_list = crate::service::file_list::query_ids( + trace_id, + org_id, + stream_type, + &stream_name, + time_range, + ) + .await?; Ok(file_id_list) }