diff --git a/src/common/meta/middleware_data.rs b/src/common/meta/middleware_data.rs index 2df07d1e21..99acfc713d 100644 --- a/src/common/meta/middleware_data.rs +++ b/src/common/meta/middleware_data.rs @@ -207,8 +207,8 @@ mod tests { for query in &[query_string_oo_tags, query_string_o2_tags] { // Create a mock ServiceRequest with the query string - let req = actix_web::test::TestRequest::with_uri(&format!("/path?{query}")) - .to_srv_request(); + let req = + actix_web::test::TestRequest::with_uri(&format!("/path?{query}")).to_srv_request(); // Call the from_query function let query_data = diff --git a/src/infra/src/file_list/mysql.rs b/src/infra/src/file_list/mysql.rs index f054cb85f0..7bf8ad61e7 100644 --- a/src/infra/src/file_list/mysql.rs +++ b/src/infra/src/file_list/mysql.rs @@ -949,12 +949,13 @@ SELECT date } else { ("org", org_id.to_string()) }; + let file_list_stream = format!("{org_id}/file_list/"); let mut sql = format!( r#" SELECT stream, MIN(min_ts) AS min_ts, MAX(max_ts) AS max_ts, CAST(COUNT(*) AS SIGNED) AS file_num, CAST(SUM(records) AS SIGNED) AS records, CAST(SUM(original_size) AS SIGNED) AS original_size, CAST(SUM(compressed_size) AS SIGNED) AS compressed_size, CAST(SUM(index_size) AS SIGNED) AS index_size FROM file_list - WHERE {field} = '{value}' + WHERE {field} = '{value}' AND stream NOT LIKE '{file_list_stream}%' "#, ); if deleted { diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs index e66ebc6047..ce60df9aaa 100644 --- a/src/infra/src/file_list/postgres.rs +++ b/src/infra/src/file_list/postgres.rs @@ -889,12 +889,13 @@ SELECT date } else { ("org", org_id.to_string()) }; + let file_list_stream = format!("{org_id}/file_list/"); let mut 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}' + WHERE {field} = '{value}' AND stream NOT LIKE '{file_list_stream}%' "# ); if deleted { diff --git a/src/infra/src/file_list/sqlite.rs b/src/infra/src/file_list/sqlite.rs index 6c4cb481b2..ba147b0a42 100644 --- a/src/infra/src/file_list/sqlite.rs +++ b/src/infra/src/file_list/sqlite.rs @@ -725,11 +725,12 @@ SELECT date } else { ("org", org_id.to_string()) }; + let file_list_stream = format!("{org_id}/file_list/"); let mut sql = format!( r#" SELECT stream, MIN(min_ts) as min_ts, MAX(max_ts) as max_ts, COUNT(*) as file_num, SUM(records) as records, SUM(original_size) as original_size, SUM(compressed_size) as compressed_size, SUM(index_size) as index_size FROM file_list - WHERE {field} = '{value}' + WHERE {field} = '{value}' AND stream NOT LIKE '{file_list_stream}%' "#, ); if deleted { diff --git a/src/service/file_list_dump.rs b/src/service/file_list_dump.rs index 76d0e99dd2..1acc2ac2b2 100644 --- a/src/service/file_list_dump.rs +++ b/src/service/file_list_dump.rs @@ -17,7 +17,7 @@ use arrow::array::{BooleanArray, Int64Array, RecordBatch, StringArray}; use chrono::Utc; use config::{ get_config, - meta::stream::{FileKey, FileListDeleted, StreamStats, StreamType}, + meta::stream::{ALL_STREAM_TYPES, FileKey, FileListDeleted, StreamStats, StreamType}, utils::time::hour_micros, }; use hashbrown::HashMap; @@ -393,6 +393,38 @@ pub async fn stats( stream_type: Option, stream_name: Option<&str>, pk_value: Option<(i64, i64)>, +) -> Result, errors::Error> { + let stream_types = match stream_type { + Some(stype) => vec![stype], + None => ALL_STREAM_TYPES.to_vec(), + }; + + let mut ret = HashMap::new(); + for stream_type in stream_types { + if stream_type == StreamType::Filelist { + continue; + } + let stream_names = match stream_name { + Some(name) => vec![name.to_string()], + None => crate::service::db::schema::list_streams_from_cache(org_id, stream_type).await, + }; + for stream_name in stream_names { + let stats = stats_inner(org_id, stream_type, &stream_name, pk_value).await?; + ret.extend(stats); + log::debug!( + "[FILE_LIST_DUMP] stats for {org_id}/{stream_type}/{stream_name}: pk: {:?}", + pk_value, + ); + } + } + Ok(ret.into_iter().collect()) +} + +async fn stats_inner( + org_id: &str, + stream_type: StreamType, + stream_name: &str, + pk_value: Option<(i64, i64)>, ) -> Result, errors::Error> { let cfg = get_config(); @@ -409,34 +441,17 @@ pub async fn stats( _ => 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) - } - }; + let stream_key = format!( + "{org_id}/{}/{org_id}_{stream_type}_{stream_name}", + StreamType::Filelist + ); + let dump_files = get_dump_files_in_range( + org_id, + Some(&stream_key), + (0, Utc::now().timestamp_micros()), + min_id, + ) + .await?; if dump_files.is_empty() { return Ok(vec![]); @@ -444,6 +459,8 @@ pub async fn stats( let dump_files: Vec<_> = dump_files.iter().map(|f| f.into()).collect(); + let field = "stream"; + let value = format!("{org_id}/{stream_type}/{stream_name}"); let sql = format!( r#" SELECT stream, MIN(min_ts) AS min_ts, MAX(max_ts) AS max_ts, COUNT(*)::BIGINT AS file_num, diff --git a/src/service/search/datafusion/storage/memory.rs b/src/service/search/datafusion/storage/memory.rs index 24ce5f8f63..c1825718d8 100644 --- a/src/service/search/datafusion/storage/memory.rs +++ b/src/service/search/datafusion/storage/memory.rs @@ -64,7 +64,6 @@ impl ObjectStore for FS { infra::cache::storage::head(&account, &location).await } - #[tracing::instrument(name = "datafusion::storage::memory::list", skip_all)] fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result> { let key = prefix.unwrap().to_string(); let objects = match super::file_list::get(&key) { diff --git a/src/service/search/partition.rs b/src/service/search/partition.rs index 0c95419d9d..ebf261a8a1 100644 --- a/src/service/search/partition.rs +++ b/src/service/search/partition.rs @@ -507,10 +507,7 @@ mod tests { .unwrap() .with_timezone(&chrono::Local); println!("start: {start_time}, end: {end_time}"); - println!( - "Partition {}: {partition_start} - {partition_end}", - i + 1 - ); + println!("Partition {}: {partition_start} - {partition_end}", i + 1); } // Verify histogram alignment diff --git a/src/service/websocket_events/utils.rs b/src/service/websocket_events/utils.rs index 2355c58501..ee3d4291ae 100644 --- a/src/service/websocket_events/utils.rs +++ b/src/service/websocket_events/utils.rs @@ -28,6 +28,7 @@ use crate::handler::http::request::search::error_utils::map_error_to_http_respon #[cfg(feature = "enterprise")] pub mod enterprise_utils { use config::meta::stream::StreamType; + use crate::common::meta; pub async fn check_permissions(