Remove deduplication of logs and traces by service name and timestamp

guilload · guilload · commit dfc661b32ba1 · 2025-07-23T14:33:47.000-04:00
diff --git a/quickwit/quickwit-ingest/src/ingest_v2/mod.rs b/quickwit/quickwit-ingest/src/ingest_v2/mod.rs
@@ -174,6 +174,14 @@ impl JsonDocBatchV2Builder {
             doc_lengths: self.doc_lengths,
         }
     }
+
+    pub fn with_num_docs(num_docs: usize) -> Self {
+        Self {
+            doc_uids: Vec::with_capacity(num_docs),
+            doc_lengths: Vec::with_capacity(num_docs),
+            ..Default::default()
+        }
+    }
 }
 
 /// Helper struct to build an [`IngestRequestV2`].
diff --git a/quickwit/quickwit-opentelemetry/src/otlp/logs.rs b/quickwit/quickwit-opentelemetry/src/otlp/logs.rs
@@ -12,8 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::cmp::{Ord, Ordering, PartialEq, PartialOrd};
-use std::collections::{BTreeSet, HashMap, btree_set};
+use std::collections::HashMap;
 
 use async_trait::async_trait;
 use prost::Message;
@@ -188,36 +187,6 @@ pub struct LogRecord {
     pub scope_dropped_attributes_count: u32,
 }
 
