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

feat: optimise file list querying for better search perf (#4518)

Browse Source
This commit is contained in:
Yashodhan Joshi
2024-09-19 22:45:33 +08:00
committed by GitHub
parent f8a27b5c0d
commit 8d5ebbb3c8
21 changed files with 560 additions and 998 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1 -1
View File
@@ -5389,7 +5389,7 @@ checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
[[package]]
name = "openobserve"
version = "0.11.0"
version = "0.12.0"
dependencies = [
"actix",
"actix-cors",
Cargo.toml
+1 -1
View File
@@ -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
src/config/src/meta/search.rs
+2 -1
View File
@@ -617,7 +617,8 @@ impl From<Request> 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(),
src/handler/grpc/request/file_list.rs
-102
View File
@@ -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 <http://www.gnu.org/licenses/>.
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<EmptyRequest>) -> Result<Response<MaxIdResponse>, 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<FileListQueryRequest>,
) -> Result<Response<FileList>, 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<FileKey> = 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 }))
}
}
src/handler/grpc/request/mod.rs
-1
View File
@@ -16,7 +16,6 @@
use opentelemetry::propagation::Extractor;
pub mod event;
pub mod file_list;
pub mod ingest;
pub mod logs;
pub mod metrics;
src/infra/src/file_list/mod.rs
+46
View File
@@ -28,6 +28,8 @@ pub mod mysql;
pub mod postgres;
pub mod sqlite;
const ID_BATCH_SIZE: usize = 2000;
static CLIENT: Lazy<Box<dyn FileList>> = Lazy::new(connect);
pub fn connect() -> Box<dyn FileList> {
@@ -71,6 +73,15 @@ pub trait FileList: Sync + Send + 'static {
time_range: Option<(i64, i64)>,
flattened: Option<bool>,
) -> Result<Vec<(String, FileMeta)>>;
async fn query_by_ids(&self, ids: &[i64]) -> Result<Vec<(String, FileMeta)>>;
async fn query_ids(
&self,
org_id: &str,
stream_type: StreamType,
stream_name: &str,
time_level: PartitionTimeLevel,
time_range: Option<(i64, i64)>,
) -> Result<Vec<FileId>>;
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<Vec<(String, FileMeta)>> {
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<Vec<FileId>> {
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<Vec<(String, bool)>> {
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,
}
src/infra/src/file_list/mysql.rs
+81
View File
@@ -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<Vec<(String, FileMeta)>> {
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::<Vec<String>>()
.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<Vec<super::FileId>> {
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,
src/infra/src/file_list/postgres.rs
+93 -14
View File
@@ -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<Vec<(String, FileMeta)>> {
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::<Vec<String>>()
.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<Vec<super::FileId>> {
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,
src/infra/src/file_list/sqlite.rs
+82
View File
@@ -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<Vec<(String, FileMeta)>> {
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::<Vec<String>>()
.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<Vec<super::FileId>> {
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,
src/main.rs
+2 -8
View File
@@ -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)
src/proto/build.rs
+1 -1
View File
@@ -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",
src/proto/proto/cluster/filelist.proto
-28
View File
@@ -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;
}
src/proto/proto/cluster/search.proto
+12 -6
View File
@@ -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 {
src/proto/src/generated/cluster.rs
+18 -380
View File
@@ -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<T> {
inner: tonic::client::Grpc<T>,
}
impl FilelistClient<tonic::transport::Channel> {
/// Attempt to create a new client by connecting to a given endpoint.
pub async fn connect<D>(dst: D) -> Result<Self, tonic::transport::Error>
where
D: TryInto<tonic::transport::Endpoint>,
D::Error: Into<StdError>,
{
let conn = tonic::transport::Endpoint::new(dst)?.connect().await?;
Ok(Self::new(conn))
}
}
impl<T> FilelistClient<T>
where
T: tonic::client::GrpcService<tonic::body::BoxBody>,
T::Error: Into<StdError>,
T::ResponseBody: Body<Data = Bytes> + Send + 'static,
<T::ResponseBody as Body>::Error: Into<StdError> + 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<F>(
inner: T,
interceptor: F,
) -> FilelistClient<InterceptedService<T, F>>
where
F: tonic::service::Interceptor,
T::ResponseBody: Default,
T: tonic::codegen::Service<
http::Request<tonic::body::BoxBody>,
Response = http::Response<
<T as tonic::client::GrpcService<tonic::body::BoxBody>>::ResponseBody,
>,
>,
<T as tonic::codegen::Service<
http::Request<tonic::body::BoxBody>,
>>::Error: Into<StdError> + 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<super::EmptyRequest>,
) -> std::result::Result<tonic::Response<super::MaxIdResponse>, 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<super::FileListQueryRequest>,
) -> std::result::Result<tonic::Response<super::FileList>, 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<super::EmptyRequest>,
) -> std::result::Result<tonic::Response<super::MaxIdResponse>, tonic::Status>;
async fn query(
&self,
request: tonic::Request<super::FileListQueryRequest>,
) -> std::result::Result<tonic::Response<super::FileList>, tonic::Status>;
}
#[derive(Debug)]
pub struct FilelistServer<T: Filelist> {
inner: _Inner<T>,
accept_compression_encodings: EnabledCompressionEncodings,
send_compression_encodings: EnabledCompressionEncodings,
max_decoding_message_size: Option<usize>,
max_encoding_message_size: Option<usize>,
}
struct _Inner<T>(Arc<T>);
impl<T: Filelist> FilelistServer<T> {
pub fn new(inner: T) -> Self {
Self::from_arc(Arc::new(inner))
}
pub fn from_arc(inner: Arc<T>) -> 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<F>(
inner: T,
interceptor: F,
) -> InterceptedService<Self, F>
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<T, B> tonic::codegen::Service<http::Request<B>> for FilelistServer<T>
where
T: Filelist,
B: Body + Send + 'static,
B::Error: Into<StdError> + Send + 'static,
{
type Response = http::Response<tonic::body::BoxBody>;
type Error = std::convert::Infallible;
type Future = BoxFuture<Self::Response, Self::Error>;
fn poll_ready(
&mut self,
_cx: &mut Context<'_>,
) -> Poll<std::result::Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, req: http::Request<B>) -> Self::Future {
let inner = self.inner.clone();
match req.uri().path() {
"/cluster.Filelist/MaxID" => {
#[allow(non_camel_case_types)]
struct MaxIDSvc<T: Filelist>(pub Arc<T>);
impl<T: Filelist> tonic::server::UnaryService<super::EmptyRequest>
for MaxIDSvc<T> {
type Response = super::MaxIdResponse;
type Future = BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
request: tonic::Request<super::EmptyRequest>,
) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move {
<T as Filelist>::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<T: Filelist>(pub Arc<T>);
impl<
T: Filelist,
> tonic::server::UnaryService<super::FileListQueryRequest>
for QuerySvc<T> {
type Response = super::FileList;
type Future = BoxFuture<
tonic::Response<Self::Response>,
tonic::Status,
>;
fn call(
&mut self,
request: tonic::Request<super::FileListQueryRequest>,
) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move {
<T as Filelist>::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<T: Filelist> Clone for FilelistServer<T> {
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<T: Filelist> Clone for _Inner<T> {
fn clone(&self) -> Self {
Self(Arc::clone(&self.0))
}
}
impl<T: std::fmt::Debug> std::fmt::Debug for _Inner<T> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{:?}", self.0)
}
}
impl<T: Filelist> tonic::server::NamedService for FilelistServer<T> {
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<SearchQuery>,
#[prost(enumeration = "AggregateMode", tag = "6")]
pub agg_mode: i32,
#[prost(message, repeated, tag = "7")]
pub file_list: ::prost::alloc::vec::Vec<FileKey>,
#[prost(int64, tag = "8")]
#[prost(int64, repeated, tag = "7")]
pub file_ids: ::prost::alloc::vec::Vec<i64>,
#[prost(message, repeated, tag = "8")]
pub idx_files: ::prost::alloc::vec::Vec<IdxFileName>,
#[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<u8>>,
}
#[derive(Eq)]
#[derive(serde::Serialize)]
#[allow(clippy::derive_partial_eq_without_eq)]
#[derive(Clone, PartialEq, ::prost::Message)]
src/service/compact/merge.rs
-2
View File
@@ -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))?;
src/service/compact/retention.rs
-3
View File
@@ -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() {
src/service/file_list.rs
+39 -238
View File
@@ -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<Vec<FileKey>, 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<Node> = 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::<Vec<_>>();
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<Vec<FileKey>, 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<Vec<FileKey>, 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<Vec<file_list::FileId>, 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<ScanStats, anyhow::Error> {
let mut stats = ScanStats::new();
src/service/promql/search/grpc/storage.rs
-1
View File
@@ -193,7 +193,6 @@ async fn get_file_list(
time_level,
time_min,
time_max,
true,
)
.await
{
src/service/search/cluster/mod.rs
+114 -207
View File
@@ -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<o2_enterprise::enterprise::search::WorkGroup> = 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<FileKey> {
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<Vec<&FileKey>> {
let mut partitions: Vec<Vec<&FileKey>> = vec![Vec::new(); num_nodes];
let sum_original_size = file_keys
.iter()
.map(|fk| fk.meta.original_size)
.sum::<i64>();
#[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<FileId> {
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<Vec<&FileId>> {
let mut partitions: Vec<Vec<&FileId>> = vec![Vec::new(); num_nodes];
let sum_original_size = file_keys.iter().map(|fk| fk.original_size).sum::<i64>();
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<RoleGroup>,
) -> Vec<Vec<&'a FileKey>> {
) -> Vec<Vec<&'a FileId>> {
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<&FileKey>> = vec![Vec::new(); idx];
let mut partitions: Vec<Vec<&FileId>> = 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<json::Value>) -> Vec<json::Value> {
async fn get_file_list_by_inverted_index(
meta: Arc<super::sql::Sql>,
mut idx_req: cluster_rpc::SearchRequest,
file_list: &[FileKey],
) -> Result<(Vec<FileKey>, 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::<HashMap<_, _>>();
// 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::<Vec<_>>()
@@ -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<i64>> = 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::<Vec<i64>>())
.map(|x| x.iter().map(|v| v.original_size).collect::<Vec<i64>>())
.enumerate()
{
assert_eq!(value.1, expected.get(value.0).unwrap().clone());
src/service/search/grpc/mod.rs
+68 -3
View File
@@ -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<FileKey> = 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<FileKey>, 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,
}
src/service/search/grpc/storage.rs
-1
View File
@@ -304,7 +304,6 @@ async fn get_file_list(
time_level,
time_min,
time_max,
true,
)
.await
{