From 41b6a0090791c90c41f4f451ac7461db146264cb Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Mon, 17 Feb 2025 21:48:34 +0800 Subject: [PATCH] fix: remove order_by from search request (#5994) --- src/cli/data/export.rs | 6 +- src/config/src/config.rs | 54 ++++++---- src/config/src/meta/search.rs | 10 +- src/handler/http/request/search/mod.rs | 12 ++- .../http/request/search/multi_streams.rs | 23 ++-- src/handler/http/request/stream/mod.rs | 3 +- src/handler/http/request/traces/mod.rs | 3 +- src/ingester/src/writer.rs | 3 +- src/job/files/parquet.rs | 5 + src/proto/proto/cluster/search.proto | 1 - src/proto/src/generated/cluster.rs | 2 - src/service/alerts/mod.rs | 3 +- src/service/compact/merge.rs | 1 + src/service/search/cluster/flight.rs | 6 ++ .../optimizer/add_sort_and_limit.rs | 18 +--- .../optimizer/limit_join_right_side.rs | 2 +- .../search/datafusion/optimizer/mod.rs | 8 +- .../search/datafusion/optimizer/utils.rs | 57 +++------- .../datafusion/table_provider/helpers.rs | 3 +- src/service/search/grpc/storage.rs | 1 + src/service/search/sql.rs | 100 +++++++++++++++--- src/service/search/super_cluster/follower.rs | 1 + 22 files changed, 191 insertions(+), 131 deletions(-) diff --git a/src/cli/data/export.rs b/src/cli/data/export.rs index 1538ece100..aae785a7a1 100644 --- a/src/cli/data/export.rs +++ b/src/cli/data/export.rs @@ -54,14 +54,16 @@ impl Context for Export { .as_ref() .and_then(|event_type| get_search_event_context_from_request(event_type, &query_map)); let query = search::Query { - sql: format!("select * from {}", table), + sql: format!( + "select * from {} ORDER BY {} ASC", + table, cfg.common.column_timestamp + ), from: 0, size: 100, quick_mode: false, query_type: "".to_owned(), start_time: c.start_time, end_time: c.end_time, - sort_by: Some(format!("{} ASC", cfg.common.column_timestamp)), track_total_hits: false, uses_zo_fn: false, query_fn: None, diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 8cc7e0e594..ab1e16e25b 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -821,24 +821,28 @@ pub struct Common { help = "Toggle inverted index cache." )] pub inverted_index_cache_enabled: bool, + #[deprecated(since = "0.14.3", note = "will be removed in 0.15.0")] #[env_config( name = "ZO_INVERTED_INDEX_SPLIT_CHARS", default = "", help = "Characters which should be used as a delimiter to split the string, default using all ascii punctuations." )] pub inverted_index_split_chars: String, + #[deprecated(since = "0.14.3", note = "will be removed in 0.15.0")] #[env_config( name = "ZO_INVERTED_INDEX_OLD_FORMAT", default = false, help = "Use old format for inverted index, it will generate same stream name for index." )] pub inverted_index_old_format: bool, + #[deprecated(since = "0.14.3", note = "will be removed in 0.15.0")] #[env_config( name = "ZO_INVERTED_INDEX_STORE_FORMAT", default = "tantivy", help = "InvertedIndex store format, parquet(default), tantivy, both" )] pub inverted_index_store_format: String, + #[deprecated(since = "0.14.3", note = "will be removed in 0.15.0")] #[env_config( name = "ZO_INVERTED_INDEX_SEARCH_FORMAT", default = "tantivy", @@ -857,6 +861,7 @@ pub struct Common { help = "Toggle inverted index count optimizer." )] pub inverted_index_count_optimizer_enabled: bool, + #[deprecated(since = "0.14.3", note = "will be removed in 0.15.0")] #[env_config( name = "ZO_FULL_TEXT_SEARCH_TYPE", default = "eq", @@ -1953,27 +1958,34 @@ fn check_common_config(cfg: &mut Config) -> Result<(), anyhow::Error> { } // check default inverted index search format - cfg.common.inverted_index_store_format = cfg.common.inverted_index_store_format.to_lowercase(); - if cfg.common.inverted_index_store_format.is_empty() { - cfg.common.inverted_index_store_format = "parquet".to_string(); - } - if !["both", "parquet", "tantivy"].contains(&cfg.common.inverted_index_store_format.as_str()) { - return Err(anyhow::anyhow!( - "ZO_INVERTED_INDEX_STORE_FORMAT must be one of parquet, tantivy, both." - )); - } - cfg.common.inverted_index_search_format = - cfg.common.inverted_index_search_format.to_lowercase(); - if cfg.common.inverted_index_search_format.is_empty() { - cfg.common.inverted_index_search_format = cfg.common.inverted_index_store_format.clone(); - } - if cfg.common.inverted_index_search_format == "both" { - cfg.common.inverted_index_search_format = "parquet".to_string(); - } - if !["parquet", "tantivy"].contains(&cfg.common.inverted_index_search_format.as_str()) { - return Err(anyhow::anyhow!( - "ZO_INVERTED_INDEX_SEARCH_FORMAT must be one of parquet, tantivy." - )); + #[allow(deprecated)] + { + cfg.common.inverted_index_store_format = + cfg.common.inverted_index_store_format.to_lowercase(); + if cfg.common.inverted_index_store_format.is_empty() { + cfg.common.inverted_index_store_format = "parquet".to_string(); + } + if !["both", "parquet", "tantivy"] + .contains(&cfg.common.inverted_index_store_format.as_str()) + { + return Err(anyhow::anyhow!( + "ZO_INVERTED_INDEX_STORE_FORMAT must be one of parquet, tantivy, both." + )); + } + cfg.common.inverted_index_search_format = + cfg.common.inverted_index_search_format.to_lowercase(); + if cfg.common.inverted_index_search_format.is_empty() { + cfg.common.inverted_index_search_format = + cfg.common.inverted_index_store_format.clone(); + } + if cfg.common.inverted_index_search_format == "both" { + cfg.common.inverted_index_search_format = "parquet".to_string(); + } + if !["parquet", "tantivy"].contains(&cfg.common.inverted_index_search_format.as_str()) { + return Err(anyhow::anyhow!( + "ZO_INVERTED_INDEX_SEARCH_FORMAT must be one of parquet, tantivy." + )); + } } // check for join match one diff --git a/src/config/src/meta/search.rs b/src/config/src/meta/search.rs index 82f6567871..5bec968f87 100644 --- a/src/config/src/meta/search.rs +++ b/src/config/src/meta/search.rs @@ -1,4 +1,4 @@ -// Copyright 2024 OpenObserve Inc. +// Copyright 2025 OpenObserve Inc. // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by @@ -106,8 +106,6 @@ pub struct Query { #[serde(default)] pub end_time: i64, #[serde(default)] - pub sort_by: Option, - #[serde(default)] pub quick_mode: bool, #[serde(default)] pub query_type: String, @@ -138,7 +136,6 @@ impl Default for Query { size: 10, start_time: 0, end_time: 0, - sort_by: None, quick_mode: false, query_type: "".to_string(), track_total_hits: false, @@ -482,7 +479,7 @@ impl SearchHistoryRequest { Ok(query) } - pub fn to_query_req(&self, search_stream_name: &str, sort_by: &str) -> Result { + pub fn to_query_req(&self, search_stream_name: &str) -> Result { let sql = self.build_query(search_stream_name)?; let search_req = Request { @@ -492,7 +489,6 @@ impl SearchHistoryRequest { size: self.size, start_time: self.start_time, end_time: self.end_time, - sort_by: Some(format!("{} DESC", sort_by)), quick_mode: false, query_type: "".to_string(), track_total_hits: false, @@ -694,7 +690,6 @@ impl From for cluster_rpc::SearchQuery { size: query.size as i32, start_time: query.start_time, end_time: query.end_time, - sort_by: query.sort_by.unwrap_or_default(), track_total_hits: query.track_total_hits, uses_zo_fn: query.uses_zo_fn, query_fn: query.query_fn.unwrap_or_default(), @@ -1023,7 +1018,6 @@ impl MultiStreamRequest { size: self.size, start_time: query.start_time.unwrap_or(self.start_time), end_time: query.end_time.unwrap_or(self.end_time), - sort_by: self.sort_by.clone(), quick_mode: self.quick_mode, query_type: self.query_type.clone(), track_total_hits: self.track_total_hits, diff --git a/src/handler/http/request/search/mod.rs b/src/handler/http/request/search/mod.rs index 2b0b36dd44..ffafa031f1 100644 --- a/src/handler/http/request/search/mod.rs +++ b/src/handler/http/request/search/mod.rs @@ -526,14 +526,15 @@ pub async fn around( .unwrap(); // search forward + let fw_sql = SearchService::sql::check_or_add_order_by_timestamp(&around_sql, false) + .unwrap_or(around_sql.to_string()); let req = config::meta::search::Request { query: config::meta::search::Query { - sql: around_sql.clone(), + sql: fw_sql, from: 0, size: around_size / 2, start_time: around_start_time, end_time: around_key, - sort_by: Some(format!("{} DESC", cfg.common.column_timestamp)), quick_mode: false, query_type: "".to_string(), track_total_hits: false, @@ -580,14 +581,15 @@ pub async fn around( }; // search backward + let bw_sql = SearchService::sql::check_or_add_order_by_timestamp(&around_sql, true) + .unwrap_or(around_sql.to_string()); let req = config::meta::search::Request { query: config::meta::search::Query { - sql: around_sql.clone(), + sql: bw_sql, from: 0, size: around_size / 2, start_time: around_key, end_time: around_end_time, - sort_by: Some(format!("{} ASC", cfg.common.column_timestamp)), quick_mode: false, query_type: "".to_string(), track_total_hits: false, @@ -1329,7 +1331,7 @@ pub async fn search_history( // Search let stream_name = USAGE_STREAM; - let search_query_req = match req.to_query_req(stream_name, &cfg.common.column_timestamp) { + let search_query_req = match req.to_query_req(stream_name) { Ok(r) => r, Err(e) => { return Ok(MetaHttpResponse::bad_request(e)); diff --git a/src/handler/http/request/search/multi_streams.rs b/src/handler/http/request/search/multi_streams.rs index 92fbdfebeb..2e357e75a9 100644 --- a/src/handler/http/request/search/multi_streams.rs +++ b/src/handler/http/request/search/multi_streams.rs @@ -1,4 +1,4 @@ -// Copyright 2024 OpenObserve Inc. +// Copyright 2025 OpenObserve Inc. // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by @@ -47,10 +47,7 @@ use crate::{ stream::get_settings_max_query_range, }, }, - service::{ - search::{self as SearchService, RESULT_ARRAY}, - self_reporting::report_request_usage_stats, - }, + service::{search as SearchService, self_reporting::report_request_usage_stats}, }; /// SearchStreamData @@ -502,9 +499,11 @@ pub async fn search_multi( // compile vrl function & apply the same before returning the response let mut input_fn = query_fn.unwrap().trim().to_string(); - let apply_over_hits = RESULT_ARRAY.is_match(&input_fn); + let apply_over_hits = SearchService::RESULT_ARRAY.is_match(&input_fn); if apply_over_hits { - input_fn = RESULT_ARRAY.replace(&input_fn, "").to_string(); + input_fn = SearchService::RESULT_ARRAY + .replace(&input_fn, "") + .to_string(); } let mut runtime = crate::common::utils::functions::init_vrl_runtime(); let program = match crate::service::ingestion::compile_vrl_function(&input_fn, &org_id) { @@ -971,14 +970,15 @@ pub async fn around_multi( .dec(); // search forward + let fw_sql = SearchService::sql::check_or_add_order_by_timestamp(around_sql, false) + .unwrap_or(around_sql.to_string()); let req = config::meta::search::Request { query: config::meta::search::Query { - sql: around_sql.clone(), + sql: fw_sql, from: 0, size: around_size / 2, start_time: around_start_time, end_time: around_key, - sort_by: Some(format!("{} DESC", cfg.common.column_timestamp)), quick_mode: false, query_type: "".to_string(), track_total_hits: false, @@ -1047,14 +1047,15 @@ pub async fn around_multi( }; // search backward + let bw_sql = SearchService::sql::check_or_add_order_by_timestamp(around_sql, true) + .unwrap_or(around_sql.to_string()); let req = config::meta::search::Request { query: config::meta::search::Query { - sql: around_sql.clone(), + sql: bw_sql, from: 0, size: around_size / 2, start_time: around_key, end_time: around_end_time, - sort_by: Some(format!("{} ASC", cfg.common.column_timestamp)), quick_mode: false, query_type: "".to_string(), track_total_hits: false, diff --git a/src/handler/http/request/stream/mod.rs b/src/handler/http/request/stream/mod.rs index 6b4b583afb..4e0a44fc8b 100644 --- a/src/handler/http/request/stream/mod.rs +++ b/src/handler/http/request/stream/mod.rs @@ -1,4 +1,4 @@ -// Copyright 2024 OpenObserve Inc. +// Copyright 2025 OpenObserve Inc. // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by @@ -201,6 +201,7 @@ async fn update_settings( // sync the data retention to index stream if stream_type.is_basic_type() && stream_settings.data_retention.is_some() { + #[allow(deprecated)] let index_stream_name = if cfg.common.inverted_index_old_format && stream_type == StreamType::Logs { stream_name.to_string() diff --git a/src/handler/http/request/traces/mod.rs b/src/handler/http/request/traces/mod.rs index 9a296cf8e7..12dc941f6b 100644 --- a/src/handler/http/request/traces/mod.rs +++ b/src/handler/http/request/traces/mod.rs @@ -1,4 +1,4 @@ -// Copyright 2024 OpenObserve Inc. +// Copyright 2025 OpenObserve Inc. // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by @@ -280,7 +280,6 @@ pub async fn get_latest_traces( size, start_time, end_time, - sort_by: None, quick_mode: false, query_type: "".to_string(), track_total_hits: false, diff --git a/src/ingester/src/writer.rs b/src/ingester/src/writer.rs index bbf201c1bc..ff3c6811ef 100644 --- a/src/ingester/src/writer.rs +++ b/src/ingester/src/writer.rs @@ -493,9 +493,10 @@ impl Writer { /// Check if the wal file size is over the threshold or the file is too old fn check_wal_threshold(&self, written_size: (usize, usize), data_size: usize) -> bool { let cfg = get_config(); - let (compressed_size, _uncompressed_size) = written_size; + let (compressed_size, uncompressed_size) = written_size; compressed_size > wal::FILE_TYPE_IDENTIFIER_LEN && (compressed_size + data_size > cfg.limit.max_file_size_on_disk + || uncompressed_size + data_size > cfg.limit.max_file_size_on_disk || self.created_at.load(Ordering::Relaxed) + Duration::try_seconds(cfg.limit.max_file_retention_time as i64) .unwrap() diff --git a/src/job/files/parquet.rs b/src/job/files/parquet.rs index a3a725d23c..1e0b809259 100644 --- a/src/job/files/parquet.rs +++ b/src/job/files/parquet.rs @@ -791,6 +791,7 @@ async fn merge_files( } // generate parquet format inverted index + #[allow(deprecated)] let index_format = InvertedIndexFormat::from(&cfg.common.inverted_index_store_format); if matches!( index_format, @@ -852,6 +853,7 @@ pub(crate) async fn generate_index_on_ingester( } let cfg = get_config(); + #[allow(deprecated)] let index_stream_name = if cfg.common.inverted_index_old_format && stream_type == StreamType::Logs { stream_name.to_string() @@ -920,6 +922,7 @@ pub(crate) async fn generate_index_on_ingester( .await; } else if let Some(schema) = schema_map.get(&index_stream_name) { // check if the schema has been updated <= v0.10.8-rc4 + #[allow(deprecated)] if cfg.common.inverted_index_old_format && stream_type == StreamType::Logs && !schema.fields_map().contains_key("segment_ids") @@ -1033,6 +1036,7 @@ pub(crate) async fn generate_index_on_compactor( return Ok(vec![]); } + #[allow(deprecated)] let index_stream_name = if get_config().common.inverted_index_old_format && stream_type == StreamType::Logs { stream_name.to_string() @@ -1207,6 +1211,7 @@ async fn prepare_index_record_batches( // split the column into terms let terms = (0..num_rows) .flat_map(|i| { + #[allow(deprecated)] split_token(column_data.value(i), &cfg.common.inverted_index_split_chars) .into_iter() .map(|s| (s, i)) diff --git a/src/proto/proto/cluster/search.proto b/src/proto/proto/cluster/search.proto index 83c6f7ba88..18adbcba51 100644 --- a/src/proto/proto/cluster/search.proto +++ b/src/proto/proto/cluster/search.proto @@ -81,7 +81,6 @@ message SearchQuery { int32 size = 5; int64 start_time = 6; int64 end_time = 7; - string sort_by = 8; bool track_total_hits = 9; string query_type = 10; bool uses_zo_fn = 12; diff --git a/src/proto/src/generated/cluster.rs b/src/proto/src/generated/cluster.rs index e213d27006..678bb3a5a0 100644 --- a/src/proto/src/generated/cluster.rs +++ b/src/proto/src/generated/cluster.rs @@ -881,8 +881,6 @@ pub struct SearchQuery { pub start_time: i64, #[prost(int64, tag = "7")] pub end_time: i64, - #[prost(string, tag = "8")] - pub sort_by: ::prost::alloc::string::String, #[prost(bool, tag = "9")] pub track_total_hits: bool, #[prost(string, tag = "10")] diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index d3c6dee95a..27c10cd890 100644 --- a/src/service/alerts/mod.rs +++ b/src/service/alerts/mod.rs @@ -1,4 +1,4 @@ -// Copyright 2024 OpenObserve Inc. +// Copyright 2025 OpenObserve Inc. // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by @@ -343,7 +343,6 @@ impl QueryConditionExt for QueryCondition { size, start_time: start_time.unwrap(), end_time, - sort_by: None, quick_mode: false, query_type: "".to_string(), track_total_hits: false, diff --git a/src/service/compact/merge.rs b/src/service/compact/merge.rs index 3ead1ec8cb..cd39f34be8 100644 --- a/src/service/compact/merge.rs +++ b/src/service/compact/merge.rs @@ -1042,6 +1042,7 @@ async fn generate_inverted_index( let cfg = get_config(); // generate parquet format inverted index + #[allow(deprecated)] let index_format = InvertedIndexFormat::from(&cfg.common.inverted_index_store_format); if matches!( index_format, diff --git a/src/service/search/cluster/flight.rs b/src/service/search/cluster/flight.rs index 71da259299..bd6ae69b19 100644 --- a/src/service/search/cluster/flight.rs +++ b/src/service/search/cluster/flight.rs @@ -371,6 +371,7 @@ pub async fn run_datafusion( .collect::>(); // check inverted index prefix search + #[allow(deprecated)] if sql.stream_type == StreamType::Index && cfg.common.full_text_search_type.to_lowercase() != "contains" { @@ -825,6 +826,7 @@ async fn get_inverted_index_file_lists( query: &SearchQuery, ) -> Result<(bool, Vec, usize, usize)> { let cfg = get_config(); + #[allow(deprecated)] let inverted_index_type = cfg.common.inverted_index_search_format.clone(); let (use_inverted_index, index_terms) = super::super::is_use_inverted_index(sql); let use_parquet_inverted_index = use_inverted_index && inverted_index_type == "parquet"; @@ -882,6 +884,7 @@ pub async fn get_inverted_index_file_list( let terms = match_terms .iter() .filter_map(|t| { + #[allow(deprecated)] let tokens = split_token(t, &cfg.common.inverted_index_split_chars); if tokens.is_empty() { None @@ -896,6 +899,7 @@ pub async fn get_inverted_index_file_list( }) .collect::>(); + #[allow(deprecated)] let fts_condition = terms .iter() .map(|x| match cfg.common.full_text_search_type.as_str() { @@ -905,6 +909,7 @@ pub async fn get_inverted_index_file_list( }) .collect::>() .join(" OR "); + #[allow(deprecated)] let fts_condition = if fts_condition.is_empty() { fts_condition } else if cfg.common.inverted_index_old_format && stream_type == StreamType::Logs { @@ -946,6 +951,7 @@ pub async fn get_inverted_index_file_list( } }; + #[allow(deprecated)] let index_stream_name = if get_config().common.inverted_index_old_format && stream_type == StreamType::Logs { stream_name.to_string() diff --git a/src/service/search/datafusion/optimizer/add_sort_and_limit.rs b/src/service/search/datafusion/optimizer/add_sort_and_limit.rs index 6657367a25..3f585d6c50 100644 --- a/src/service/search/datafusion/optimizer/add_sort_and_limit.rs +++ b/src/service/search/datafusion/optimizer/add_sort_and_limit.rs @@ -13,7 +13,6 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . -use config::meta::sql::OrderBy; use datafusion::{ common::{ tree_node::{Transformed, TreeNode, TreeNodeRecursion}, @@ -29,16 +28,11 @@ use super::utils::AddSortAndLimit; pub struct AddSortAndLimitRule { limit: usize, offset: usize, - order_by: Option<(String, OrderBy)>, } impl AddSortAndLimitRule { - pub fn new(limit: usize, offset: usize, order_by: Option<(String, OrderBy)>) -> Self { - Self { - limit, - offset, - order_by, - } + pub fn new(limit: usize, offset: usize) -> Self { + Self { limit, offset } } } @@ -60,11 +54,7 @@ impl OptimizerRule for AddSortAndLimitRule { plan: LogicalPlan, _config: &dyn OptimizerConfig, ) -> Result> { - let mut plan = plan.rewrite(&mut AddSortAndLimit::new( - self.limit, - self.offset, - self.order_by.clone(), - ))?; + let mut plan = plan.rewrite(&mut AddSortAndLimit::new(self.limit, self.offset))?; plan.tnr = TreeNodeRecursion::Stop; Ok(plan) } @@ -202,7 +192,7 @@ mod tests { let ctx = SessionContext::new(); let provider = MemTable::try_new(schema, vec![vec![batch]]).unwrap(); ctx.register_table("t", Arc::new(provider)).unwrap(); - ctx.add_optimizer_rule(Arc::new(AddSortAndLimitRule::new(2, 0, None))); + ctx.add_optimizer_rule(Arc::new(AddSortAndLimitRule::new(2, 0))); for item in sqls { let df = ctx.sql(item.0).await.unwrap(); diff --git a/src/service/search/datafusion/optimizer/limit_join_right_side.rs b/src/service/search/datafusion/optimizer/limit_join_right_side.rs index 895f691809..95902a0932 100644 --- a/src/service/search/datafusion/optimizer/limit_join_right_side.rs +++ b/src/service/search/datafusion/optimizer/limit_join_right_side.rs @@ -73,7 +73,7 @@ impl OptimizerRule for LimitJoinRightSide { if right_column.is_empty() { let plan = (*join.right) .clone() - .rewrite(&mut AddSortAndLimit::new(self.limit, 0, None))? + .rewrite(&mut AddSortAndLimit::new(self.limit, 0))? .data; join.right = Arc::new(plan); Ok(Transformed::yes(LogicalPlan::Join(join))) diff --git a/src/service/search/datafusion/optimizer/mod.rs b/src/service/search/datafusion/optimizer/mod.rs index c30059caa0..8a17a79a64 100644 --- a/src/service/search/datafusion/optimizer/mod.rs +++ b/src/service/search/datafusion/optimizer/mod.rs @@ -1,4 +1,4 @@ -// Copyright 2024 OpenObserve Inc. +// Copyright 2025 OpenObserve Inc. // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by @@ -111,11 +111,7 @@ pub fn generate_optimizer_rules(sql: &Sql) -> Vec bool { pub struct AddSortAndLimit { pub limit: usize, pub offset: usize, - pub order_by: Option<(String, OrderBy)>, pub deduplication_columns: Vec, } impl AddSortAndLimit { - pub fn new(limit: usize, offset: usize, order_by: Option<(String, OrderBy)>) -> Self { + pub fn new(limit: usize, offset: usize) -> Self { Self { limit, offset, - order_by, deduplication_columns: vec![], } } @@ -75,7 +73,6 @@ impl AddSortAndLimit { Self { limit, offset, - order_by: None, deduplication_columns, } } @@ -109,11 +106,7 @@ impl TreeNodeRewriter for AddSortAndLimit { // the add sort plan should reflect the limit let fetch = get_int_from_expr(&limit.fetch); let skip = get_int_from_expr(&limit.skip); - let (sort, schema) = generate_sort_plan( - limit.input.clone(), - fetch + skip, - self.order_by.clone(), - ); + let (sort, schema) = generate_sort_plan(limit.input.clone(), fetch + skip); limit.input = Arc::new(sort); (Transformed::yes(LogicalPlan::Limit(limit)), schema) } @@ -142,12 +135,8 @@ impl TreeNodeRewriter for AddSortAndLimit { None, ) } else { - let (plan, schema) = generate_limit_and_sort_plan( - Arc::new(node), - self.limit, - self.offset, - self.order_by.clone(), - ); + let (plan, schema) = + generate_limit_and_sort_plan(Arc::new(node), self.limit, self.offset); (Transformed::yes(plan), schema) } } @@ -220,29 +209,18 @@ fn generate_limit_plan(input: Arc, limit: usize, skip: usize) -> Lo fn generate_sort_plan( input: Arc, limit: usize, - order_by: Option<(String, OrderBy)>, ) -> (LogicalPlan, Option>) { - let cfg = get_config(); - let (sort_expr, sort_field) = match order_by { - Some((field, order_by)) => ( - SortExpr { - expr: col(field.clone()), - asc: order_by == OrderBy::Asc, - nulls_first: false, - }, - field, - ), - None => ( - SortExpr { - expr: col(cfg.common.column_timestamp.clone()), - asc: false, - nulls_first: false, - }, - cfg.common.column_timestamp.clone(), - ), + let config = get_config(); + let timestamp = SortExpr { + expr: col(config.common.column_timestamp.clone()), + asc: false, + nulls_first: false, }; let schema = input.schema().clone(); - if schema.field_with_name(None, sort_field.as_str()).is_err() { + if schema + .field_with_name(None, config.common.column_timestamp.as_str()) + .is_err() + { let mut input = input.as_ref().clone(); input = input .rewrite(&mut ChangeTableScanSchema::new()) @@ -250,7 +228,7 @@ fn generate_sort_plan( .unwrap(); ( LogicalPlan::Sort(Sort { - expr: vec![sort_expr], + expr: vec![timestamp], input: Arc::new(input), fetch: Some(limit), }), @@ -261,7 +239,7 @@ fn generate_sort_plan( input = input.rewrite(&mut SortByTime::new()).data().unwrap(); ( LogicalPlan::Sort(Sort { - expr: vec![sort_expr], + expr: vec![timestamp], input: Arc::new(input), fetch: Some(limit), }), @@ -274,9 +252,8 @@ fn generate_limit_and_sort_plan( input: Arc, limit: usize, skip: usize, - order_by: Option<(String, OrderBy)>, ) -> (LogicalPlan, Option>) { - let (sort, schema) = generate_sort_plan(input, limit + skip, order_by); + let (sort, schema) = generate_sort_plan(input, limit + skip); ( LogicalPlan::Limit(Limit { skip: Some(Box::new(Expr::Literal(ScalarValue::Int64(Some( diff --git a/src/service/search/datafusion/table_provider/helpers.rs b/src/service/search/datafusion/table_provider/helpers.rs index 691ac1a701..31acdd72ec 100644 --- a/src/service/search/datafusion/table_provider/helpers.rs +++ b/src/service/search/datafusion/table_provider/helpers.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -// Copyright 2024 OpenObserve Inc. +// Copyright 2025 OpenObserve Inc. // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by @@ -171,6 +171,7 @@ pub fn split_files( } pub fn generate_access_plan(file: &PartitionedFile) -> Option> { + #[allow(deprecated)] if config::get_config() .common .inverted_index_search_format diff --git a/src/service/search/grpc/storage.rs b/src/service/search/grpc/storage.rs index e13d2b8da6..38255efb1e 100644 --- a/src/service/search/grpc/storage.rs +++ b/src/service/search/grpc/storage.rs @@ -123,6 +123,7 @@ pub async fn search( // check inverted index let cfg = get_config(); + #[allow(deprecated)] let inverted_index_type = cfg.common.inverted_index_search_format.clone(); let use_inverted_index = query.use_inverted_index && inverted_index_type == "tantivy"; if use_inverted_index { diff --git a/src/service/search/sql.rs b/src/service/search/sql.rs index b8bdab8dea..e6435d152e 100644 --- a/src/service/search/sql.rs +++ b/src/service/search/sql.rs @@ -43,8 +43,9 @@ use regex::Regex; use sqlparser::{ ast::{ BinaryOperator, DuplicateTreatment, Expr, Function, FunctionArg, FunctionArgExpr, - FunctionArgumentList, FunctionArguments, GroupByExpr, Ident, ObjectName, Query, Select, - SelectItem, SetExpr, Statement, TableFactor, TableWithJoins, VisitMut, VisitorMut, + FunctionArgumentList, FunctionArguments, GroupByExpr, Ident, ObjectName, OrderByExpr, + Query, Select, SelectItem, SetExpr, Statement, TableFactor, TableWithJoins, VisitMut, + VisitorMut, }, dialect::PostgreSqlDialect, parser::Parser, @@ -150,17 +151,6 @@ impl Sql { .collect::>(); let group_by = column_visitor.group_by; let mut order_by = column_visitor.order_by; - if order_by.is_empty() && !query.sort_by.is_empty() { - // example: sort_by: "timestamp ASC" - let sort_by = query.sort_by.split_whitespace().collect::>(); - if sort_by.len() >= 2 { - if sort_by[1].to_lowercase() == "desc" { - order_by.push((sort_by[0].to_string(), OrderBy::Desc)); - } else { - order_by.push((sort_by[0].to_string(), OrderBy::Asc)); - } - } - } // check if need sort by time if order_by.is_empty() @@ -253,6 +243,7 @@ impl Sql { // 11. generate tantivy query let mut index_condition = None; let mut can_optimize = false; + #[allow(deprecated)] if cfg.common.inverted_index_search_format.eq("tantivy") && stream_names.len() == 1 && cfg.common.inverted_index_enabled @@ -1670,6 +1661,51 @@ pub fn get_cipher_key_names(sql: &str) -> Result, Error> { } } +/// check if the sql is complex query, if not, add ordering term by timestamp +pub fn check_or_add_order_by_timestamp(sql: &str, is_asc: bool) -> infra::errors::Result { + let mut statement = Parser::parse_sql(&PostgreSqlDialect {}, sql) + .map_err(|e| Error::Message(e.to_string()))? + .pop() + .unwrap(); + if is_complex_query(&mut statement) { + return Ok(sql.to_string()); + } + let mut visitor = + AddOrderingTermVisitor::new(get_config().common.column_timestamp.to_string(), is_asc); + statement.visit(&mut visitor); + Ok(statement.to_string()) +} + +struct AddOrderingTermVisitor { + field: String, + is_asc: bool, +} + +impl AddOrderingTermVisitor { + fn new(field: String, is_asc: bool) -> Self { + Self { field, is_asc } + } +} + +impl VisitorMut for AddOrderingTermVisitor { + type Break = (); + + fn pre_visit_query(&mut self, query: &mut Query) -> ControlFlow { + if query.order_by.is_none() { + query.order_by = Some(sqlparser::ast::OrderBy { + exprs: vec![OrderByExpr { + expr: Expr::Identifier(Ident::new(self.field.clone())), + asc: Some(self.is_asc), + nulls_first: None, + with_fill: None, + }], + interpolate: None, + }); + } + ControlFlow::Continue(()) + } +} + #[cfg(test)] mod tests { @@ -1917,4 +1953,42 @@ mod tests { .unwrap(); assert_eq!(is_simple_count_query(&mut statement), false); } + + #[test] + fn test_check_or_add_order_by_timestamp_no_order_asc() { + let sql = "SELECT * FROM logs"; + let result = check_or_add_order_by_timestamp(sql, true).unwrap(); + assert_eq!(result, "SELECT * FROM logs ORDER BY _timestamp ASC"); + } + + #[test] + fn test_check_or_add_order_by_timestamp_no_order_desc() { + let sql = "SELECT * FROM logs"; + let result = check_or_add_order_by_timestamp(sql, false).unwrap(); + assert_eq!(result, "SELECT * FROM logs ORDER BY _timestamp DESC"); + } + + #[test] + fn test_check_or_add_order_by_timestamp_aggregation() { + let sql = "SELECT COUNT(*) FROM logs"; + let result = check_or_add_order_by_timestamp(sql, true).unwrap(); + assert_eq!(result, "SELECT COUNT(*) FROM logs"); + } + + #[test] + fn test_check_or_add_order_by_timestamp_existing_order() { + let sql = "SELECT * FROM logs ORDER BY field1 DESC"; + let result = check_or_add_order_by_timestamp(sql, true).unwrap(); + assert_eq!(sql, result); + } + + #[test] + fn test_check_or_add_order_by_timestamp_with_where() { + let sql = "SELECT * FROM logs WHERE field1 = 'value'"; + let result = check_or_add_order_by_timestamp(sql, true).unwrap(); + assert_eq!( + result, + "SELECT * FROM logs WHERE field1 = 'value' ORDER BY _timestamp ASC" + ); + } } diff --git a/src/service/search/super_cluster/follower.rs b/src/service/search/super_cluster/follower.rs index 0415457492..f762452213 100644 --- a/src/service/search/super_cluster/follower.rs +++ b/src/service/search/super_cluster/follower.rs @@ -278,6 +278,7 @@ async fn get_inverted_index_file_lists( match_terms: &[String], ) -> Result<(bool, Vec, usize, usize)> { let cfg = config::get_config(); + #[allow(deprecated)] let inverted_index_type = cfg.common.inverted_index_search_format.clone(); let use_inverted_index = req.use_inverted_index; let use_parquet_inverted_index = use_inverted_index && inverted_index_type == "parquet";