From 627d620a2fd10b7da5f8cd3676e210fd9a4e276e Mon Sep 17 00:00:00 2001 From: Huaijin Date: Mon, 11 May 2026 16:53:05 +0800 Subject: [PATCH] fix: simpel select asc issue and distinct time range (#11715) - [x] simple select do not consider asc - [x] simple distinct do not consider time range filter --- src/service/search/grpc/flight.rs | 32 +++++++++++++++----- src/service/search/grpc/storage.rs | 48 +++++++++++++++++++++++------- 2 files changed, 61 insertions(+), 19 deletions(-) diff --git a/src/service/search/grpc/flight.rs b/src/service/search/grpc/flight.rs index 60916803ac..2f3e3de37b 100644 --- a/src/service/search/grpc/flight.rs +++ b/src/service/search/grpc/flight.rs @@ -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, file_list: Vec, index_updated_at: i64, + time_range: (i64, i64), ) -> Result<(Vec, Vec), 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, index_updated_at: i64, + time_range: Option<(i64, i64)>, ) -> (Vec, Vec) { - 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); } diff --git a/src/service/search/grpc/storage.rs b/src/service/search/grpc/storage.rs index 2a0948b722..2b4c0a463b 100644 --- a/src/service/search/grpc/storage.rs +++ b/src/service/search/grpc/storage.rs @@ -966,22 +966,22 @@ fn partition_tantivy_files( idx_optimize_mode: &Option, target_partitions: usize, ) -> (Vec>, 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> { +fn regroup_tantivy_files(file_groups: Vec>, ascend: bool) -> Vec> { 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![Vec::new(); max_group_len]; @@ -998,7 +998,9 @@ fn regroup_tantivy_files(file_groups: Vec>) -> Vec> { 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![]; - 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"); + } }