feat: add records for search partition (#2589)

src/common/meta/search.rs

@@ -267,6 +267,7 @@ pub struct SearchPartitionRequest {
 #[derive(Clone, Debug, Default, Serialize, Deserialize, ToSchema)]
 pub struct SearchPartitionResponse {
     pub file_num: usize,
+    pub records: usize,
     pub original_size: usize,
     pub compressed_size: usize,
     pub partitions: Vec<[i64; 2]>,

src/ingester/src/immutable.rs

@@ -120,6 +120,10 @@ pub(crate) async fn persist() -> Result<()> {
                     drop(permit);
                     return Ok(None);
                 };
+                log::info!(
+                    "[INGESTER:WAL] start persist file: {}",
+                    path.to_string_lossy(),
+                );
                 // persist entry to local disk
                 let ret = immutable.persist(&path).await;
                 drop(r);
@@ -135,7 +139,7 @@ pub(crate) async fn persist() -> Result<()> {
     for task in tasks {
         if let Some((path, json_size, arrow_size)) = task? {
             log::info!(
-                "[INGESTER:WAL] persist file: {}, json_size: {}, arrow_size: {}",
+                "[INGESTER:WAL] done persist file: {}, json_size: {}, arrow_size: {}",
                 path.to_string_lossy(),
                 json_size,
                 arrow_size

src/ingester/src/writer.rs

@@ -197,10 +197,24 @@ impl Writer {
             let key = self.key.clone();
             let path = old_wal.path().clone();
             tokio::task::spawn(async move {
-                IMMUTABLES
-                    .write()
-                    .await
-                    .insert(path, immutable::Immutable::new(thread_id, key, old_mem));
+                log::info!(
+                    "[INGESTER:WAL] start add to IMMUTABLES, file: {}/{}/{}/{}.wal",
+                    thread_id,
+                    &key.org_id,
+                    &key.stream_type,
+                    wal_id
+                );
+                IMMUTABLES.write().await.insert(
+                    path,
+                    immutable::Immutable::new(thread_id, key.clone(), old_mem),
+                );
+                log::info!(
+                    "[INGESTER:WAL] done add to IMMUTABLES, file: {}/{}/{}/{}.wal",
+                    thread_id,
+                    &key.org_id,
+                    &key.stream_type,
+                    wal_id
+                );
             });
         }
 

src/service/search/mod.rs

@@ -107,10 +107,21 @@ pub async fn search_partition(
     let nodes = cluster::get_cached_online_querier_nodes().unwrap_or_default();
     let cpu_cores = nodes.iter().map(|n| n.cpu_num).sum::<u64>() as usize;
 
+    let (records, original_size, compressed_size) =
+        files
+            .iter()
+            .fold((0, 0, 0), |(records, original_size, compressed_size), f| {
+                (
+                    records + f.meta.records,
+                    original_size + f.meta.original_size,
+                    compressed_size + f.meta.compressed_size,
+                )
+            });
     let mut resp = search::SearchPartitionResponse {
         file_num: files.len(),
-        original_size: files.iter().map(|f| f.meta.original_size).sum::<i64>() as usize,
-        compressed_size: files.iter().map(|f| f.meta.compressed_size).sum::<i64>() as usize,
+        records: records as usize,
+        original_size: original_size as usize,
+        compressed_size: compressed_size as usize,
         partitions: vec![],
     };
     let mut total_secs = resp.original_size / CONFIG.limit.query_group_base_speed / cpu_cores;