fix: improve stats from file_list_dump (#7471)

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 =

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 {

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 {

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 {

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<StreamType>,
     stream_name: Option<&str>,
     pk_value: Option<(i64, i64)>,
+) -> Result<Vec<(String, StreamStats)>, 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<Vec<(String, StreamStats)>, 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, 

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<ObjectMeta>> {
         let key = prefix.unwrap().to_string();
         let objects = match super::file_list::get(&key) {

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

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(