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

fix: simpel select asc issue and distinct time range (#11715)

Browse Source 
- [x] simple select do not consider asc
- [x] simple distinct do not consider time range filter
This commit is contained in:
Huaijin
2026-05-11 08:53:05 +00:00
committed by GitHub
parent 3e070bff8c
commit 627d620a2f
2 changed files with 61 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/service/search/grpc/flight.rs
+24 -8
View File
@@ -252,6 +252,7 @@ pub async fn search(
&mut storage_idx_optimize_rule, // pass by mutable reference
file_list,
index_updated_at,
query_params.time_range,
)
.await?;
log::info!(
@@ -672,6 +673,7 @@ async fn handle_tantivy_optimize(
idx_optimize_rule: &mut Option<IndexOptimizeMode>,
file_list: Vec<FileKey>,
index_updated_at: i64,
time_range: (i64, i64),
) -> Result<(Vec<FileKey>, Vec<FileKey>), Error> {
// early return if not simple count, histogram or topn
if !matches!(
@@ -686,8 +688,18 @@ async fn handle_tantivy_optimize(
let index_updated_at = update_index_updated_at(idx_optimize_rule, index_updated_at).await;
// TODO: support IndexOptimizeMode::SimpleDistinct for add timestamp
// filter to tantivy search
let time_range = if matches!(
idx_optimize_rule,
Some(IndexOptimizeMode::SimpleDistinct(..))
) {
Some(time_range)
} else {
None
};
let (tantivy_files, datafusion_files) =
split_file_list_by_time_range(file_list, index_updated_at);
split_file_list_by_time_range(file_list, index_updated_at, time_range);
// set optimize rule to None, because datafusion should not use it
*idx_optimize_rule = None;
@@ -714,10 +726,14 @@ async fn update_index_updated_at(
fn split_file_list_by_time_range(
file_list: Vec<FileKey>,
index_updated_at: i64,
time_range: Option<(i64, i64)>,
) -> (Vec<FileKey>, Vec<FileKey>) {
file_list
.into_iter()
.partition(|file| file.meta.min_ts >= index_updated_at && file.meta.index_size > 0)
file_list.into_iter().partition(|file| {
file.meta.min_ts >= index_updated_at
&& file.meta.index_size > 0
&& time_range
.is_none_or(|(start, end)| file.meta.min_ts >= start && file.meta.max_ts <= end)
})
}
fn collect_stats(files: &[FileKey]) -> ScanStats {
@@ -769,7 +785,7 @@ mod tests {
#[test]
fn test_split_file_list_empty() {
let (tantivy, datafusion) = split_file_list_by_time_range(vec![], 0);
let (tantivy, datafusion) = split_file_list_by_time_range(vec![], 0, None);
assert!(tantivy.is_empty());
assert!(datafusion.is_empty());
}
@@ -777,7 +793,7 @@ mod tests {
#[test]
fn test_split_file_list_all_tantivy() {
let files = vec![make_file(100, 200, 512), make_file(300, 400, 1024)];
let (tantivy, datafusion) = split_file_list_by_time_range(files, 0);
let (tantivy, datafusion) = split_file_list_by_time_range(files, 0, None);
assert_eq!(tantivy.len(), 2);
assert!(datafusion.is_empty());
}
@@ -785,7 +801,7 @@ mod tests {
#[test]
fn test_split_file_list_no_index_goes_to_datafusion() {
let files = vec![make_file(100, 200, 0)]; // index_size == 0
let (tantivy, datafusion) = split_file_list_by_time_range(files, 0);
let (tantivy, datafusion) = split_file_list_by_time_range(files, 0, None);
assert!(tantivy.is_empty());
assert_eq!(datafusion.len(), 1);
}
@@ -793,7 +809,7 @@ mod tests {
#[test]
fn test_split_file_list_before_index_updated_at() {
let files = vec![make_file(100, 200, 512)];
let (tantivy, datafusion) = split_file_list_by_time_range(files, 500);
let (tantivy, datafusion) = split_file_list_by_time_range(files, 500, None);
assert!(tantivy.is_empty());
assert_eq!(datafusion.len(), 1);
}
src/service/search/grpc/storage.rs
+37 -11
View File
@@ -966,22 +966,22 @@ fn partition_tantivy_files(
idx_optimize_mode: &Option<IndexOptimizeMode>,
target_partitions: usize,
) -> (Vec<Vec<FileKey>>, usize) {
let (file_groups, limit) = if let Some(IndexOptimizeMode::SimpleSelect(limit, _ascend)) =
let (file_groups, limit, ascend) = if let Some(IndexOptimizeMode::SimpleSelect(limit, ascend)) =
idx_optimize_mode
&& *limit > 0
{
let file_groups = group_files_by_time_range(index_parquet_files, target_partitions);
(file_groups, *limit)
(file_groups, *limit, *ascend)
} else {
// splite the filter groups by target partitions
let file_groups = into_chunks(index_parquet_files, target_partitions);
(file_groups, 0)
(file_groups, 0, false)
};
if limit == 0 {
(file_groups, limit)
} else {
(regroup_tantivy_files(file_groups), limit)
(regroup_tantivy_files(file_groups, ascend), limit)
}
}
@@ -990,7 +990,7 @@ fn partition_tantivy_files(
// group's time range not overlap, when execute the tantivy search, we get the last file in each
// group and do the tantivy search.
// so in this function, we recursive collect the last file in each group
fn regroup_tantivy_files(file_groups: Vec<Vec<FileKey>>) -> Vec<Vec<FileKey>> {
fn regroup_tantivy_files(file_groups: Vec<Vec<FileKey>>, ascend: bool) -> Vec<Vec<FileKey>> {
let group_num = file_groups.len();
let max_group_len = file_groups.iter().map(|g| g.len()).max().unwrap_or(0);
let mut new_file_groups: Vec<Vec<FileKey>> = vec![Vec::new(); max_group_len];
@@ -998,7 +998,9 @@ fn regroup_tantivy_files(file_groups: Vec<Vec<FileKey>>) -> Vec<Vec<FileKey>> {
let mut file_groups: Vec<_> = file_groups
.into_iter()
.map(|mut group| {
group.reverse();
if !ascend {
group.reverse();
}
group.into_iter()
})
.collect();
@@ -1310,7 +1312,7 @@ mod tests {
vec![create_file_key(1, 10), create_file_key(11, 20)],
vec![create_file_key(21, 30), create_file_key(31, 40)],
];
let result = regroup_tantivy_files(file_groups);
let result = regroup_tantivy_files(file_groups, false);
// Should have 2 groups (max length of input groups)
assert_eq!(result.len(), 2);
@@ -1336,7 +1338,7 @@ mod tests {
],
vec![create_file_key(31, 40)],
];
let result = regroup_tantivy_files(file_groups);
let result = regroup_tantivy_files(file_groups, false);
// Should have 3 groups (max length of input groups)
assert_eq!(result.len(), 3);
@@ -1358,7 +1360,7 @@ mod tests {
#[test]
fn test_regroup_tantivy_files_empty_groups() {
let file_groups: Vec<Vec<FileKey>> = vec![];
let result = regroup_tantivy_files(file_groups);
let result = regroup_tantivy_files(file_groups, false);
assert_eq!(result.len(), 0);
}
@@ -1369,7 +1371,7 @@ mod tests {
create_file_key(11, 20),
create_file_key(21, 30),
]];
let result = regroup_tantivy_files(file_groups);
let result = regroup_tantivy_files(file_groups, false);
// Should have 3 groups (length of the single input group)
assert_eq!(result.len(), 3);
@@ -1681,7 +1683,7 @@ mod tests {
vec![create_file_key(11, 20)],
vec![create_file_key(21, 30)],
];
let result = regroup_tantivy_files(file_groups);
let result = regroup_tantivy_files(file_groups, false);
assert_eq!(result.len(), 1); // Max group length is 1
assert_eq!(result[0].len(), 3); // Should contain all files
@@ -1689,4 +1691,28 @@ mod tests {
assert_eq!(result[0][1].key, "file_11_20");
assert_eq!(result[0][2].key, "file_21_30");
}
#[test]
fn test_regroup_tantivy_files_ascend() {
// Same input as test_regroup_tantivy_files_basic, but with ascend=true.
// Without the reverse, each group stays in ascending order, so the
// interleaved output groups contain oldest files first.
let file_groups = vec![
vec![create_file_key(1, 10), create_file_key(11, 20)],
vec![create_file_key(21, 30), create_file_key(31, 40)],
];
let result = regroup_tantivy_files(file_groups, true);
assert_eq!(result.len(), 2);
// First group should contain the first (oldest) file from each input group
assert_eq!(result[0].len(), 2);
assert_eq!(result[0][0].key, "file_1_10");
assert_eq!(result[0][1].key, "file_21_30");
// Second group should contain the last (newest) file from each input group
assert_eq!(result[1].len(), 2);
assert_eq!(result[1][0].key, "file_11_20");
assert_eq!(result[1][1].key, "file_31_40");
}
}