From ff29e0c9c09bc00a70cf5e678db9ebec1518ef7a Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Thu, 18 Sep 2025 13:40:47 +0800 Subject: [PATCH] fix: stream stats min_ts error (#8509) --- src/infra/src/file_list/mysql.rs | 10 ++- src/infra/src/file_list/postgres.rs | 20 +++--- src/infra/src/file_list/sqlite.rs | 20 +++--- src/infra/src/queue/mod.rs | 3 +- src/infra/src/queue/nop.rs | 96 ----------------------------- 5 files changed, 31 insertions(+), 118 deletions(-) delete mode 100644 src/infra/src/queue/nop.rs diff --git a/src/infra/src/file_list/mysql.rs b/src/infra/src/file_list/mysql.rs index 07b075ffe2..372fc3734d 100644 --- a/src/infra/src/file_list/mysql.rs +++ b/src/infra/src/file_list/mysql.rs @@ -1115,8 +1115,12 @@ INSERT INTO stream_stats r#" UPDATE stream_stats SET file_num = file_num + ?, - min_ts = CASE WHEN ? = 0 THEN min_ts ELSE ? END, - max_ts = CASE WHEN ? = 0 THEN max_ts ELSE ? END, + min_ts = CASE WHEN ? = 0 THEN min_ts + WHEN min_ts = 0 OR ? < min_ts THEN ? + ELSE min_ts END, + max_ts = CASE WHEN ? = 0 THEN max_ts + WHEN max_ts = 0 OR ? > max_ts THEN ? + ELSE max_ts END, records = records + ?, original_size = original_size + ?, compressed_size = compressed_size + ?, @@ -1127,6 +1131,8 @@ UPDATE stream_stats .bind(stats.file_num) .bind(stats.doc_time_min) .bind(stats.doc_time_min) + .bind(stats.doc_time_min) + .bind(stats.doc_time_max) .bind(stats.doc_time_max) .bind(stats.doc_time_max) .bind(stats.doc_num) diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs index 08b86ea7d4..823fe6affe 100644 --- a/src/infra/src/file_list/postgres.rs +++ b/src/infra/src/file_list/postgres.rs @@ -1059,19 +1059,21 @@ INSERT INTO stream_stats r#" UPDATE stream_stats SET file_num = file_num + $1, - min_ts = CASE WHEN $2 = 0 THEN min_ts ELSE $3 END, - max_ts = CASE WHEN $4 = 0 THEN max_ts ELSE $5 END, - records = records + $6, - original_size = original_size + $7, - compressed_size = compressed_size + $8, - index_size = index_size + $9 - WHERE stream = $10; + min_ts = CASE WHEN $2 = 0 THEN min_ts + WHEN min_ts = 0 OR $2 < min_ts THEN $2 + ELSE min_ts END, + max_ts = CASE WHEN $3 = 0 THEN max_ts + WHEN max_ts = 0 OR $3 > max_ts THEN $3 + ELSE max_ts END, + records = records + $4, + original_size = original_size + $5, + compressed_size = compressed_size + $6, + index_size = index_size + $7 + WHERE stream = $8; "#, ) .bind(stats.file_num) .bind(stats.doc_time_min) - .bind(stats.doc_time_min) - .bind(stats.doc_time_max) .bind(stats.doc_time_max) .bind(stats.doc_num) .bind(stats.storage_size as i64) diff --git a/src/infra/src/file_list/sqlite.rs b/src/infra/src/file_list/sqlite.rs index 04d4d262ec..eee396a620 100644 --- a/src/infra/src/file_list/sqlite.rs +++ b/src/infra/src/file_list/sqlite.rs @@ -880,19 +880,21 @@ INSERT INTO stream_stats r#" UPDATE stream_stats SET file_num = file_num + $1, - min_ts = CASE WHEN $2 = 0 THEN min_ts ELSE $3 END, - max_ts = CASE WHEN $4 = 0 THEN max_ts ELSE $5 END, - records = records + $6, - original_size = original_size + $7, - compressed_size = compressed_size + $8, - index_size = index_size + $9 - WHERE stream = $10; + min_ts = CASE WHEN $2 = 0 THEN min_ts + WHEN min_ts = 0 OR $2 < min_ts THEN $2 + ELSE min_ts END, + max_ts = CASE WHEN $3 = 0 THEN max_ts + WHEN max_ts = 0 OR $3 > max_ts THEN $3 + ELSE max_ts END, + records = records + $4, + original_size = original_size + $5, + compressed_size = compressed_size + $6, + index_size = index_size + $7 + WHERE stream = $8; "#, ) .bind(stats.file_num) .bind(stats.doc_time_min) - .bind(stats.doc_time_min) - .bind(stats.doc_time_max) .bind(stats.doc_time_max) .bind(stats.doc_num) .bind(stats.storage_size as i64) diff --git a/src/infra/src/queue/mod.rs b/src/infra/src/queue/mod.rs index cfa8093819..aa1b974c48 100644 --- a/src/infra/src/queue/mod.rs +++ b/src/infra/src/queue/mod.rs @@ -23,7 +23,6 @@ use tokio::sync::{OnceCell, mpsc}; use crate::errors::{Error, Result}; pub mod nats; -pub mod nop; #[derive(Debug)] pub enum RetentionPolicy { @@ -56,7 +55,7 @@ pub async fn init() -> Result<()> { async fn default() -> Box { match config::get_config().common.queue_store.as_str().into() { MetaStore::Nats => Box::::default(), - _ => Box::::default(), + _ => Box::::default(), } } diff --git a/src/infra/src/queue/nop.rs b/src/infra/src/queue/nop.rs deleted file mode 100644 index 1d67d4d74c..0000000000 --- a/src/infra/src/queue/nop.rs +++ /dev/null @@ -1,96 +0,0 @@ -// Copyright 2025 OpenObserve Inc. -// -// This program is free software: you can redistribute it and/or modify -// it under the terms of the GNU Affero General Public License as published by -// the Free Software Foundation, either version 3 of the License, or -// (at your option) any later version. -// -// This program is distributed in the hope that it will be useful -// but WITHOUT ANY WARRANTY; without even the implied warranty of -// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -// GNU Affero General Public License for more details. -// -// You should have received a copy of the GNU Affero General Public License -// along with this program. If not, see . - -use std::sync::Arc; - -use async_trait::async_trait; -use bytes::Bytes; -use tokio::sync::mpsc; - -use crate::{ - errors::*, - queue::{DeliverPolicy, RetentionPolicy}, -}; - -pub async fn init() -> Result<()> { - Ok(()) -} - -pub struct NopQueue {} - -impl NopQueue { - pub fn new(_prefix: &str) -> Self { - Self {} - } - - pub fn super_cluster() -> Self { - Self::new("") - } -} - -impl Default for NopQueue { - fn default() -> Self { - Self::new("") - } -} - -#[async_trait] -impl super::Queue for NopQueue { - async fn create(&self, _topic: &str) -> Result<()> { - self.create_with_retention_policy(_topic, RetentionPolicy::Limits) - .await - } - - async fn create_with_retention_policy( - &self, - _topic: &str, - _retention_policy: RetentionPolicy, - ) -> Result<()> { - let max_age = config::get_config().nats.queue_max_age; // days - let max_age_secs = std::time::Duration::from_secs(max_age * 24 * 60 * 60); - self.create_with_retention_policy_and_max_age(_topic, _retention_policy, max_age_secs) - .await - } - - async fn create_with_max_age(&self, _topic: &str, _max_age: std::time::Duration) -> Result<()> { - self.create_with_retention_policy_and_max_age(_topic, RetentionPolicy::Limits, _max_age) - .await - } - - async fn create_with_retention_policy_and_max_age( - &self, - _topic: &str, - _retention_policy: RetentionPolicy, - _max_age: std::time::Duration, - ) -> Result<()> { - todo!() - } - - async fn publish(&self, _topic: &str, _value: Bytes) -> Result<()> { - todo!() - } - - async fn consume( - &self, - _topic: &str, - _deliver_policy: Option, - ) -> Result>> { - todo!() - } - - async fn purge(&self, _topic: &str, _sequence: usize) -> Result<()> { - todo!() - } -}