starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-04 16:09:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: stream stats min_ts error (#8509)

Browse Source
This commit is contained in:
Hengfei Yang
2025-09-18 13:40:47 +08:00
committed by GitHub
parent c2e8d73e65
commit ff29e0c9c0
5 changed files with 31 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/infra/src/file_list/mysql.rs
+8 -2
View File
@@ -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)
src/infra/src/file_list/postgres.rs
+11 -9
View File
@@ -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)
src/infra/src/file_list/sqlite.rs
+11 -9
View File
@@ -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)
src/infra/src/queue/mod.rs
+1 -2
View File
@@ -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<dyn Queue> {
match config::get_config().common.queue_store.as_str().into() {
MetaStore::Nats => Box::<nats::NatsQueue>::default(),
_ => Box::<nop::NopQueue>::default(),
_ => Box::<nats::NatsQueue>::default(),
}
}
src/infra/src/queue/nop.rs
-96
View File
@@ -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 <http://www.gnu.org/licenses/>.
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<DeliverPolicy>,
) -> Result<Arc<mpsc::Receiver<super::Message>>> {
todo!()
}
async fn purge(&self, _topic: &str, _sequence: usize) -> Result<()> {
todo!()
}
}