diff --git a/Cargo.lock b/Cargo.lock index c27734aceb..7ce4f3da7b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5389,7 +5389,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381" [[package]] name = "openobserve" -version = "0.11.0" +version = "0.12.0" dependencies = [ "actix", "actix-cors", diff --git a/Cargo.toml b/Cargo.toml index 429ccb75bd..6351746067 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,7 +17,7 @@ keywords = [ license = "AGPL-3.0-only" name = "openobserve" repository = "https://github.com/openobserve/openobserve/" -version = "0.11.0" +version = "0.12.0" publish = false # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html diff --git a/src/config/src/meta/search.rs b/src/config/src/meta/search.rs index 7db61d68c3..e7b8813d3b 100644 --- a/src/config/src/meta/search.rs +++ b/src/config/src/meta/search.rs @@ -617,7 +617,8 @@ impl From for cluster_rpc::SearchRequest { stype: cluster_rpc::SearchType::User.into(), agg_mode: cluster_rpc::AggregateMode::Final.into(), query: Some(req_query), - file_list: vec![], + file_ids: vec![], + idx_files: vec![], stream_type: "".to_string(), timeout: req.timeout, work_group: "".to_string(), diff --git a/src/handler/grpc/request/file_list.rs b/src/handler/grpc/request/file_list.rs deleted file mode 100644 index 825975ff9f..0000000000 --- a/src/handler/grpc/request/file_list.rs +++ /dev/null @@ -1,102 +0,0 @@ -// Copyright 2024 Zinc Labs 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 config::{ - meta::stream::{PartitionTimeLevel, StreamType}, - metrics, -}; -use infra::file_list as infra_file_list; -use opentelemetry::global; -use proto::cluster_rpc::{ - filelist_server::Filelist, EmptyRequest, FileKey, FileList, FileListQueryRequest, MaxIdResponse, -}; -use tonic::{Request, Response, Status}; -use tracing_opentelemetry::OpenTelemetrySpanExt; - -pub struct Filelister; - -#[tonic::async_trait] -impl Filelist for Filelister { - async fn max_id(&self, req: Request) -> Result, Status> { - let start = std::time::Instant::now(); - let parent_cx = global::get_text_map_propagator(|prop| { - prop.extract(&super::MetadataMap(req.metadata())) - }); - tracing::Span::current().set_parent(parent_cx); - - let max_id = infra_file_list::get_max_pk_value() - .await - .map_err(|e| Status::internal(e.to_string()))?; - - // metrics - let time = start.elapsed().as_secs_f64(); - metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/file_list/max_id", "200", "", "", ""]) - .observe(time); - metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/file_list/max_id", "200", "", "", ""]) - .inc(); - - Ok(Response::new(MaxIdResponse { max_id })) - } - - async fn query( - &self, - req: Request, - ) -> Result, Status> { - let start = std::time::Instant::now(); - let parent_cx = global::get_text_map_propagator(|prop| { - prop.extract(&super::MetadataMap(req.metadata())) - }); - tracing::Span::current().set_parent(parent_cx); - - let req: &FileListQueryRequest = req.get_ref(); - let org_id = &req.org_id; - let stream_type = StreamType::from(req.stream_type.as_str()); - let stream_name = &req.stream_name; - let time_level = PartitionTimeLevel::from(req.time_level.as_str()); - let time_range = Some((req.start_time, req.end_time)); - let files = infra_file_list::query( - org_id, - stream_type, - stream_name, - time_level, - time_range, - None, - ) - .await - .map_err(|e| Status::internal(e.to_string()))?; - let items: Vec = files - .into_iter() - .map(|(file, meta)| FileKey { - key: file, - meta: Some((&meta).into()), - deleted: false, - segment_ids: None, - }) - .collect::<_>(); - - // metrics - let time = start.elapsed().as_secs_f64(); - metrics::GRPC_RESPONSE_TIME - .with_label_values(&["/file_list/query", "200", "", "", ""]) - .observe(time); - metrics::GRPC_INCOMING_REQUESTS - .with_label_values(&["/file_list/query", "200", "", "", ""]) - .inc(); - - Ok(Response::new(FileList { items })) - } -} diff --git a/src/handler/grpc/request/mod.rs b/src/handler/grpc/request/mod.rs index c96c638895..b4e3136acc 100644 --- a/src/handler/grpc/request/mod.rs +++ b/src/handler/grpc/request/mod.rs @@ -16,7 +16,6 @@ use opentelemetry::propagation::Extractor; pub mod event; -pub mod file_list; pub mod ingest; pub mod logs; pub mod metrics; diff --git a/src/infra/src/file_list/mod.rs b/src/infra/src/file_list/mod.rs index e14afe4656..726c7a66e7 100644 --- a/src/infra/src/file_list/mod.rs +++ b/src/infra/src/file_list/mod.rs @@ -28,6 +28,8 @@ pub mod mysql; pub mod postgres; pub mod sqlite; +const ID_BATCH_SIZE: usize = 2000; + static CLIENT: Lazy> = Lazy::new(connect); pub fn connect() -> Box { @@ -71,6 +73,15 @@ pub trait FileList: Sync + Send + 'static { time_range: Option<(i64, i64)>, flattened: Option, ) -> Result>; + async fn query_by_ids(&self, ids: &[i64]) -> Result>; + async fn query_ids( + &self, + org_id: &str, + stream_type: StreamType, + stream_name: &str, + time_level: PartitionTimeLevel, + time_range: Option<(i64, i64)>, + ) -> Result>; async fn query_deleted( &self, org_id: &str, @@ -239,6 +250,35 @@ pub async fn query( .await } +#[inline] +#[tracing::instrument(name = "infra:file_list:query_db_by_ids", skip_all)] +pub async fn query_by_ids(ids: &[i64]) -> Result> { + CLIENT.query_by_ids(ids).await +} + +#[inline] +#[tracing::instrument( + name = "infra:file_list:query_db_ids", + skip_all, + fields(org_id = org_id, stream_name = stream_name) +)] +pub async fn query_ids( + org_id: &str, + stream_type: StreamType, + stream_name: &str, + time_level: PartitionTimeLevel, + time_range: Option<(i64, i64)>, +) -> Result> { + if let Some((start, end)) = time_range { + if start > end || start == 0 || end == 0 { + return Err(Error::Message("[file_list] invalid time range".to_string())); + } + } + CLIENT + .query_ids(org_id, stream_type, stream_name, time_level, time_range) + .await +} + #[inline] pub async fn query_deleted(org_id: &str, time_max: i64, limit: i64) -> Result> { CLIENT.query_deleted(org_id, time_max, limit).await @@ -447,3 +487,9 @@ pub enum FileListJobStatus { Running, Done, } + +#[derive(Clone, Debug, Default, sqlx::FromRow)] +pub struct FileId { + pub id: i64, + pub original_size: i64, +} diff --git a/src/infra/src/file_list/mysql.rs b/src/infra/src/file_list/mysql.rs index 762bf0eada..60b921af1d 100644 --- a/src/infra/src/file_list/mysql.rs +++ b/src/infra/src/file_list/mysql.rs @@ -346,6 +346,87 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp .collect()) } + async fn query_by_ids(&self, ids: &[i64]) -> Result> { + if ids.is_empty() { + return Ok(Vec::default()); + } + let mut ret = Vec::new(); + let pool = CLIENT.clone(); + + for chunk in ids.chunks(super::ID_BATCH_SIZE) { + if chunk.is_empty() { + continue; + } + let ids = chunk + .iter() + .map(|id| id.to_string()) + .collect::>() + .join(","); + let query_str = format!( + "SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, flattened FROM file_list WHERE id IN ({ids})" + ); + let res = sqlx::query_as::<_, super::FileRecord>(&query_str) + .fetch_all(&pool) + .await?; + ret.extend_from_slice(&res); + } + + Ok(ret + .into_iter() + .map(|r| { + ( + format!("files/{}/{}/{}", r.stream, r.date, r.file), + FileMeta::from(&r), + ) + }) + .collect()) + } + + async fn query_ids( + &self, + org_id: &str, + stream_type: StreamType, + stream_name: &str, + _time_level: PartitionTimeLevel, + time_range: Option<(i64, i64)>, + ) -> Result> { + if let Some((start, end)) = time_range { + if start == 0 && end == 0 { + return Ok(Vec::new()); + } + } + + let stream_key = format!("{org_id}/{stream_type}/{stream_name}"); + + let pool = CLIENT.clone(); + + let (time_start, time_end) = time_range.unwrap_or((0, 0)); + let cfg = get_config(); + let ret = if cfg.limit.use_upper_bound_for_max_ts { + let max_ts_upper_bound = time_end + cfg.limit.upper_bound_for_max_ts * 60 * 1_000_000; + sqlx::query_as::<_, super::FileId>( + r#"SELECT id, original_size FROM file_list FORCE INDEX (file_list_stream_ts_idx) WHERE stream = ? AND max_ts >= ? AND max_ts <= ? AND min_ts <= ?;"#, + ) + .bind(stream_key) + .bind(time_start) + .bind(max_ts_upper_bound) + .bind(time_end) + .fetch_all(&pool) + .await? + } else { + sqlx::query_as::<_, super::FileId>( + r#"SELECT id, original_size FROM file_list FORCE INDEX (file_list_stream_ts_idx) WHERE stream = ? AND max_ts >= ? AND min_ts <= ?;"#, + ) + .bind(stream_key) + .bind(time_start) + .bind(time_end) + .fetch_all(&pool) + .await? + }; + + Ok(ret) + } + async fn query_deleted( &self, org_id: &str, diff --git a/src/infra/src/file_list/postgres.rs b/src/infra/src/file_list/postgres.rs index 9350aa33f5..e5dc8d18b0 100644 --- a/src/infra/src/file_list/postgres.rs +++ b/src/infra/src/file_list/postgres.rs @@ -230,9 +230,8 @@ impl super::FileList for PostgresFileList { let pool = CLIENT.clone(); let (stream_key, date_key, file_name) = parse_file_key_columns(file).map_err(|e| Error::Message(e.to_string()))?; - let ret = sqlx - ::query_as::<_, super::FileRecord>( - r#" + let ret = sqlx::query_as::<_, super::FileRecord>( + r#" SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, flattened FROM file_list WHERE stream = $1 AND date = $2 AND file = $3; "# @@ -301,9 +300,8 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp let pool = CLIENT.clone(); let ret = if flattened.is_some() { - sqlx - ::query_as::<_, super::FileRecord>( - r#" + sqlx::query_as::<_, super::FileRecord>( + r#" SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, flattened FROM file_list WHERE stream = $1 AND flattened = $2 LIMIT 1000; @@ -319,10 +317,10 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp let max_ts_upper_bound = time_end + cfg.limit.upper_bound_for_max_ts * 60 * 1_000_000; let sql = r#" - SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, flattened - FROM file_list - WHERE stream = $1 AND max_ts >= $2 AND max_ts <= $3 AND min_ts <= $4; - "#; +SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, flattened + FROM file_list + WHERE stream = $1 AND max_ts >= $2 AND max_ts <= $3 AND min_ts <= $4; + "#; sqlx::query_as::<_, super::FileRecord>(sql) .bind(stream_key) @@ -333,10 +331,10 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp .await } else { let sql = r#" - SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, flattened - FROM file_list - WHERE stream = $1 AND max_ts >= $2 AND min_ts <= $3; - "#; +SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, flattened + FROM file_list + WHERE stream = $1 AND max_ts >= $2 AND min_ts <= $3; + "#; sqlx::query_as::<_, super::FileRecord>(sql) .bind(stream_key) @@ -357,6 +355,87 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp .collect()) } + async fn query_by_ids(&self, ids: &[i64]) -> Result> { + if ids.is_empty() { + return Ok(Vec::default()); + } + let mut ret = Vec::new(); + let pool = CLIENT.clone(); + + for chunk in ids.chunks(super::ID_BATCH_SIZE) { + if chunk.is_empty() { + continue; + } + let ids = chunk + .iter() + .map(|id| id.to_string()) + .collect::>() + .join(","); + let query_str = format!( + "SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, flattened FROM file_list WHERE id IN ({ids})" + ); + let res = sqlx::query_as::<_, super::FileRecord>(&query_str) + .fetch_all(&pool) + .await?; + ret.extend_from_slice(&res); + } + + Ok(ret + .into_iter() + .map(|r| { + ( + format!("files/{}/{}/{}", r.stream, r.date, r.file), + FileMeta::from(&r), + ) + }) + .collect()) + } + + async fn query_ids( + &self, + org_id: &str, + stream_type: StreamType, + stream_name: &str, + _time_level: PartitionTimeLevel, + time_range: Option<(i64, i64)>, + ) -> Result> { + if let Some((start, end)) = time_range { + if start == 0 && end == 0 { + return Ok(Vec::new()); + } + } + + let stream_key = format!("{org_id}/{stream_type}/{stream_name}"); + + let pool = CLIENT.clone(); + + let (time_start, time_end) = time_range.unwrap_or((0, 0)); + let cfg = get_config(); + let ret = if cfg.limit.use_upper_bound_for_max_ts { + let max_ts_upper_bound = time_end + cfg.limit.upper_bound_for_max_ts * 60 * 1_000_000; + sqlx::query_as::<_, super::FileId>( + r#"SELECT id, original_size FROM file_list WHERE stream = $1 AND max_ts >= $2 AND max_ts <= $3 AND min_ts <= $4;"#, + ) + .bind(stream_key) + .bind(time_start) + .bind(max_ts_upper_bound) + .bind(time_end) + .fetch_all(&pool) + .await? + } else { + sqlx::query_as::<_, super::FileId>( + r#"SELECT id, original_size FROM file_list WHERE stream = $1 AND max_ts >= $2 AND min_ts <= $3;"#, + ) + .bind(stream_key) + .bind(time_start) + .bind(time_end) + .fetch_all(&pool) + .await? + }; + + Ok(ret) + } + async fn query_deleted( &self, org_id: &str, diff --git a/src/infra/src/file_list/sqlite.rs b/src/infra/src/file_list/sqlite.rs index fe0c9ecee6..973676063b 100644 --- a/src/infra/src/file_list/sqlite.rs +++ b/src/infra/src/file_list/sqlite.rs @@ -355,6 +355,88 @@ SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, comp .collect()) } + async fn query_by_ids(&self, ids: &[i64]) -> Result> { + if ids.is_empty() { + return Ok(Vec::default()); + } + + let mut ret = Vec::new(); + let pool = CLIENT_RO.clone(); + + for chunk in ids.chunks(super::ID_BATCH_SIZE) { + if chunk.is_empty() { + continue; + } + let ids = chunk + .iter() + .map(|id| id.to_string()) + .collect::>() + .join(","); + let query_str = format!( + "SELECT stream, date, file, deleted, min_ts, max_ts, records, original_size, compressed_size, flattened FROM file_list WHERE id IN ({ids})" + ); + let res = sqlx::query_as::<_, super::FileRecord>(&query_str) + .fetch_all(&pool) + .await?; + ret.extend_from_slice(&res); + } + + Ok(ret + .iter() + .map(|r| { + ( + format!("files/{}/{}/{}", r.stream, r.date, r.file), + r.into(), + ) + }) + .collect()) + } + + async fn query_ids( + &self, + org_id: &str, + stream_type: StreamType, + stream_name: &str, + _time_level: PartitionTimeLevel, + time_range: Option<(i64, i64)>, + ) -> Result> { + if let Some((start, end)) = time_range { + if start == 0 && end == 0 { + return Ok(Vec::new()); + } + } + + let stream_key = format!("{org_id}/{stream_type}/{stream_name}"); + + let pool = CLIENT_RO.clone(); + + let (time_start, time_end) = time_range.unwrap_or((0, 0)); + let cfg = get_config(); + let ret = if cfg.limit.use_upper_bound_for_max_ts { + let max_ts_upper_bound = time_end + cfg.limit.upper_bound_for_max_ts * 60 * 1_000_000; + sqlx::query_as::<_, super::FileId>( + r#"SELECT id, original_size FROM file_list WHERE stream = $1 AND max_ts >= $2 AND max_ts <= $3 AND min_ts <= $4;"#, + ) + .bind(stream_key) + .bind(time_start) + .bind(max_ts_upper_bound) + .bind(time_end) + .fetch_all(&pool) + .await? + } else { + sqlx::query_as::<_, super::FileId>( + r#"SELECT id, original_size FROM file_list WHERE stream = $1 AND max_ts >= $2 AND min_ts <= $3;"#, + ) + .bind(stream_key) + .bind(time_start) + .bind(time_end) + .fetch_all(&pool) + .await? + }; + + Ok(ret) + } + async fn query_deleted( &self, org_id: &str, diff --git a/src/main.rs b/src/main.rs index 3d2c00fa4a..ecb3020120 100644 --- a/src/main.rs +++ b/src/main.rs @@ -31,7 +31,6 @@ use openobserve::{ auth::check_auth, request::{ event::Eventer, - file_list::Filelister, ingest::Ingester, logs::LogsServer, metrics::{ingester::MetricsIngester, querier::MetricsQuerier}, @@ -54,9 +53,8 @@ use opentelemetry_proto::tonic::collector::{ }; use opentelemetry_sdk::{propagation::TraceContextPropagator, trace as sdktrace, Resource}; use proto::cluster_rpc::{ - event_server::EventServer, filelist_server::FilelistServer, ingest_server::IngestServer, - metrics_server::MetricsServer, query_cache_server::QueryCacheServer, - search_server::SearchServer, usage_server::UsageServer, + event_server::EventServer, ingest_server::IngestServer, metrics_server::MetricsServer, + query_cache_server::QueryCacheServer, search_server::SearchServer, usage_server::UsageServer, }; #[cfg(feature = "profiling")] use pyroscope::PyroscopeAgent; @@ -385,9 +383,6 @@ async fn init_common_grpc_server( .accept_compressed(CompressionEncoding::Gzip) .max_decoding_message_size(cfg.grpc.max_message_size * 1024 * 1024) .max_encoding_message_size(cfg.grpc.max_message_size * 1024 * 1024); - let filelist_svc = FilelistServer::new(Filelister) - .send_compressed(CompressionEncoding::Gzip) - .accept_compressed(CompressionEncoding::Gzip); let metrics_svc = MetricsServer::new(MetricsQuerier) .send_compressed(CompressionEncoding::Gzip) .accept_compressed(CompressionEncoding::Gzip) @@ -421,7 +416,6 @@ async fn init_common_grpc_server( .layer(tonic::service::interceptor(check_auth)) .add_service(event_svc) .add_service(search_svc) - .add_service(filelist_svc) .add_service(metrics_svc) .add_service(metrics_ingest_svc) .add_service(trace_svc) diff --git a/src/proto/build.rs b/src/proto/build.rs index e51b8fcf9e..32e63be13a 100644 --- a/src/proto/build.rs +++ b/src/proto/build.rs @@ -26,6 +26,7 @@ fn main() -> Result<()> { .type_attribute("FileList", "#[derive(serde::Serialize)]") .type_attribute("FileKey", "#[derive(Eq)]") .type_attribute("FileKey", "#[derive(serde::Serialize)]") + .type_attribute("IdxFileName", "#[derive(Eq)]") .type_attribute("FileDescriptor", "#[derive(Eq)]") .type_attribute("FileDescriptor", "#[derive(serde::Serialize)]") .type_attribute("FileMeta", "#[derive(Eq)]") @@ -55,7 +56,6 @@ fn main() -> Result<()> { &[ "proto/cluster/common.proto", "proto/cluster/event.proto", - "proto/cluster/filelist.proto", "proto/cluster/metrics.proto", "proto/cluster/search.proto", "proto/cluster/ingest.proto", diff --git a/src/proto/proto/cluster/filelist.proto b/src/proto/proto/cluster/filelist.proto deleted file mode 100644 index df89a25162..0000000000 --- a/src/proto/proto/cluster/filelist.proto +++ /dev/null @@ -1,28 +0,0 @@ -syntax = "proto3"; - -option java_multiple_files = true; -option java_package = "org.openobserve.cluster"; -option java_outer_classname = "filelistProto"; - -package cluster; - -import "cluster/common.proto"; - -service Filelist { - rpc MaxID (EmptyRequest) returns (MaxIdResponse) {} - rpc Query (FileListQueryRequest) returns (FileList) {} -} - - -message MaxIdResponse { - int64 max_id = 1; -} - -message FileListQueryRequest { - string org_id = 1; - string stream_type = 2; - string stream_name = 3; - string time_level = 4; - int64 start_time = 5; - int64 end_time = 6; -} diff --git a/src/proto/proto/cluster/search.proto b/src/proto/proto/cluster/search.proto index 6638311732..1785e49988 100644 --- a/src/proto/proto/cluster/search.proto +++ b/src/proto/proto/cluster/search.proto @@ -40,12 +40,18 @@ message SearchRequest { SearchType stype = 4; SearchQuery query = 5; AggregateMode agg_mode = 6; - repeated FileKey file_list = 7; - int64 timeout = 8; - string work_group = 9; - string index_type = 10; - optional string user_id = 11; - optional string search_event_type = 12; + repeated int64 file_ids = 7; + repeated IdxFileName idx_files = 8; + int64 timeout = 9; + string work_group = 10; + string index_type = 11; + optional string user_id = 12; + optional string search_event_type = 13; +} + +message IdxFileName { + string key = 1; + optional bytes segment_ids = 2; } message SearchResponse { diff --git a/src/proto/src/generated/cluster.rs b/src/proto/src/generated/cluster.rs index ab2e4668f3..33056137e0 100644 --- a/src/proto/src/generated/cluster.rs +++ b/src/proto/src/generated/cluster.rs @@ -461,379 +461,6 @@ pub mod event_server { const NAME: &'static str = "cluster.Event"; } } -#[allow(clippy::derive_partial_eq_without_eq)] -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct MaxIdResponse { - #[prost(int64, tag = "1")] - pub max_id: i64, -} -#[allow(clippy::derive_partial_eq_without_eq)] -#[derive(Clone, PartialEq, ::prost::Message)] -pub struct FileListQueryRequest { - #[prost(string, tag = "1")] - pub org_id: ::prost::alloc::string::String, - #[prost(string, tag = "2")] - pub stream_type: ::prost::alloc::string::String, - #[prost(string, tag = "3")] - pub stream_name: ::prost::alloc::string::String, - #[prost(string, tag = "4")] - pub time_level: ::prost::alloc::string::String, - #[prost(int64, tag = "5")] - pub start_time: i64, - #[prost(int64, tag = "6")] - pub end_time: i64, -} -/// Generated client implementations. -pub mod filelist_client { - #![allow(unused_variables, dead_code, missing_docs, clippy::let_unit_value)] - use tonic::codegen::*; - use tonic::codegen::http::Uri; - #[derive(Debug, Clone)] - pub struct FilelistClient { - inner: tonic::client::Grpc, - } - impl FilelistClient { - /// Attempt to create a new client by connecting to a given endpoint. - pub async fn connect(dst: D) -> Result - where - D: TryInto, - D::Error: Into, - { - let conn = tonic::transport::Endpoint::new(dst)?.connect().await?; - Ok(Self::new(conn)) - } - } - impl FilelistClient - where - T: tonic::client::GrpcService, - T::Error: Into, - T::ResponseBody: Body + Send + 'static, - ::Error: Into + Send, - { - pub fn new(inner: T) -> Self { - let inner = tonic::client::Grpc::new(inner); - Self { inner } - } - pub fn with_origin(inner: T, origin: Uri) -> Self { - let inner = tonic::client::Grpc::with_origin(inner, origin); - Self { inner } - } - pub fn with_interceptor( - inner: T, - interceptor: F, - ) -> FilelistClient> - where - F: tonic::service::Interceptor, - T::ResponseBody: Default, - T: tonic::codegen::Service< - http::Request, - Response = http::Response< - >::ResponseBody, - >, - >, - , - >>::Error: Into + Send + Sync, - { - FilelistClient::new(InterceptedService::new(inner, interceptor)) - } - /// Compress requests with the given encoding. - /// - /// This requires the server to support it otherwise it might respond with an - /// error. - #[must_use] - pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { - self.inner = self.inner.send_compressed(encoding); - self - } - /// Enable decompressing responses. - #[must_use] - pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { - self.inner = self.inner.accept_compressed(encoding); - self - } - /// Limits the maximum size of a decoded message. - /// - /// Default: `4MB` - #[must_use] - pub fn max_decoding_message_size(mut self, limit: usize) -> Self { - self.inner = self.inner.max_decoding_message_size(limit); - self - } - /// Limits the maximum size of an encoded message. - /// - /// Default: `usize::MAX` - #[must_use] - pub fn max_encoding_message_size(mut self, limit: usize) -> Self { - self.inner = self.inner.max_encoding_message_size(limit); - self - } - pub async fn max_id( - &mut self, - request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { - self.inner - .ready() - .await - .map_err(|e| { - tonic::Status::new( - tonic::Code::Unknown, - format!("Service was not ready: {}", e.into()), - ) - })?; - let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/cluster.Filelist/MaxID"); - let mut req = request.into_request(); - req.extensions_mut().insert(GrpcMethod::new("cluster.Filelist", "MaxID")); - self.inner.unary(req, path, codec).await - } - pub async fn query( - &mut self, - request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { - self.inner - .ready() - .await - .map_err(|e| { - tonic::Status::new( - tonic::Code::Unknown, - format!("Service was not ready: {}", e.into()), - ) - })?; - let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/cluster.Filelist/Query"); - let mut req = request.into_request(); - req.extensions_mut().insert(GrpcMethod::new("cluster.Filelist", "Query")); - self.inner.unary(req, path, codec).await - } - } -} -/// Generated server implementations. -pub mod filelist_server { - #![allow(unused_variables, dead_code, missing_docs, clippy::let_unit_value)] - use tonic::codegen::*; - /// Generated trait containing gRPC methods that should be implemented for use with FilelistServer. - #[async_trait] - pub trait Filelist: Send + Sync + 'static { - async fn max_id( - &self, - request: tonic::Request, - ) -> std::result::Result, tonic::Status>; - async fn query( - &self, - request: tonic::Request, - ) -> std::result::Result, tonic::Status>; - } - #[derive(Debug)] - pub struct FilelistServer { - inner: _Inner, - accept_compression_encodings: EnabledCompressionEncodings, - send_compression_encodings: EnabledCompressionEncodings, - max_decoding_message_size: Option, - max_encoding_message_size: Option, - } - struct _Inner(Arc); - impl FilelistServer { - pub fn new(inner: T) -> Self { - Self::from_arc(Arc::new(inner)) - } - pub fn from_arc(inner: Arc) -> Self { - let inner = _Inner(inner); - Self { - inner, - accept_compression_encodings: Default::default(), - send_compression_encodings: Default::default(), - max_decoding_message_size: None, - max_encoding_message_size: None, - } - } - pub fn with_interceptor( - inner: T, - interceptor: F, - ) -> InterceptedService - where - F: tonic::service::Interceptor, - { - InterceptedService::new(Self::new(inner), interceptor) - } - /// Enable decompressing requests with the given encoding. - #[must_use] - pub fn accept_compressed(mut self, encoding: CompressionEncoding) -> Self { - self.accept_compression_encodings.enable(encoding); - self - } - /// Compress responses with the given encoding, if the client supports it. - #[must_use] - pub fn send_compressed(mut self, encoding: CompressionEncoding) -> Self { - self.send_compression_encodings.enable(encoding); - self - } - /// Limits the maximum size of a decoded message. - /// - /// Default: `4MB` - #[must_use] - pub fn max_decoding_message_size(mut self, limit: usize) -> Self { - self.max_decoding_message_size = Some(limit); - self - } - /// Limits the maximum size of an encoded message. - /// - /// Default: `usize::MAX` - #[must_use] - pub fn max_encoding_message_size(mut self, limit: usize) -> Self { - self.max_encoding_message_size = Some(limit); - self - } - } - impl tonic::codegen::Service> for FilelistServer - where - T: Filelist, - B: Body + Send + 'static, - B::Error: Into + Send + 'static, - { - type Response = http::Response; - type Error = std::convert::Infallible; - type Future = BoxFuture; - fn poll_ready( - &mut self, - _cx: &mut Context<'_>, - ) -> Poll> { - Poll::Ready(Ok(())) - } - fn call(&mut self, req: http::Request) -> Self::Future { - let inner = self.inner.clone(); - match req.uri().path() { - "/cluster.Filelist/MaxID" => { - #[allow(non_camel_case_types)] - struct MaxIDSvc(pub Arc); - impl tonic::server::UnaryService - for MaxIDSvc { - type Response = super::MaxIdResponse; - type Future = BoxFuture< - tonic::Response, - tonic::Status, - >; - fn call( - &mut self, - request: tonic::Request, - ) -> Self::Future { - let inner = Arc::clone(&self.0); - let fut = async move { - ::max_id(&inner, request).await - }; - Box::pin(fut) - } - } - let accept_compression_encodings = self.accept_compression_encodings; - let send_compression_encodings = self.send_compression_encodings; - let max_decoding_message_size = self.max_decoding_message_size; - let max_encoding_message_size = self.max_encoding_message_size; - let inner = self.inner.clone(); - let fut = async move { - let inner = inner.0; - let method = MaxIDSvc(inner); - let codec = tonic::codec::ProstCodec::default(); - let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config( - accept_compression_encodings, - send_compression_encodings, - ) - .apply_max_message_size_config( - max_decoding_message_size, - max_encoding_message_size, - ); - let res = grpc.unary(method, req).await; - Ok(res) - }; - Box::pin(fut) - } - "/cluster.Filelist/Query" => { - #[allow(non_camel_case_types)] - struct QuerySvc(pub Arc); - impl< - T: Filelist, - > tonic::server::UnaryService - for QuerySvc { - type Response = super::FileList; - type Future = BoxFuture< - tonic::Response, - tonic::Status, - >; - fn call( - &mut self, - request: tonic::Request, - ) -> Self::Future { - let inner = Arc::clone(&self.0); - let fut = async move { - ::query(&inner, request).await - }; - Box::pin(fut) - } - } - let accept_compression_encodings = self.accept_compression_encodings; - let send_compression_encodings = self.send_compression_encodings; - let max_decoding_message_size = self.max_decoding_message_size; - let max_encoding_message_size = self.max_encoding_message_size; - let inner = self.inner.clone(); - let fut = async move { - let inner = inner.0; - let method = QuerySvc(inner); - let codec = tonic::codec::ProstCodec::default(); - let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config( - accept_compression_encodings, - send_compression_encodings, - ) - .apply_max_message_size_config( - max_decoding_message_size, - max_encoding_message_size, - ); - let res = grpc.unary(method, req).await; - Ok(res) - }; - Box::pin(fut) - } - _ => { - Box::pin(async move { - Ok( - http::Response::builder() - .status(200) - .header("grpc-status", "12") - .header("content-type", "application/grpc") - .body(empty_body()) - .unwrap(), - ) - }) - } - } - } - } - impl Clone for FilelistServer { - fn clone(&self) -> Self { - let inner = self.inner.clone(); - Self { - inner, - accept_compression_encodings: self.accept_compression_encodings, - send_compression_encodings: self.send_compression_encodings, - max_decoding_message_size: self.max_decoding_message_size, - max_encoding_message_size: self.max_encoding_message_size, - } - } - } - impl Clone for _Inner { - fn clone(&self) -> Self { - Self(Arc::clone(&self.0)) - } - } - impl std::fmt::Debug for _Inner { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "{:?}", self.0) - } - } - impl tonic::server::NamedService for FilelistServer { - const NAME: &'static str = "cluster.Filelist"; - } -} #[derive(serde::Serialize)] #[allow(clippy::derive_partial_eq_without_eq)] #[derive(Clone, PartialEq, ::prost::Message)] @@ -1365,20 +992,31 @@ pub struct SearchRequest { pub query: ::core::option::Option, #[prost(enumeration = "AggregateMode", tag = "6")] pub agg_mode: i32, - #[prost(message, repeated, tag = "7")] - pub file_list: ::prost::alloc::vec::Vec, - #[prost(int64, tag = "8")] + #[prost(int64, repeated, tag = "7")] + pub file_ids: ::prost::alloc::vec::Vec, + #[prost(message, repeated, tag = "8")] + pub idx_files: ::prost::alloc::vec::Vec, + #[prost(int64, tag = "9")] pub timeout: i64, - #[prost(string, tag = "9")] - pub work_group: ::prost::alloc::string::String, #[prost(string, tag = "10")] + pub work_group: ::prost::alloc::string::String, + #[prost(string, tag = "11")] pub index_type: ::prost::alloc::string::String, - #[prost(string, optional, tag = "11")] - pub user_id: ::core::option::Option<::prost::alloc::string::String>, #[prost(string, optional, tag = "12")] + pub user_id: ::core::option::Option<::prost::alloc::string::String>, + #[prost(string, optional, tag = "13")] pub search_event_type: ::core::option::Option<::prost::alloc::string::String>, } #[derive(Eq)] +#[allow(clippy::derive_partial_eq_without_eq)] +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IdxFileName { + #[prost(string, tag = "1")] + pub key: ::prost::alloc::string::String, + #[prost(bytes = "vec", optional, tag = "2")] + pub segment_ids: ::core::option::Option<::prost::alloc::vec::Vec>, +} +#[derive(Eq)] #[derive(serde::Serialize)] #[allow(clippy::derive_partial_eq_without_eq)] #[derive(Clone, PartialEq, ::prost::Message)] diff --git a/src/service/compact/merge.rs b/src/service/compact/merge.rs index b0fb674060..430f888b2b 100644 --- a/src/service/compact/merge.rs +++ b/src/service/compact/merge.rs @@ -294,7 +294,6 @@ pub async fn merge_by_stream( partition_time_level, partition_offset_start, partition_offset_end, - true, ) .await .map_err(|e| anyhow::anyhow!("query file list failed: {}", e))?; @@ -318,7 +317,6 @@ pub async fn merge_by_stream( partition_time_level, lookback_offset_start, lookback_offset_end, - true, ) .await .map_err(|e| anyhow::anyhow!("query lookback file list failed: {}", e))?; diff --git a/src/service/compact/retention.rs b/src/service/compact/retention.rs index 62b77a24b6..d3c62c73dd 100644 --- a/src/service/compact/retention.rs +++ b/src/service/compact/retention.rs @@ -126,7 +126,6 @@ pub async fn delete_all( PartitionTimeLevel::Unset, start_time, end_time, - true, ) .await?; if cfg.compact.data_retention_history { @@ -232,7 +231,6 @@ pub async fn delete_by_date( PartitionTimeLevel::Unset, time_range.0, time_range.1, - true, ) .await?; if cfg.compact.data_retention_history { @@ -306,7 +304,6 @@ async fn delete_from_file_list( PartitionTimeLevel::Unset, time_range.0, time_range.1, - true, ) .await?; if files.is_empty() { diff --git a/src/service/file_list.rs b/src/service/file_list.rs index 1bf95aba95..351603dde5 100644 --- a/src/service/file_list.rs +++ b/src/service/file_list.rs @@ -16,32 +16,16 @@ use std::io::Write; use config::{ - cluster::LOCAL_NODE, get_config, ider, meta::{ - cluster::{Node, RoleGroup}, search::ScanStats, stream::{FileKey, FileMeta, PartitionTimeLevel, StreamType}, }, utils::{file::get_file_meta as util_get_file_meta, json}, }; -use futures::future::try_join_all; -use infra::{ - errors::{Error, ErrorCodes}, - file_list, storage, -}; -use proto::cluster_rpc; -use tonic::{ - codec::CompressionEncoding, - metadata::{MetadataKey, MetadataValue}, - Request, -}; -use tracing_opentelemetry::OpenTelemetrySpanExt; +use infra::{file_list, storage}; -use crate::{ - common::infra::cluster, - service::{db, grpc::get_cached_channel, search::MetadataMap}, -}; +use crate::service::db; #[tracing::instrument( name = "service::file_list::query", @@ -55,226 +39,6 @@ pub async fn query( time_level: PartitionTimeLevel, time_min: i64, time_max: i64, - is_local: bool, -) -> Result, anyhow::Error> { - let cfg = get_config(); - if is_local || cfg.common.local_mode { - return query_inner( - org_id, - stream_name, - stream_type, - time_level, - time_min, - time_max, - ) - .await; - } - - // cluster mode - let start: std::time::Instant = std::time::Instant::now(); - let nodes = cluster::get_cached_online_querier_nodes(Some(RoleGroup::Interactive)) - .await - .unwrap_or_default(); - if nodes.is_empty() { - return Ok(Vec::new()); - } - let mut tasks = Vec::with_capacity(3); - // get first three nodes to check file list max id - for node in nodes.into_iter().take(3) { - let cfg = cfg.clone(); - let org_id = org_id.to_string(); - let task = tokio::task::spawn(async move { - let req = cluster_rpc::EmptyRequest::default(); - let org_id: MetadataValue<_> = org_id - .parse() - .map_err(|_| Error::Message("invalid org_id".to_string()))?; - let mut request = tonic::Request::new(req); - - opentelemetry::global::get_text_map_propagator(|propagator| { - propagator.inject_context( - &tracing::Span::current().context(), - &mut MetadataMap(request.metadata_mut()), - ) - }); - - let org_header_key: MetadataKey<_> = cfg - .grpc - .org_header_key - .parse() - .map_err(|_| Error::Message("invalid org_header_key".to_string()))?; - let token: MetadataValue<_> = cluster::get_internal_grpc_token() - .parse() - .map_err(|_| Error::Message("invalid token".to_string()))?; - let channel = get_cached_channel(&node.grpc_addr).await.map_err(|err| { - log::error!( - "file_list->grpc: node: {}, connect err: {:?}", - &node.grpc_addr, - err - ); - Error::ErrorCode(ErrorCodes::ServerInternalError( - "connect querier error".to_string(), - )) - })?; - let mut client = cluster_rpc::filelist_client::FilelistClient::with_interceptor( - channel, - move |mut req: Request<()>| { - req.metadata_mut().insert("authorization", token.clone()); - req.metadata_mut() - .insert(org_header_key.clone(), org_id.clone()); - Ok(req) - }, - ); - client = client - .send_compressed(CompressionEncoding::Gzip) - .accept_compressed(CompressionEncoding::Gzip) - .max_decoding_message_size(cfg.grpc.max_message_size * 1024 * 1024) - .max_encoding_message_size(cfg.grpc.max_message_size * 1024 * 1024); - let response: cluster_rpc::MaxIdResponse = match client.max_id(request).await { - Ok(res) => res.into_inner(), - Err(err) => { - log::error!( - "file_list->grpc: node: {}, query max_id err: {:?}", - &node.grpc_addr, - err - ); - if err.code() == tonic::Code::Internal { - let err = ErrorCodes::from_json(err.message())?; - return Err(Error::ErrorCode(err)); - } - return Err(Error::ErrorCode(ErrorCodes::ServerInternalError(format!( - "search node response error: {}", - err - )))); - } - }; - Ok((node, response.max_id)) - }); - tasks.push(task); - } - let mut max_id: i64 = 0; - let mut max_id_node: Option = None; - let task_results = try_join_all(tasks).await?; - for task in task_results { - let res = task?; - if res.1 > max_id { - max_id = res.1; - max_id_node = Some(res.0); - } - } - if max_id_node.is_none() { - return Ok(Vec::new()); - } - let node = max_id_node.unwrap(); - if node.uuid.eq(LOCAL_NODE.uuid.as_str()) { - // local node, no need grpc call - let files = query_inner( - org_id, - stream_name, - stream_type, - time_level, - time_min, - time_max, - ) - .await?; - log::info!( - "file_list->local: query list: {}, time: {}ms", - files.len(), - start.elapsed().as_millis() - ); - return Ok(files); - } - // use the max_id node to query file_list - let req = cluster_rpc::FileListQueryRequest { - org_id: org_id.to_string(), - stream_name: stream_name.to_string(), - stream_type: stream_type.to_string(), - time_level: time_level.to_string(), - start_time: time_min, - end_time: time_max, - }; - let org_id: MetadataValue<_> = org_id - .parse() - .map_err(|_| Error::Message("invalid org_id".to_string()))?; - let mut request = tonic::Request::new(req); - - opentelemetry::global::get_text_map_propagator(|propagator| { - propagator.inject_context( - &tracing::Span::current().context(), - &mut MetadataMap(request.metadata_mut()), - ) - }); - - let org_header_key: MetadataKey<_> = cfg - .grpc - .org_header_key - .parse() - .map_err(|_| Error::Message("invalid org_header_key".to_string()))?; - let token: MetadataValue<_> = cluster::get_internal_grpc_token() - .parse() - .map_err(|_| Error::Message("invalid token".to_string()))?; - let channel = get_cached_channel(&node.grpc_addr).await.map_err(|err| { - log::error!( - "file_list->grpc: node: {}, connect err: {:?}", - &node.grpc_addr, - err - ); - Error::ErrorCode(ErrorCodes::ServerInternalError( - "connect querier error".to_string(), - )) - })?; - let mut client = cluster_rpc::filelist_client::FilelistClient::with_interceptor( - channel, - move |mut req: Request<()>| { - req.metadata_mut().insert("authorization", token.clone()); - req.metadata_mut() - .insert(org_header_key.clone(), org_id.clone()); - Ok(req) - }, - ); - - client = client - .send_compressed(CompressionEncoding::Gzip) - .accept_compressed(CompressionEncoding::Gzip) - .max_decoding_message_size(cfg.grpc.max_message_size * 1024 * 1024) - .max_encoding_message_size(cfg.grpc.max_message_size * 1024 * 1024); - let response: cluster_rpc::FileList = match client.query(request).await { - Ok(res) => res.into_inner(), - Err(err) => { - log::error!( - "file_list->grpc: node: {}, query list err: {:?}", - &node.grpc_addr, - err - ); - if err.code() == tonic::Code::Internal { - let err = ErrorCodes::from_json(err.message())?; - return Err(anyhow::anyhow!(Error::ErrorCode(err).to_string())); - } - return Err(anyhow::anyhow!("search node error".to_string())); - } - }; - let files = response.items.iter().map(FileKey::from).collect::>(); - log::info!( - "file_list->grpc: node: {}, query list: {}, time: {}ms", - &node.grpc_addr, - files.len(), - start.elapsed().as_millis() - ); - Ok(files) -} - -#[inline] -#[tracing::instrument( - name = "service:file_list:query_inner", - skip_all, - fields(org_id = org_id, stream_name = stream_name) -)] -async fn query_inner( - org_id: &str, - stream_name: &str, - stream_type: StreamType, - time_level: PartitionTimeLevel, - time_min: i64, - time_max: i64, ) -> Result, anyhow::Error> { let files = file_list::query( org_id, @@ -297,6 +61,43 @@ async fn query_inner( Ok(file_keys) } +pub async fn query_by_ids(ids: &[i64]) -> Result, anyhow::Error> { + let files = file_list::query_by_ids(ids).await?; + let mut file_keys = Vec::with_capacity(files.len()); + for (key, meta) in files { + file_keys.push(FileKey { + key, + meta, + deleted: false, + segment_ids: None, + }); + } + Ok(file_keys) +} + +#[tracing::instrument( + name = "service::file_list::query_ids", + skip_all, + fields(org_id = org_id, stream_name = stream_name) +)] +pub async fn query_ids( + org_id: &str, + stream_name: &str, + stream_type: StreamType, + time_level: PartitionTimeLevel, + time_min: i64, + time_max: i64, +) -> Result, anyhow::Error> { + Ok(file_list::query_ids( + org_id, + stream_type, + stream_name, + time_level, + Some((time_min, time_max)), + ) + .await?) +} + #[inline] pub async fn calculate_files_size(files: &[FileKey]) -> Result { let mut stats = ScanStats::new(); diff --git a/src/service/promql/search/grpc/storage.rs b/src/service/promql/search/grpc/storage.rs index a877143815..3e8b3ff839 100644 --- a/src/service/promql/search/grpc/storage.rs +++ b/src/service/promql/search/grpc/storage.rs @@ -193,7 +193,6 @@ async fn get_file_list( time_level, time_min, time_max, - true, ) .await { diff --git a/src/service/search/cluster/mod.rs b/src/service/search/cluster/mod.rs index f087c07df7..b32c7c45eb 100644 --- a/src/service/search/cluster/mod.rs +++ b/src/service/search/cluster/mod.rs @@ -35,6 +35,7 @@ use hashbrown::{HashMap, HashSet}; use infra::{ dist_lock, errors::{Error, ErrorCodes, Result}, + file_list::FileId, schema::{unwrap_partition_time_level, unwrap_stream_settings}, }; use proto::cluster_rpc; @@ -144,7 +145,7 @@ pub async fn search( unwrap_partition_time_level(stream_settings.partition_time_level, stream_type); // get file list - let mut file_list = get_file_list( + let file_id_list = get_file_id_list( trace_id, &meta, stream_type, @@ -152,11 +153,12 @@ pub async fn search( &stream_settings.partition_keys, ) .await; + let file_list_took = start.elapsed().as_millis() as usize; log::info!( "[trace_id {trace_id}] search: get file_list time_range: {:?}, num: {}, took: {} ms", meta.meta.time_range, - file_list.len(), + file_id_list.len(), file_list_took, ); @@ -164,16 +166,25 @@ pub async fn search( // then filter the file list based on the inverted index. let mut idx_scan_size = 0; let mut idx_took = 0; + let idx_file_list; if use_inverted_index { - (file_list, idx_scan_size, idx_took) = - get_file_list_by_inverted_index(meta.clone(), req.clone(), &file_list).await?; + (idx_file_list, idx_scan_size, idx_took) = + get_file_list_by_inverted_index(meta.clone(), req.clone()).await?; log::info!( - "[trace_id {trace_id}] search: get file_list from inverted index time_range: {:?}, num: {}, scan_size: {}, took: {} ms", + "[trace_id {trace_id}] search: get idx_file_list from inverted index time_range: {:?}, num: {}, scan_size: {}, took: {} ms", meta.meta.time_range, - file_list.len(), + idx_file_list.len(), idx_scan_size, idx_took, ); + + req.idx_files = idx_file_list + .into_iter() + .map(|f| cluster_rpc::IdxFileName { + key: f.key, + segment_ids: f.segment_ids, + }) + .collect(); } metrics::QUERY_PENDING_NUMS @@ -185,7 +196,7 @@ pub async fn search( // 1. get work group #[cfg(feature = "enterprise")] let work_group: Option = Some( - o2_enterprise::enterprise::search::work_group::predict(&nodes, &file_list), + o2_enterprise::enterprise::search::work_group::predict(&nodes, &file_id_list), ); #[cfg(feature = "enterprise")] super::SEARCH_SERVER @@ -265,7 +276,7 @@ pub async fn search( req.work_group = work_group_str; let mut partition_files = Vec::new(); - let mut file_num = file_list.len(); + let mut file_num = file_id_list.len(); let mut partition_strategy = QueryPartitionStrategy::from(&cfg.common.feature_query_partition_strategy); if cfg.memory_cache.cache_latest_files { @@ -280,13 +291,13 @@ pub async fn search( } } QueryPartitionStrategy::FileSize => { - let files = partition_file_by_bytes(&file_list, querier_num); + let files = partition_file_by_bytes(&file_id_list, querier_num); file_num = files.len(); partition_files = files; 1 } QueryPartitionStrategy::FileHash => { - let files = partition_file_by_hash(&file_list, &nodes, req_node_group).await; + let files = partition_file_by_hash(&file_id_list, &nodes, req_node_group).await; file_num = files.len(); partition_files = files; 1 @@ -300,20 +311,17 @@ pub async fn search( #[cfg(feature = "enterprise")] { - let mut records = 0; + let records = 0; + let compressed_size = 0; let mut original_size = 0; - let mut compressed_size = 0; - for file in file_list.iter() { - let file_meta = &file.meta; - records += file_meta.records; - original_size += file_meta.original_size; - compressed_size += file_meta.compressed_size; + for file in file_id_list.iter() { + original_size += file.original_size; } original_size += idx_scan_size as i64; super::SEARCH_SERVER .add_file_stats( trace_id, - file_list.len() as i64, + file_id_list.len() as i64, records, original_size, compressed_size, @@ -344,23 +352,23 @@ pub async fn search( req.stype = cluster_rpc::SearchType::Cluster as _; match partition_strategy { QueryPartitionStrategy::FileNum => { - req.file_list = file_list + req.file_ids = file_id_list [offset_start..min(offset_start + offset, file_num)] .to_vec() .iter() - .map(cluster_rpc::FileKey::from) + .map(|f| f.id) .collect(); offset_start += offset; } QueryPartitionStrategy::FileSize | QueryPartitionStrategy::FileHash => { - req.file_list = partition_files + req.file_ids = partition_files .get(offset_start) .unwrap() .iter() - .map(|f| cluster_rpc::FileKey::from(*f)) + .map(|f| f.id) .collect(); offset_start += offset; - if req.file_list.is_empty() { + if req.file_ids.is_empty() { if node.is_ingester() { req.stype = cluster_rpc::SearchType::WalOnly as _; } else { @@ -374,7 +382,7 @@ pub async fn search( } } - let req_files = req.file_list.len(); + let req_files = req.file_ids.len(); let node_addr = node.grpc_addr.clone(); let grpc_span = info_span!( "service:search:cluster:grpc_search", @@ -835,12 +843,6 @@ pub(crate) async fn get_file_list( time_level: PartitionTimeLevel, partition_keys: &[StreamPartition], ) -> Vec { - let is_local = get_config().common.meta_store_external - || infra_cluster::get_cached_online_querier_nodes(Some(RoleGroup::Interactive)) - .await - .unwrap_or_default() - .len() - <= 1; let (time_min, time_max) = sql.meta.time_range.unwrap(); let file_list = file_list::query( &sql.org_id, @@ -849,7 +851,6 @@ pub(crate) async fn get_file_list( time_level, time_min, time_max, - is_local, ) .await .unwrap_or_default(); @@ -868,22 +869,37 @@ pub(crate) async fn get_file_list( files } -pub(crate) fn partition_file_by_bytes( - file_keys: &[FileKey], - num_nodes: usize, -) -> Vec> { - let mut partitions: Vec> = vec![Vec::new(); num_nodes]; - let sum_original_size = file_keys - .iter() - .map(|fk| fk.meta.original_size) - .sum::(); +#[tracing::instrument(skip(sql), fields(org_id = sql.org_id, stream_name = sql.stream_name))] +pub(crate) async fn get_file_id_list( + _trace_id: &str, + sql: &super::sql::Sql, + stream_type: StreamType, + time_level: PartitionTimeLevel, + partition_keys: &[StreamPartition], +) -> Vec { + let (time_min, time_max) = sql.meta.time_range.unwrap(); + file_list::query_ids( + &sql.org_id, + &sql.stream_name, + stream_type, + time_level, + time_min, + time_max, + ) + .await + .unwrap_or_default() +} + +pub(crate) fn partition_file_by_bytes(file_keys: &[FileId], num_nodes: usize) -> Vec> { + let mut partitions: Vec> = vec![Vec::new(); num_nodes]; + let sum_original_size = file_keys.iter().map(|fk| fk.original_size).sum::(); let avg_size = sum_original_size / num_nodes as i64; let mut node_size = 0; let mut node_k = 0; for fk in file_keys { - node_size += fk.meta.original_size; + node_size += fk.original_size; if node_size >= avg_size && node_k != num_nodes - 1 && !partitions[node_k].is_empty() { - node_size = fk.meta.original_size; + node_size = fk.original_size; node_k += 1; partitions[node_k].push(fk); continue; @@ -894,10 +910,10 @@ pub(crate) fn partition_file_by_bytes( } pub(crate) async fn partition_file_by_hash<'a>( - file_keys: &'a [FileKey], + file_keys: &'a [FileId], nodes: &'a [Node], group: Option, -) -> Vec> { +) -> Vec> { let mut node_idx = HashMap::with_capacity(nodes.len()); let mut idx = 0; for node in nodes { @@ -907,10 +923,10 @@ pub(crate) async fn partition_file_by_hash<'a>( node_idx.insert(&node.name, idx); idx += 1; } - let mut partitions: Vec> = vec![Vec::new(); idx]; + let mut partitions: Vec> = vec![Vec::new(); idx]; for fk in file_keys { let node_name = - infra_cluster::get_node_from_consistent_hash(&fk.key, &Role::Querier, group) + infra_cluster::get_node_from_consistent_hash(&fk.id.to_string(), &Role::Querier, group) .await .expect("there is no querier node in consistent hash ring"); let idx = match node_idx.get(&node_name) { @@ -995,16 +1011,11 @@ fn handle_metrics_response(sources: Vec) -> Vec { async fn get_file_list_by_inverted_index( meta: Arc, mut idx_req: cluster_rpc::SearchRequest, - file_list: &[FileKey], ) -> Result<(Vec, usize, usize)> { let start = std::time::Instant::now(); let cfg = get_config(); let stream_type = StreamType::from(idx_req.stream_type.as_str()); - let file_list_map = file_list - .iter() - .map(|f| (&f.key, &f.meta)) - .collect::>(); // Get all the unique terms which the user has searched. let terms = meta @@ -1028,9 +1039,8 @@ async fn get_file_list_by_inverted_index( let fts_condition = terms .iter() .map(|x| match cfg.common.full_text_search_type.as_str() { - "contains" => format!("term LIKE '%{x}%'"), "eq" => format!("term = '{x}'"), - // Default to "prefix" + "contains" => format!("term LIKE '%{x}%'"), _ => format!("term LIKE '{x}%'"), }) .collect::>() @@ -1069,7 +1079,7 @@ async fn get_file_list_by_inverted_index( // If both empty return original file list with other params as 0 let search_condition = match (index_condition.is_empty(), fts_condition.is_empty()) { (true, true) => { - return Ok((file_list.to_vec(), 0, 0)); + return Ok((vec![], 0, 0)); } (true, false) => fts_condition, (false, true) => index_condition, @@ -1125,9 +1135,6 @@ async fn get_file_list_by_inverted_index( "files/{}/{}/{}/{}", meta.org_id, stream_type, meta.stream_name, filename ); - let Some(file_meta) = file_list_map.get(&prefixed_filename) else { - continue; - }; let segment_ids = match item.get("segment_ids") { None => None, Some(v) => hex::decode(v.as_str().unwrap()).ok(), @@ -1136,9 +1143,7 @@ async fn get_file_list_by_inverted_index( .entry(prefixed_filename.clone()) .or_insert(FileKey { key: prefixed_filename, - meta: (*file_meta).clone(), - deleted: false, - segment_ids: None, + ..Default::default() }); match (&entry.segment_ids, &segment_ids) { (Some(_), None) => {} @@ -1171,153 +1176,55 @@ mod tests { #[test] fn test_partition_file_by_bytes() { - use config::meta::stream::FileMeta; - let vec = vec![ - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 256, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 256, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 100, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 256, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 1, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 256, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 200, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 30, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 90, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 256, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 5, - compressed_size: -1, - flattened: false, - }, - false, - ), - FileKey::new( - "", - FileMeta { - min_ts: -1, - max_ts: -1, - records: -1, - original_size: 150, - compressed_size: -1, - flattened: false, - }, - false, - ), + FileId { + id: 0, + original_size: 256, + }, + FileId { + id: 0, + original_size: 256, + }, + FileId { + id: 0, + original_size: 100, + }, + FileId { + id: 0, + original_size: 256, + }, + FileId { + id: 0, + original_size: 1, + }, + FileId { + id: 0, + original_size: 256, + }, + FileId { + id: 0, + original_size: 200, + }, + FileId { + id: 0, + original_size: 30, + }, + FileId { + id: 0, + original_size: 90, + }, + FileId { + id: 0, + original_size: 256, + }, + FileId { + id: 0, + original_size: 5, + }, + FileId { + id: 0, + original_size: 150, + }, ]; let expected: Vec> = vec![ vec![256, 256, 100], @@ -1327,7 +1234,7 @@ mod tests { let byte = partition_file_by_bytes(&vec, 3); for value in byte .iter() - .map(|x| x.iter().map(|v| v.meta.original_size).collect::>()) + .map(|x| x.iter().map(|v| v.original_size).collect::>()) .enumerate() { assert_eq!(value.1, expected.get(value.0).unwrap().clone()); diff --git a/src/service/search/grpc/mod.rs b/src/service/search/grpc/mod.rs index 836be11c2a..b277a1ea04 100644 --- a/src/service/search/grpc/mod.rs +++ b/src/service/search/grpc/mod.rs @@ -22,7 +22,7 @@ use config::{ get_config, meta::{ search::{ScanStats, SearchType, StorageType}, - stream::{FileKey, StreamType}, + stream::{FileKey, StreamPartition, StreamType}, }, utils::record_batch_ext::format_recordbatch_by_schema, }; @@ -31,10 +31,11 @@ use infra::{ errors::{Error, ErrorCodes}, schema::unwrap_stream_settings, }; -use proto::cluster_rpc; +use proto::cluster_rpc::{self, IdxFileName}; use crate::service::{ db, + file_list::query_by_ids, search::{ datafusion::exec, generate_search_schema, generate_select_start_search_schema, @@ -141,7 +142,22 @@ pub async fn search( // search in object storage let req_stype = req.stype; if req_stype != cluster_rpc::SearchType::WalOnly as i32 { - let file_list: Vec = req.file_list.iter().map(FileKey::from).collect(); + let (file_list, file_list_took) = get_file_list_by_ids( + &trace_id, + &req.file_ids, + &sql, + stream_type, + &stream_settings.partition_keys, + &req.idx_files, + ) + .await; + log::info!( + "[trace_id {trace_id}] grpc->search in: part_id: {}, get file_list by ids, num: {}, took: {} ms", + req.job.as_ref().unwrap().partition, + file_list.len(), + file_list_took, + ); + let (tbls, stats, partitions) = storage::search(&trace_id, sql.clone(), &file_list, stream_type, &work_group).await?; tables.extend(tbls); @@ -366,6 +382,55 @@ fn check_memory_circuit_breaker(trace_id: &str, scan_stats: &ScanStats) -> Resul Ok(()) } +#[tracing::instrument(skip_all, fields(org_id = sql.org_id, stream_name = sql.stream_name))] +pub(crate) async fn get_file_list_by_ids( + _trace_id: &str, + ids: &[i64], + sql: &super::sql::Sql, + stream_type: StreamType, + partition_keys: &[StreamPartition], + idx_file_list: &[IdxFileName], +) -> (Vec, usize) { + let start = std::time::Instant::now(); + let file_list = query_by_ids(ids).await.unwrap_or_default(); + // if there are any files in idx_files_list, use them to filter the files we got from ids, + // otherwise use all the files we got from ids + let file_list = if idx_file_list.is_empty() { + file_list + } else { + let mut files = Vec::with_capacity(idx_file_list.len()); + let file_list_map: HashMap<_, _> = + file_list.into_iter().map(|f| (f.key.clone(), f)).collect(); + for idx_file in idx_file_list.iter() { + if let Some(file) = file_list_map.get(&idx_file.key) { + let mut new_file = file.clone(); + if let Some(segment_ids) = idx_file.segment_ids.as_ref() { + new_file.segment_ids = Some(segment_ids.clone()); + } + files.push(new_file); + } + } + files + }; + + // filter file_list by partition keys + // cannot use .iter().filter() as the function is async cannot be used in filter closure + let mut files = Vec::with_capacity(file_list.len()); + for file in file_list { + if sql + .match_source(&file, false, false, stream_type, partition_keys) + .await + { + files.push(file); + } + } + + // sort by min_ts and dedup + files.sort_by(|a, b| a.meta.min_ts.cmp(&b.meta.min_ts)); + files.dedup_by(|a, b| a.key == b.key); + (files, start.elapsed().as_millis() as usize) +} + struct ReleaseFileGuard { trace_id: String, } diff --git a/src/service/search/grpc/storage.rs b/src/service/search/grpc/storage.rs index 5aa7472c47..a9f188caa0 100644 --- a/src/service/search/grpc/storage.rs +++ b/src/service/search/grpc/storage.rs @@ -304,7 +304,6 @@ async fn get_file_list( time_level, time_min, time_max, - true, ) .await {