-/// A wrapper around `LogRecord` that implements `Ord` to allow insertion of log records into a
-/// `BTreeSet`.
-#[derive(Debug)]
-struct OrdLogRecord(LogRecord);
-
-impl Ord for OrdLogRecord {
-    fn cmp(&self, other: &Self) -> Ordering {
-        self.0
-            .service_name
-            .cmp(&other.0.service_name)
-            .then(self.0.timestamp_nanos.cmp(&other.0.timestamp_nanos))
-    }
-}
-
-impl PartialOrd for OrdLogRecord {
-    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        Some(self.cmp(other))
-    }
-}
-
-impl PartialEq for OrdLogRecord {
-    fn eq(&self, other: &Self) -> bool {
-        self.0.timestamp_nanos == other.0.timestamp_nanos
-            && self.0.service_name == other.0.service_name
-            && self.0.body == other.0.body
-    }
-}
-
-impl Eq for OrdLogRecord {}
-
 struct ParsedLogRecords {
     doc_batch: DocBatchV2,
     num_log_records: u64,
@@ -301,11 +270,11 @@ impl OtlpGrpcLogsService {
         let num_log_records = log_records.len() as u64;
         let mut error_message = String::new();
 
-        let mut doc_batch_builder = JsonDocBatchV2Builder::default();
+        let mut doc_batch_builder = JsonDocBatchV2Builder::with_num_docs(num_log_records as usize);
         let mut doc_uid_generator = DocUidGenerator::default();
         for log_record in log_records {
             let doc_uid = doc_uid_generator.next_doc_uid();
-            if let Err(error) = doc_batch_builder.add_doc(doc_uid, log_record.0) {
+            if let Err(error) = doc_batch_builder.add_doc(doc_uid, log_record) {
                 error!(error=?error, "failed to JSON serialize span");
                 error_message = format!("failed to JSON serialize span: {error:?}");
                 num_parse_errors += 1;
@@ -393,19 +362,24 @@ impl LogsService for OtlpGrpcLogsService {
     }
 }
 
-fn parse_otlp_logs(
-    request: ExportLogsServiceRequest,
-) -> Result<BTreeSet<OrdLogRecord>, OtlpLogsError> {
-    let mut log_records = BTreeSet::new();
-    for resource_log in request.resource_logs {
+fn parse_otlp_logs(request: ExportLogsServiceRequest) -> Result<Vec<LogRecord>, OtlpLogsError> {
+    let num_log_records = request
+        .resource_logs
+        .iter()
+        .flat_map(|resource_log| resource_log.scope_logs.iter())
+        .map(|scope_logs| scope_logs.log_records.len())
+        .sum();
+    let mut log_records = Vec::with_capacity(num_log_records);
+
+    for resource_logs in request.resource_logs {
         let mut resource_attributes = extract_attributes(
-            resource_log
+            resource_logs
                 .resource
                 .clone()
                 .map(|rsrc| rsrc.attributes)
                 .unwrap_or_default(),
         );
-        let resource_dropped_attributes_count = resource_log
+        let resource_dropped_attributes_count = resource_logs
             .resource
             .map(|rsrc| rsrc.dropped_attributes_count)
             .unwrap_or(0);
@@ -414,31 +388,31 @@ fn parse_otlp_logs(
             Some(JsonValue::String(value)) => value.to_string(),
             _ => "unknown_service".to_string(),
         };
-        for scope_log in resource_log.scope_logs {
-            let scope_name = scope_log
+        for scope_logs in resource_logs.scope_logs {
+            let scope_name = scope_logs
                 .scope
                 .as_ref()
                 .map(|scope| &scope.name)
                 .filter(|name| !name.is_empty());
-            let scope_version = scope_log
+            let scope_version = scope_logs
                 .scope
                 .as_ref()
                 .map(|scope| &scope.version)
                 .filter(|version| !version.is_empty());
             let scope_attributes = extract_attributes(
-                scope_log
+                scope_logs
                     .scope
                     .clone()
                     .map(|scope| scope.attributes)
                     .unwrap_or_default(),
             );
-            let scope_dropped_attributes_count = scope_log
+            let scope_dropped_attributes_count = scope_logs
                 .scope
                 .as_ref()
                 .map(|scope| scope.dropped_attributes_count)
                 .unwrap_or(0);
 
-            for log_record in scope_log.log_records {
+            for log_record in scope_logs.log_records {
                 let observed_timestamp_nanos = if log_record.observed_time_unix_nano == 0 {
                     // As per OTEL model spec, this field SHOULD be set once the
                     // event is observed by OpenTelemetry. If it's not set, we
@@ -503,7 +477,7 @@ fn parse_otlp_logs(
                     scope_attributes: scope_attributes.clone(),
                     scope_dropped_attributes_count,
                 };
-                log_records.insert(OrdLogRecord(log_record));
+                log_records.push(log_record);
             }
         }
     }
@@ -512,15 +486,15 @@ fn parse_otlp_logs(
 
 /// An iterator of JSON OTLP log records for use in the doc processor.
 pub struct JsonLogIterator {
-    logs: btree_set::IntoIter<OrdLogRecord>,
+    logs: std::vec::IntoIter<LogRecord>,
     current_log_idx: usize,
     num_logs: usize,
     avg_log_size: usize,
     avg_log_size_rem: usize,
 }
 
 impl JsonLogIterator {
-    fn new(logs: BTreeSet<OrdLogRecord>, num_bytes: usize) -> Self {
+    fn new(logs: Vec<LogRecord>, num_bytes: usize) -> Self {
         let num_logs = logs.len();
         let avg_log_size = num_bytes.checked_div(num_logs).unwrap_or(0);
         let avg_log_size_rem = avg_log_size + num_bytes.checked_rem(num_logs).unwrap_or(0);
@@ -539,9 +513,10 @@ impl Iterator for JsonLogIterator {
     type Item = (JsonValue, usize);
 
     fn next(&mut self) -> Option<Self::Item> {
-        let log_opt = self.logs.next().map(|OrdLogRecord(log)| {
-            serde_json::to_value(log).expect("`LogRecord` should be JSON serializable")
-        });
+        let log_opt = self
+            .logs
+            .next()
+            .map(|log| serde_json::to_value(log).expect("`LogRecord` should be JSON serializable"));
         if log_opt.is_some() {
             self.current_log_idx += 1;
         }
diff --git a/quickwit/quickwit-opentelemetry/src/otlp/traces.rs b/quickwit/quickwit-opentelemetry/src/otlp/traces.rs
@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::cmp::{Ord, Ordering, PartialEq, PartialOrd};
-use std::collections::{BTreeSet, HashMap, btree_set};
+use std::cmp::PartialEq;
+use std::collections::HashMap;
 use std::str::FromStr;
 
 use async_trait::async_trait;
@@ -323,43 +323,6 @@ impl Span {
     }
 }
 
-/// A wrapper around `Span` that implements `Ord` to allow insertion of spans into a `BTreeSet`.
-#[derive(Debug)]
-struct OrdSpan(Span);
-
-impl Ord for OrdSpan {
-    /// Sort spans by trace ID, span name, start timestamp, and span ID in an attempt to group the
-    /// spans by trace ID and span name in the same docstore blocks. At some point, the
-    /// cost–benefit of this approach should be evaluated via a benchmark and revisited if
-    /// necessary.
-    fn cmp(&self, other: &Self) -> Ordering {
-        self.0
-            .trace_id
-            .cmp(&other.0.trace_id)
-            .then(self.0.span_name.cmp(&other.0.span_name))
-            .then(
-                self.0
-                    .span_start_timestamp_nanos
-                    .cmp(&other.0.span_start_timestamp_nanos),
-            )
-            .then(self.0.span_id.cmp(&other.0.span_id))
-    }
-}
-
-impl PartialOrd for OrdSpan {
-    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
-        Some(self.cmp(other))
-    }
-}
-
-impl PartialEq for OrdSpan {
-    fn eq(&self, other: &Self) -> bool {
-        self.cmp(other) == Ordering::Equal
-    }
-}
-
-impl Eq for OrdSpan {}
-
 #[derive(Debug, Clone)]
 pub struct SpanKind(i32);
 
@@ -650,10 +613,14 @@ impl Link {
     }
 }
 
-fn parse_otlp_spans(
-    request: ExportTraceServiceRequest,
-) -> Result<BTreeSet<OrdSpan>, OtlpTracesError> {
-    let mut spans = BTreeSet::new();
+fn parse_otlp_spans(request: ExportTraceServiceRequest) -> Result<Vec<Span>, OtlpTracesError> {
+    let num_spans = request
+        .resource_spans
+        .iter()
+        .flat_map(|resource_spans| resource_spans.scope_spans.iter())
+        .map(|scope_spans| scope_spans.spans.len())
+        .sum();
+    let mut spans = Vec::with_capacity(num_spans);
 
     for resource_spans in request.resource_spans {
         let resource = resource_spans
@@ -664,7 +631,7 @@ fn parse_otlp_spans(
             let scope = scope_spans.scope.map(Scope::from_otlp).unwrap_or_default();
             for span in scope_spans.spans {
                 let span = Span::from_otlp(span, &resource, &scope)?;
-                spans.insert(OrdSpan(span));
+                spans.push(span);
             }
         }
     }
@@ -765,11 +732,11 @@ impl OtlpGrpcTracesService {
         let mut num_parse_errors = 0;
         let mut error_message = String::new();
 
-        let mut doc_batch_builder = JsonDocBatchV2Builder::default();
+        let mut doc_batch_builder = JsonDocBatchV2Builder::with_num_docs(num_spans as usize);
         let mut doc_uid_generator = DocUidGenerator::default();
         for span in spans {
             let doc_uid = doc_uid_generator.next_doc_uid();
-            if let Err(error) = doc_batch_builder.add_doc(doc_uid, span.0) {
+            if let Err(error) = doc_batch_builder.add_doc(doc_uid, span) {
                 error!(error=?error, "failed to JSON serialize span");
                 error_message = format!("failed to JSON serialize span: {error:?}");
                 num_parse_errors += 1;
@@ -859,15 +826,15 @@ impl TraceService for OtlpGrpcTracesService {
 
 /// An iterator of JSON OTLP spans for use in the doc processor.
 pub struct JsonSpanIterator {
-    spans: btree_set::IntoIter<OrdSpan>,
+    spans: std::vec::IntoIter<Span>,
     current_span_idx: usize,
     num_spans: usize,
     avg_span_size: usize,
     avg_span_size_rem: usize,
 }
 
 impl JsonSpanIterator {
-    fn new(spans: BTreeSet<OrdSpan>, num_bytes: usize) -> Self {
+    fn new(spans: Vec<Span>, num_bytes: usize) -> Self {
         let num_spans = spans.len();
         let avg_span_size = num_bytes.checked_div(num_spans).unwrap_or(0);
         let avg_span_size_rem = avg_span_size + num_bytes.checked_rem(num_spans).unwrap_or(0);
@@ -886,9 +853,10 @@ impl Iterator for JsonSpanIterator {
     type Item = (JsonValue, usize);
 
     fn next(&mut self) -> Option<Self::Item> {
-        let span_opt = self.spans.next().map(|OrdSpan(span)| {
-            serde_json::to_value(span).expect("`Span` should be JSON serializable")
-        });
+        let span_opt = self
+            .spans
+            .next()
+            .map(|span| serde_json::to_value(span).expect("`Span` should be JSON serializable"));
         if span_opt.is_some() {
             self.current_span_idx += 1;
         }
@@ -1348,7 +1316,7 @@ mod tests {
 
     #[test]
     fn test_json_span_iterator() {
-        let mut json_span_iterator = JsonSpanIterator::new(BTreeSet::new(), 0);
+        let mut json_span_iterator = JsonSpanIterator::new(Vec::new(), 0);
         assert!(json_span_iterator.next().is_none());
 
         let span_0 = Span {
@@ -1384,7 +1352,7 @@ mod tests {
             links: Vec::new(),
         };
 
-        let spans = BTreeSet::from_iter([OrdSpan(span_0.clone())]);
+        let spans = vec![span_0.clone()];
         let mut json_span_iterator = JsonSpanIterator::new(spans, 3);
 
         assert_eq!(
@@ -1396,7 +1364,7 @@ mod tests {
         let mut span_1 = span_0.clone();
         span_1.span_id = SpanId::new([3; 8]);
 
-        let spans = BTreeSet::from_iter([OrdSpan(span_0.clone()), OrdSpan(span_1.clone())]);
+        let spans = vec![span_0.clone(), span_1.clone()];
         let mut json_span_iterator = JsonSpanIterator::new(spans, 7);
 
         assert_eq!(

Original file line number	Diff line number	Diff line change
`@@ -174,6 +174,14 @@ impl JsonDocBatchV2Builder {`
`174`	`174`	`doc_lengths: self.doc_lengths,`
`175`	`175`	`}`
`176`	`176`	`}`
	`177`	`+`
	`178`	`+ pub fn with_num_docs(num_docs: usize) -> Self {`
	`179`	`+ Self {`
	`180`	`+ doc_uids: Vec::with_capacity(num_docs),`
	`181`	`+ doc_lengths: Vec::with_capacity(num_docs),`
	`182`	`+ ..Default::default()`
	`183`	`+ }`
	`184`	`+ }`
`177`	`185`	`}`
`178`	`186`
`179`	`187`	/// Helper struct to build an [`IngestRequestV2`].