Merge branch 'main' into fix-join-order

haohuaijin · web-flow · commit 6919dd9d3829 · 2025-10-22T16:27:41.000+08:00
diff --git a/src/handler/http/request/status/mod.rs b/src/handler/http/request/status/mod.rs
@@ -505,8 +505,6 @@ async fn get_stream_schema_status() -> (usize, usize, usize) {
     drop(r);
     let r = STREAM_SCHEMAS_LATEST.read().await;
     for (key, schema) in r.iter() {
-        stream_num += 1;
-        stream_schema_num += 1;
         mem_size += std::mem::size_of::<String>() + key.len();
         mem_size += schema.size();
     }
diff --git a/src/infra/src/file_list/mod.rs b/src/infra/src/file_list/mod.rs
@@ -110,12 +110,6 @@ pub trait FileList: Sync + Send + 'static {
         limit: i64,
     ) -> Result<Vec<FileListDeleted>>;
     async fn list_deleted(&self) -> Result<Vec<FileListDeleted>>;
-    async fn get_min_ts(
-        &self,
-        org_id: &str,
-        stream_type: StreamType,
-        stream_name: &str,
-    ) -> Result<i64>;
     async fn get_min_date(
         &self,
         org_id: &str,
@@ -355,11 +349,6 @@ pub async fn list_deleted() -> Result<Vec<FileListDeleted>> {
     CLIENT.list_deleted().await
 }
 
-#[inline]
-pub async fn get_min_ts(org_id: &str, stream_type: StreamType, stream_name: &str) -> Result<i64> {
-    CLIENT.get_min_ts(org_id, stream_type, stream_name).await
-}
-
 #[inline]
 pub async fn get_min_date(
     org_id: &str,
diff --git a/src/infra/src/file_list/mysql.rs b/src/infra/src/file_list/mysql.rs
@@ -872,28 +872,6 @@ SELECT date
             .collect())
     }
 
-    async fn get_min_ts(
-        &self,
-        org_id: &str,
-        stream_type: StreamType,
-        stream_name: &str,
-    ) -> Result<i64> {
-        let stream_key = format!("{org_id}/{stream_type}/{stream_name}");
-        let min_ts = config::utils::time::BASE_TIME.timestamp_micros();
-        let pool = CLIENT_RO.clone();
-        DB_QUERY_NUMS
-            .with_label_values(&["select", "file_list"])
-            .inc();
-        let ret: Option<i64> = sqlx::query_scalar(
-            r#"SELECT MIN(min_ts) AS num FROM file_list WHERE stream = ? AND min_ts > ?;"#,
-        )
-        .bind(stream_key)
-        .bind(min_ts)
-        .fetch_one(&pool)
-        .await?;
-        Ok(ret.unwrap_or_default())
-    }
-
     async fn get_min_date(
         &self,
         org_id: &str,
diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs
@@ -815,28 +815,6 @@ SELECT date
             .collect())
     }
 
-    async fn get_min_ts(
-        &self,
-        org_id: &str,
-        stream_type: StreamType,
-        stream_name: &str,
-    ) -> Result<i64> {
-        let stream_key = format!("{org_id}/{stream_type}/{stream_name}");
-        let min_ts = config::utils::time::BASE_TIME.timestamp_micros();
-        let pool = CLIENT_RO.clone();
-        DB_QUERY_NUMS
-            .with_label_values(&["select", "file_list"])
-            .inc();
-        let ret: Option<i64> = sqlx::query_scalar(
-            r#"SELECT MIN(min_ts)::BIGINT AS num FROM file_list WHERE stream = $1 AND min_ts > $2;"#,
-        )
-        .bind(stream_key)
-        .bind(min_ts)
-        .fetch_one(&pool)
-        .await?;
-        Ok(ret.unwrap_or_default())
-    }
-
     async fn get_min_date(
         &self,
         org_id: &str,
diff --git a/src/infra/src/file_list/sqlite.rs b/src/infra/src/file_list/sqlite.rs
@@ -661,25 +661,6 @@ SELECT date
             .collect())
     }
 
-    async fn get_min_ts(
-        &self,
-        org_id: &str,
-        stream_type: StreamType,
-        stream_name: &str,
-    ) -> Result<i64> {
-        let stream_key = format!("{org_id}/{stream_type}/{stream_name}");
-        let min_ts = config::utils::time::BASE_TIME.timestamp_micros();
-        let pool = CLIENT_RO.clone();
-        let ret: Option<i64> = sqlx::query_scalar(
-            r#"SELECT MIN(min_ts) AS num FROM file_list WHERE stream = $1 AND min_ts > $2;"#,
-        )
-        .bind(stream_key)
-        .bind(min_ts)
-        .fetch_one(&pool)
-        .await?;
-        Ok(ret.unwrap_or_default())
-    }
-
     async fn get_min_date(
         &self,
         org_id: &str,
diff --git a/src/service/compact/retention.rs b/src/service/compact/retention.rs
@@ -430,12 +430,8 @@ pub async fn delete_by_date(
 
     // update stream stats retention time
     let mut stats = cache::stats::get_stream_stats(org_id, stream_name, stream_type);
-    let mut min_ts = infra_file_list::get_min_ts(org_id, stream_type, stream_name)
-        .await
-        .unwrap_or_default();
-    if min_ts == 0 {
-        min_ts = stats.doc_time_min;
-    };
+    // we use date_end as the new min doc time
+    let min_ts = date_end.timestamp_micros();
     infra_file_list::reset_stream_stats_min_ts(
         org_id,
         format!("{org_id}/{stream_type}/{stream_name}").as_str(),

Original file line number	Diff line number	Diff line change
`@@ -505,8 +505,6 @@ async fn get_stream_schema_status() -> (usize, usize, usize) {`
`505`	`505`	`drop(r);`
`506`	`506`	`let r = STREAM_SCHEMAS_LATEST.read().await;`
`507`	`507`	`for (key, schema) in r.iter() {`
`508`		`- stream_num += 1;`
`509`		`- stream_schema_num += 1;`
`510`	`508`	`mem_size += std::mem::size_of::<String>() + key.len();`
`511`	`509`	`mem_size += schema.size();`
`512`	`510`	`}`