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

fix: remove order_by from search request (#5994)

Browse Source
This commit is contained in:
Hengfei Yang
2025-02-17 21:48:34 +08:00
committed by GitHub
parent e2b380c83f
commit 41b6a00907
22 changed files with 191 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cli/data/export.rs
+4 -2
View File
@@ -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,
src/config/src/config.rs
+33 -21
View File
@@ -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
src/config/src/meta/search.rs
+2 -8
View File
@@ -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<String>,
#[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<Request, String> {
pub fn to_query_req(&self, search_stream_name: &str) -> Result<Request, String> {
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<Query> 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,
src/handler/http/request/search/mod.rs
+7 -5
View File
@@ -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));
src/handler/http/request/search/multi_streams.rs
+12 -11
View File
@@ -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,
src/handler/http/request/stream/mod.rs
+2 -1
View File
@@ -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()
src/handler/http/request/traces/mod.rs
+1 -2
View File
@@ -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,
src/ingester/src/writer.rs
+2 -1
View File
@@ -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()
src/job/files/parquet.rs
+5
View File
@@ -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))
src/proto/proto/cluster/search.proto
-1
View File
@@ -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;
src/proto/src/generated/cluster.rs
-2
View File
@@ -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")]
src/service/alerts/mod.rs
+1 -2
View File
@@ -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,
src/service/compact/merge.rs
+1
View File
@@ -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,
src/service/search/cluster/flight.rs
+6
View File
@@ -371,6 +371,7 @@ pub async fn run_datafusion(
.collect::<HashMap<_, _>>();
// 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<FileKey>, 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::<HashSet<String>>();
#[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::<Vec<_>>()
.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()
src/service/search/datafusion/optimizer/add_sort_and_limit.rs
+4 -14
View File
@@ -13,7 +13,6 @@
// 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::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<Transformed<LogicalPlan>> {
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();
src/service/search/datafusion/optimizer/limit_join_right_side.rs
+1 -1
View File
@@ -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)))
src/service/search/datafusion/optimizer/mod.rs
+2 -6
View File
@@ -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<Arc<dyn OptimizerRule + Send +
// *********** custom rules ***********
rules.push(Arc::new(RewriteHistogram::new(start_time, end_time)));
if let Some(limit) = limit {
rules.push(Arc::new(AddSortAndLimitRule::new(
limit,
offset,
sql.order_by.first().cloned(),
)));
rules.push(Arc::new(AddSortAndLimitRule::new(limit, offset)));
};
rules.push(Arc::new(AddTimestampRule::new(start_time, end_time)));
#[cfg(feature = "enterprise")]
src/service/search/datafusion/optimizer/utils.rs
+17 -40
View File
@@ -15,7 +15,7 @@
use std::sync::Arc;
use config::{get_config, meta::sql::OrderBy};
use config::get_config;
use datafusion::{
common::{
tree_node::{
@@ -53,16 +53,14 @@ pub fn is_complex_query(plan: &LogicalPlan) -> bool {
pub struct AddSortAndLimit {
pub limit: usize,
pub offset: usize,
pub order_by: Option<(String, OrderBy)>,
pub deduplication_columns: Vec<Column>,
}
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<LogicalPlan>, limit: usize, skip: usize) -> Lo
fn generate_sort_plan(
input: Arc<LogicalPlan>,
limit: usize,
order_by: Option<(String, OrderBy)>,
) -> (LogicalPlan, Option<Arc<DFSchema>>) {
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<LogicalPlan>,
limit: usize,
skip: usize,
order_by: Option<(String, OrderBy)>,
) -> (LogicalPlan, Option<Arc<DFSchema>>) {
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(
src/service/search/datafusion/table_provider/helpers.rs
+2 -1
View File
@@ -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<Arc<ParquetAccessPlan>> {
#[allow(deprecated)]
if config::get_config()
.common
.inverted_index_search_format
src/service/search/grpc/storage.rs
+1
View File
@@ -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 {
src/service/search/sql.rs
+87 -13
View File
@@ -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::<Vec<_>>();
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::<Vec<_>>();
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<Vec<String>, 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<String> {
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<Self::Break> {
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"
);
}
}
src/service/search/super_cluster/follower.rs
+1
View File
@@ -278,6 +278,7 @@ async fn get_inverted_index_file_lists(
match_terms: &[String],
) -> Result<(bool, Vec<FileKey>, 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";