fix: reset stream stats with limit (openobserve#7001)

Author: hengfeiyang

src/infra/src/file_list/mysql.rs

@@ -971,13 +971,16 @@ UPDATE stream_stats
             let start = std::time::Instant::now();
             let limit = config::get_config().limit.calculate_stats_step_limit;
             loop {
-                match sqlx::query("DELETE FROM file_list WHERE id IN (SELECT id FROM file_list WHERE deleted IS TRUE AND id <= ? LIMIT ?);")
+                let start = std::time::Instant::now();
+                match sqlx::query("DELETE FROM file_list WHERE id IN (SELECT id FROM file_list WHERE org = ? AND deleted IS TRUE AND id <= ? LIMIT ?);")
+                    .bind(org_id)
                     .bind(max_id)
                     .bind(limit)
                     .execute(&mut *tx)
                     .await
                 {
                     Ok(v) => {
+                        log::debug!("[MYSQL] delete file list rows affected: {}, took: {} ms", v.rows_affected(), start.elapsed().as_millis());
                         if v.rows_affected() == 0 {
                             break;
                         }

src/infra/src/file_list/postgres.rs

@@ -988,13 +988,16 @@ UPDATE stream_stats
             let start = std::time::Instant::now();
             let limit = config::get_config().limit.calculate_stats_step_limit;
             loop {
-                match sqlx::query("DELETE FROM file_list WHERE id IN (SELECT id FROM file_list WHERE deleted IS TRUE AND id <= $1 LIMIT $2);")
+                let start = std::time::Instant::now();
+                match sqlx::query("DELETE FROM file_list WHERE id IN (SELECT id FROM file_list WHERE org = $1 AND deleted IS TRUE AND id <= $2 LIMIT $3);")
+                    .bind(org_id)
                     .bind(max_id)
                     .bind(limit)
                     .execute(&mut *tx)
                     .await
                 {
                     Ok(v) => {
+                        log::debug!("[POSTGRES] delete file list rows affected: {}, took: {} ms", v.rows_affected(), start.elapsed().as_millis());
                         if v.rows_affected() == 0 {
                             break;
                         }

src/infra/src/file_list/sqlite.rs

@@ -869,13 +869,16 @@ UPDATE stream_stats
         if let Some((_min_id, max_id)) = pk_value {
             let limit = config::get_config().limit.calculate_stats_step_limit;
             loop {
-                match sqlx::query("DELETE FROM file_list WHERE id IN (SELECT id FROM file_list WHERE deleted IS TRUE AND id <= $1 LIMIT $2);")
+                let start = std::time::Instant::now();
+                match sqlx::query("DELETE FROM file_list WHERE id IN (SELECT id FROM file_list WHERE org = $1 AND deleted IS TRUE AND id <= $2 LIMIT $3);")
+                    .bind(org_id)
                     .bind(max_id)
                     .bind(limit)
                     .execute(&mut *tx)
                     .await
                 {
                     Ok(v) => {
+                        log::debug!("[SQLITE] delete file list rows affected: {}, took: {} ms", v.rows_affected(), start.elapsed().as_millis());
                         if v.rows_affected() == 0 {
                             break;
                         }