non-ambiguous internal aggregations (#5715)

* add proxy struct for aggregations

* use postcard for aggregations internally

* use vec instead of hashmaps

Author: trinity-1686a

quickwit/Cargo.lock

@@ -207,6 +207,7 @@ reqwest = { version = "0.12", default-features = false, features = [
 reqwest-middleware = "0.4"
 reqwest-retry = "0.7"
 rust-embed = "6.8.1"
+rustc-hash = "2.1.1"
 rustls = "0.21"
 rustls-pemfile = "1.0.0"
 sea-query = { version = "0.30" }

quickwit/Cargo.toml

@@ -14,6 +14,7 @@ license.workspace = true
 async-trait = { workspace = true }
 itertools = { workspace = true }
 once_cell = { workspace = true }
+postcard = { workspace = true }
 prost = { workspace = true }
 prost-types = { workspace = true }
 serde = { workspace = true }

quickwit/quickwit-jaeger/Cargo.toml

@@ -282,11 +282,11 @@ impl JaegerService {
         };
         let search_response = self.search_service.root_search(search_request).await?;
 
-        let Some(agg_result_json) = search_response.aggregation else {
+        let Some(agg_result_postcard) = search_response.aggregation_postcard else {
             debug!("the query matched no traces");
             return Ok((Vec::new(), 0..=0));
         };
-        let trace_ids = collect_trace_ids(&agg_result_json)?;
+        let trace_ids = collect_trace_ids(&agg_result_postcard)?;
         debug!("the query matched {} traces.", trace_ids.0.len());
         Ok(trace_ids)
     }
@@ -1087,9 +1087,11 @@ fn qw_event_to_jaeger_log(event: QwEvent) -> Result<JaegerLog, Status> {
     Ok(log)
 }
 
-fn collect_trace_ids(trace_ids_json: &str) -> Result<(Vec<TraceId>, TimeIntervalSecs), Status> {
+fn collect_trace_ids(
+    trace_ids_postcard: &[u8],
+) -> Result<(Vec<TraceId>, TimeIntervalSecs), Status> {
     let collector_fruit: <FindTraceIdsCollector as Collector>::Fruit =
-        json_deserialize(trace_ids_json, "trace IDs aggregation")?;
+        postcard_deserialize(trace_ids_postcard, "trace IDs aggregation")?;
     if collector_fruit.is_empty() {
         return Ok((Vec::new(), 0..=0));
     }
@@ -1118,6 +1120,19 @@ where T: Deserialize<'a> {
     }
 }
 
+fn postcard_deserialize<'a, T>(json: &'a [u8], label: &'static str) -> Result<T, Status>
+where T: Deserialize<'a> {
+    match postcard::from_bytes(json) {
+        Ok(deserialized) => Ok(deserialized),
+        Err(error) => {
+            error!("failed to deserialize {label}: {error:?}");
+            Err(Status::internal(format!(
+                "Failed to deserialize {label}: {error:?}."
+            )))
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use quickwit_opentelemetry::otlp::{OTEL_TRACES_INDEX_ID_PATTERN, OtelSignal};
@@ -2496,34 +2511,50 @@ mod tests {
 
     #[test]
     fn test_collect_trace_ids() {
+        use quickwit_opentelemetry::otlp::TraceId;
+        use quickwit_search::Span;
+        use tantivy::DateTime;
         {
-            let agg_result_json = r#"[]"#;
-            let (trace_ids, _span_timestamps_range) = collect_trace_ids(agg_result_json).unwrap();
+            let agg_result: Vec<Span> = Vec::new();
+            let agg_result_postcard = postcard::to_stdvec(&agg_result).unwrap();
+            let (trace_ids, _span_timestamps_range) =
+                collect_trace_ids(&agg_result_postcard).unwrap();
             assert!(trace_ids.is_empty());
         }
         {
-            let agg_result_json = r#"[
-                {
-                    "trace_id": "01010101010101010101010101010101",
-                    "span_timestamp": 1684857492783747000
-                }
-            ]"#;
-            let (trace_ids, span_timestamps_range) = collect_trace_ids(agg_result_json).unwrap();
+            let agg_result = vec![Span {
+                trace_id: TraceId::new([
+                    0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01,
+                    0x01, 0x01, 0x01,
+                ]),
+                span_timestamp: DateTime::from_timestamp_nanos(1684857492783747000),
+            }];
+            let agg_result_postcard = postcard::to_stdvec(&agg_result).unwrap();
+            let (trace_ids, span_timestamps_range) =
+                collect_trace_ids(&agg_result_postcard).unwrap();
             assert_eq!(trace_ids.len(), 1);
             assert_eq!(span_timestamps_range, 1684857492..=1684857492);
         }
         {
-            let agg_result_json = r#"[
-                {
-                    "trace_id": "0102030405060708090a0b0c0d0e0f10",
-                    "span_timestamp": 1684857492783747000
+            let agg_result = vec![
+                Span {
+                    trace_id: TraceId::new([
+                        0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c,
+                        0x0d, 0x0e, 0x0f, 0x10,
+                    ]),
+                    span_timestamp: DateTime::from_timestamp_nanos(1684857492783747000),
                 },
-                {
-                    "trace_id": "02020202020202020202020202020202",
-                    "span_timestamp": 1684857826019627000
-                }
-            ]"#;
-            let (trace_ids, span_timestamps_range) = collect_trace_ids(agg_result_json).unwrap();
+                Span {
+                    trace_id: TraceId::new([
+                        0x02, 0x02, 0x02, 0x02, 0x02, 0x02, 0x02, 0x02, 0x02, 0x02, 0x02, 0x02,
+                        0x02, 0x02, 0x02, 0x02,
+                    ]),
+                    span_timestamp: DateTime::from_timestamp_nanos(1684857826019627000),
+                },
+            ];
+            let agg_result_postcard = postcard::to_stdvec(&agg_result).unwrap();
+            let (trace_ids, span_timestamps_range) =
+                collect_trace_ids(&agg_result_postcard).unwrap();
             assert_eq!(trace_ids.len(), 2);
             assert_eq!(span_timestamps_range, 1684857492..=1684857826);
         }

quickwit/quickwit-jaeger/src/lib.rs

@@ -289,8 +289,11 @@ message SearchResponse {
   // The searcherrors that occurred formatted as string.
   repeated string errors = 4;
 
-  // Serialized aggregation response
-  optional string aggregation = 5;
+  // used to be json-encoded aggregation
+  reserved 5;
+
+  // Postcard-encoded aggregation response
+  optional bytes aggregation_postcard = 9;
 
   // Scroll Id (only set if scroll_secs was set in the request)
   optional string scroll_id = 6;

quickwit/quickwit-proto/protos/quickwit/search.proto

@@ -26,6 +26,7 @@ tantivy = { workspace = true }
 tantivy-fst = { workspace = true }
 time = { workspace = true }
 thiserror = { workspace = true }
+rustc-hash = { workspace = true }
 whichlang = { workspace = true, optional = true }
 
 quickwit-common = { workspace = true }