ISSUE-1118 Check collector logic at root and optimize (#1121)

Author: linxGnu

quickwit-search/src/collector.rs

@@ -321,7 +321,8 @@ fn merge_leaf_responses(
         .sum();
     let failed_splits = leaf_responses
         .iter()
-        .flat_map(|leaf_response| leaf_response.failed_splits.iter().cloned())
+        .flat_map(|leaf_response| leaf_response.failed_splits.iter())
+        .cloned()
         .collect_vec();
     let all_partial_hits: Vec<PartialHit> = leaf_responses
         .into_iter()
@@ -342,7 +343,7 @@ fn merge_leaf_responses(
 ///
 /// TODO we could possibly optimize the sort away (but I doubt it matters).
 fn top_k_partial_hits(mut partial_hits: Vec<PartialHit>, num_hits: usize) -> Vec<PartialHit> {
-    partial_hits.sort_by(|left, right| {
+    partial_hits.sort_unstable_by(|left, right| {
         let left_key = partial_hit_sorting_key(left);
         let right_key = partial_hit_sorting_key(right);
         left_key.cmp(&right_key)

quickwit-search/src/leaf.rs

@@ -267,7 +267,11 @@ pub async fn leaf_search(
                 Err(err) => Either::Right(err),
             });
 
+    // Creates a collector which merges responses into one
     let merge_collector = make_merge_collector(request);
+
+    // Merging is a cpu-bound task.
+    // It should be executed by Tokio's blocking threads.
     let mut merged_search_response =
         spawn_blocking(move || merge_collector.merge_fruits(split_search_responses))
             .instrument(info_span!("merge_search_responses"))

quickwit-search/src/root.rs

@@ -167,7 +167,12 @@ pub async fn root_search(
             }),
     )
     .await?;
+
+    // Creates a collector which merges responses into one
     let merge_collector = make_merge_collector(search_request);
+
+    // Merging is a cpu-bound task.
+    // It should be executed by Tokio's blocking threads.
     let leaf_search_response =
         spawn_blocking(move || merge_collector.merge_fruits(leaf_search_responses))
             .await?
@@ -222,7 +227,7 @@ pub async fn root_search(
         .into_iter()
         .flat_map(|response| response.hits)
         .collect();
-    hits.sort_by_key(|hit| {
+    hits.sort_unstable_by_key(|hit| {
         Reverse(
             hit.partial_hit
                 .as_ref()