starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-05 16:10:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement file list dump (#6467)

Browse Source 
This implements the file list dump feature, where we dump the file list
by hour after compaction into parquet files and use those files while
searching through files lists. This also adds following variables -
- ZO_FILE_LIST_DUMP_ENABLED : if this dumping is enabled or not, default
is false
- ZO_FILE_LIST_DUMP_DUAL_WRITE : should the file list entries be kept in
the file list even after dumping. default is true, recommended is false.
Default can be changed after more verification.
- ZO_FILE_LIST_DUMP_MIN_HOUR : number of hours in past, only files older
than this will be considered for dumping, to avoid hot files issues.

The dump files entries themselves are stored in the file_list table as
was decided, so they also become a part of the normal compactor flow.
However, there is check to ensure these files are not dumped.

---------

Signed-off-by: Yashodhan Joshi <yjdoc2@gmail.com>
Co-authored-by: Yashodhan Joshi <yjdoc2@gmail.com>
Co-authored-by: Hengfei Yang <hengfei.yang@gmail.com>
This commit is contained in:
Yashodhan Joshi
2025-04-24 11:42:43 +05:30
committed by GitHub
co-authored by Yashodhan Joshi Hengfei Yang
parent 072a8aa9cb
commit 1f575799fc
23 changed files with 1860 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cli/basic/test.rs
+8 -3
View File
@@ -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::<i64>().unwrap_or(5) * 1024 * 1024 * 1024;
src/config/src/config.rs
+13
View File
@@ -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(())
}
src/config/src/meta/stream.rs
+21
View File
@@ -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 {
src/infra/src/file_list/mod.rs
+46 -5
View File
@@ -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<bool>,
) -> Result<Vec<(String, FileMeta)>>;
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<i64>,
) -> Result<Vec<FileRecord>>;
async fn get_pending_dump_jobs(&self) -> Result<Vec<(i64, String, String, i64)>>;
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<Vec<(String, FileMeta)>> {
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<i64>,
) -> Result<Vec<FileRecord>> {
CLIENT
.get_entries_in_range(org, stream, start_time, end_time, min_id)
.await
}
#[inline]
pub async fn get_pending_dump_jobs() -> Result<Vec<(i64, String, String, i64)>> {
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,
src/infra/src/file_list/mysql.rs
+195 -9
View File
@@ -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::<Vec<String>>()
.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::<Vec<_>>()
.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<i64>,
) -> Result<Vec<super::FileRecord>> {
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<Vec<(i64, String, String, i64)>> {
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",
src/infra/src/file_list/postgres.rs
+198 -10
View File
@@ -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::<Vec<String>>()
.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::<Vec<_>>()
.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<i64>,
) -> Result<Vec<super::FileRecord>> {
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<Vec<(i64, String, String, i64)>> {
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",
src/infra/src/file_list/sqlite.rs
+184 -8
View File
@@ -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::<Vec<String>>()
.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::<Vec<_>>()
.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<i64>,
) -> Result<Vec<super::FileRecord>> {
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<Vec<(i64, String, String, i64)>> {
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",
src/job/file_list_dump.rs
+450
View File
@@ -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 <http://www.gnu.org/licenses/>.
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<Arc<Schema>> = 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<JobInfo>,
receiver: Arc<Mutex<Receiver<JobInfo>>>,
}
impl DownloadQueue {
fn new(sender: Sender<JobInfo>, receiver: Arc<Mutex<Receiver<JobInfo>>>) -> 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<DownloadQueue> = Lazy::new(|| {
let (tx, rx) = tokio::sync::mpsc::channel::<JobInfo>(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<RwLock<HashSet<i64>>> = 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::<Vec<_>>();
// 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::<Vec<&str>>();
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<Schema>,
buf: &mut Vec<u8>,
) -> Result<AsyncArrowWriter<&mut Vec<u8>>, 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<FileRecord>) -> Result<RecordBatch, anyhow::Error> {
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<FileRecord>,
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<i64> = 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<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(())
}
src/job/mod.rs
+6
View File
@@ -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
src/service/compact/merge.rs
+4 -3
View File
@@ -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: {}",
src/service/compact/retention.rs
+17
View File
@@ -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::<u64>().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,
src/service/compact/stats.rs
+12 -1
View File
@@ -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<Option<(i64, i64)>, anyhow::Error> {
// get last offset
@@ -49,13 +52,21 @@ pub async fn update_stats_from_file_list() -> Result<Option<(i64, i64)>, 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;
src/service/db/file_list/mod.rs
+2 -2
View File
@@ -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::<Vec<&str>>();
let org_id = columns[0];
src/service/file_list.rs
+184 -8
View File
@@ -13,12 +13,13 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
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<Vec<FileKey>> {
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::<HashSet<_>>();
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<Vec<FileKey>> {
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<Vec<FileKey>> {
pub async fn query_by_ids(
trace_id: &str,
ids: &[i64],
org: &str,
stream_type: StreamType,
stream: &str,
time_range: Option<(i64, i64)>,
) -> Result<Vec<FileKey>> {
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<Vec<FileKey>> {
)
);
// 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::<HashSet<_>>();
let db_ids = db_files.iter().map(|entry| entry.0).collect::<Vec<_>>();
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<Vec<FileKey>> {
// 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<Vec<FileKey>> {
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<Vec<file_list::FileId>> {
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::<HashSet<_>>();
let db_ids = files.iter().map(|f| f.id).collect::<Vec<_>>();
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)
src/service/file_list_dump.rs
+487
View File
@@ -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 <http://www.gnu.org/licenses/>.
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<i64>,
) -> Result<Vec<FileRecord>, 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::<Vec<&str>>();
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<FileRecord> {
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<FileId> {
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<Vec<RecordBatch>, 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<Vec<RecordBatch>, 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<i64>,
) -> Result<Vec<FileRecord>, 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<Vec<FileId>, 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<StreamType>,
stream_name: Option<&str>,
pk_value: Option<(i64, i64)>,
) -> Result<Vec<(String, StreamStats)>, 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::<u64>().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)
}
src/service/mod.rs
+1
View File
@@ -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;
src/service/promql/search/grpc/mod.rs
+3 -1
View File
@@ -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,
src/service/promql/search/grpc/storage.rs
+1
View File
@@ -219,6 +219,7 @@ async fn get_file_list(
) -> Result<Vec<FileKey>> {
let (time_min, time_max) = time_range;
let results = match file_list::query(
trace_id,
org_id,
stream_name,
StreamType::Metrics,
src/service/search/cluster/flight.rs
+4 -1
View File
@@ -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)
src/service/search/datafusion/exec.rs
+3
View File
@@ -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)
};
src/service/search/grpc/flight.rs
+9 -1
View File
@@ -396,7 +396,15 @@ async fn get_file_list_by_ids(
idx_file_list: &[cluster_rpc::IdxFileName],
) -> Result<(Vec<FileKey>, 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() {
src/service/search/mod.rs
+1
View File
@@ -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,
src/service/search/super_cluster/follower.rs
+11 -3
View File
@@ -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::<Vec<_>>();
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)
}