Use optimized append_..._n methods in syslog receiver when constructing OTAP record (open-telemetry#1196)

Closes open-telemetry#879 

Use the "append_..._n" methods in the syslog receiver when constructing
the arrow records. These methods have some nice performance
optimizations:
- when appending a non-null value if the underlying builder is a
dictionary array, the builder is able to append the value multiple times
without doing multiple key lookups. For example:

https://github.com/apache/arrow-rs/blob/0737c61e76057b127312dd8058887649ece702b8/arrow-array/src/builder/generic_bytes_dictionary_builder.rs#L304-L316

- when the value is null, we can use the optimized code path to append
multiple nulls to the null buffer:

https://github.com/apache/arrow-rs/blob/0737c61e76057b127312dd8058887649ece702b8/arrow-buffer/src/builder/boolean.rs#L193

Benchmark results:
```
arrow_batch_creation/rfc3164_arrow_batch_100_msgs
                        time:   [82.608 µs 82.774 µs 82.950 µs]
                        change: [−3.9462% −3.4305% −2.9199%] (p = 0.00 < 0.05)
                        Performance has improved.
                        
arrow_batch_creation/rfc5424_arrow_batch_100_msgs
                        time:   [46.139 µs 46.238 µs 46.338 µs]
                        change: [−4.2893% −3.9924% −3.7179%] (p = 0.00 < 0.05)
                        Performance has improved.
                        
arrow_batch_creation/cef_arrow_batch_100_msgs
                        time:   [40.973 µs 41.141 µs 41.308 µs]
                        change: [−3.8351% −3.3433% −2.8615%] (p = 0.00 < 0.05)
                        Performance has improved.
```

Author: albertlockett

rust/otap-dataflow/crates/otap/src/encoder.rs

@@ -304,7 +304,8 @@ where
                             .as_ref()
                             .expect("LogRecord should not be None")
                             .time_unix_nano()
-                            .map(|v| v as i64),
+                            .map(|v| v as i64)
+                            .unwrap_or(0),
                     );
                 }
                 for log_record in log_records_slice {
@@ -313,7 +314,8 @@ where
                             .as_ref()
                             .expect("LogRecord should not be None")
                             .observed_time_unix_nano()
-                            .map(|v| v as i64),
+                            .map(|v| v as i64)
+                            .unwrap_or(0),
                     );
                 }
                 logs.append_schema_url_n(scope_schema_url, logs_count);

rust/otap-dataflow/crates/otap/src/syslog_cef_receiver/arrow_records_encoder.rs

@@ -47,7 +47,7 @@ impl ArrowRecordsBuilder {
     /// Appends a parsed syslog message to the builder.
     pub fn append_syslog(&mut self, syslog_message: ParsedSyslogMessage<'_>) {
         self.logs
-            .append_time_unix_nano(syslog_message.timestamp().map(|v| v as i64));
+            .append_time_unix_nano(syslog_message.timestamp().map(|v| v as i64).unwrap_or(0));
 
         let (severity_number, severity_text) =
             syslog_message.severity().unwrap_or((0, "UNSPECIFIED"));
@@ -89,32 +89,15 @@ impl ArrowRecordsBuilder {
             .append_dropped_attributes_count_n(0, log_record_count);
 
         let observed_time = Utc::now().timestamp_nanos_opt().unwrap_or(0);
-        for _ in 0..log_record_count {
-            self.logs
-                .append_observed_time_unix_nano(Some(observed_time));
-        }
-
+        self.logs
+            .append_observed_time_unix_nano_n(observed_time, log_record_count);
         self.logs.append_schema_url_n(None, log_record_count);
+        self.logs
+            .append_dropped_attributes_count_n(0, log_record_count);
 
-        for _ in 0..log_record_count {
-            self.logs.append_dropped_attributes_count(0);
-        }
-
-        for _ in 0..log_record_count {
-            self.logs.append_flags(None);
-        }
-
-        for _ in 0..log_record_count {
-            _ = self.logs.append_trace_id(None);
-        }
-
-        for _logs in 0..log_record_count {
-            _ = self.logs.append_trace_id(None);
-        }
-
-        for _logs in 0..log_record_count {
-            _ = self.logs.append_span_id(None);
-        }
+        self.logs.append_flags_n(None, log_record_count);
+        _ = self.logs.append_trace_id_n(None, log_record_count);
+        _ = self.logs.append_span_id_n(None, log_record_count);
 
         let mut otap_batch = OtapArrowRecords::Logs(Logs::default());
 

rust/otel-arrow-rust/src/encode/record/array.rs

@@ -144,6 +144,9 @@ pub trait CheckedArrayAppendSlice {
     /// append a slice of T to the builder. Note that this does not append an individual
     /// element for each value in the slice, it appends the slice as a single row
     fn append_slice(&mut self, val: &[Self::Native]) -> Result<(), ArrowError>;
+
+    /// append the slice to the builder `n` times
+    fn append_slice_n(&mut self, val: &[Self::Native], n: usize) -> Result<(), ArrowError>;
 }
 
 /// Used by the builder to identify the default value of the array that is being built. By default
@@ -614,6 +617,15 @@ where
             retry = { self.append_slice(value) }
         )
     }
+
+    fn append_slice_n(&mut self, value: &[Self::Native], n: usize) -> Result<(), ArrowError> {
+        handle_append_checked!(
+            self,
+            append_slice_n(value, n),
+            default_check = Self::is_default_value(&self.default_value, &value),
+            retry = { self.append_slice_n(value, n) }
+        )
+    }
 }
 
 impl<T, TArgs, TN, TD8, TD16> AdaptiveArrayBuilder<T, TArgs, TN, TD8, TD16>
@@ -1278,6 +1290,7 @@ pub mod test {
         assert!(builder.append_slice(&valid_values[0]).is_ok());
         assert!(builder.append_slice(&valid_values[1]).is_ok());
         builder.append_nulls(2);
+        assert!(builder.append_slice_n(&valid_values[1], 2).is_ok());
 
         let result = builder.finish().unwrap();
         assert_eq!(
@@ -1287,7 +1300,7 @@ pub mod test {
                 Box::new(DataType::FixedSizeBinary(1))
             )
         );
-        assert_eq!(result.len(), 7);
+        assert_eq!(result.len(), 9);
 
         let dict_array = result
             .as_any()
@@ -1296,7 +1309,17 @@ pub mod test {
         let dict_keys = dict_array.keys();
         assert_eq!(
             dict_keys,
-            &UInt8Array::from_iter(vec![Some(0), Some(1), None, Some(0), Some(1), None, None])
+            &UInt8Array::from_iter(vec![
+                Some(0),
+                Some(1),
+                None,
+                Some(0),
+                Some(1),
+                None,
+                None,
+                Some(1),
+                Some(1)
+            ])
         );
         let dict_values = dict_array
             .values()

rust/otel-arrow-rust/src/encode/record/array/dictionary.rs

@@ -96,6 +96,9 @@ pub trait CheckedDictionaryAppendSlice {
     /// append a slice of T to the builder. Note that this does not append an individual
     /// element for each value in the slice, it appends the slice as a single row
     fn append_slice(&mut self, val: &[Self::Native]) -> checked::Result<usize>;
+
+    /// append the slice to the builder `n` times.
+    fn append_slice_n(&mut self, val: &[Self::Native], n: usize) -> checked::Result<usize>;
 }
 
 // This is the error type for the result that is returned by CheckedDictionaryArrayAppend trait.
@@ -593,6 +596,34 @@ where
             }
         }
     }
+
+    pub fn append_slice_n_checked(&mut self, value: &[T], n: usize) -> checked::Result<usize> {
+        loop {
+            let append_result = match &mut self.variant {
+                DictIndexVariant::UInt8(dict_builder) => dict_builder.append_slice_n(value, n),
+                DictIndexVariant::UInt16(dict_builder) => dict_builder.append_slice_n(value, n),
+            };
+
+            match append_result {
+                Ok(index) => {
+                    if index + 1 > self.max_cardinality as usize {
+                        self.overflow_index = Some(index);
+                        return Err(checked::DictionaryBuilderError::DictOverflow {});
+                    }
+                    return Ok(index);
+                }
+                Err(checked::DictionaryBuilderError::DictOverflow {}) => {
+                    self.upgrade_key().map_err(|err| match err {
+                        DictionaryBuilderError::DictOverflow {} => {
+                            checked::DictionaryBuilderError::DictOverflow {}
+                        }
+                    })?;
+                    // continue the loop and retry
+                }
+                other => return other,
+            }
+        }
+    }
 }
 
 impl<T8, T16> AdaptiveDictionaryBuilder<T8, T16>

rust/otel-arrow-rust/src/encode/record/array/fixed_size_binary.rs

@@ -50,6 +50,14 @@ impl CheckedArrayAppendSlice for FixedSizeBinaryBuilder {
     fn append_slice(&mut self, val: &[Self::Native]) -> Result<(), ArrowError> {
         self.append_value(val)
     }
+
+    fn append_slice_n(&mut self, val: &[Self::Native], n: usize) -> Result<(), ArrowError> {
+        for _ in 0..n {
+            self.append_value(val)?;
+        }
+
+        Ok(())
+    }
 }
 
 impl ArrayAppendNulls for FixedSizeBinaryBuilder {
@@ -114,6 +122,21 @@ where
             ),
         }
     }
+
+    fn append_slice_n(
+        &mut self,
+        val: &[Self::Native],
+        n: usize,
+    ) -> super::dictionary::checked::Result<usize> {
+        // TODO use optimized method once we upgrade to latest arrow:
+        // https://github.com/apache/arrow-rs/pull/8498
+        let mut index = 0;
+        for _ in 0..n {
+            index = self.append_slice(val)?;
+        }
+
+        Ok(index)
+    }
 }
 
 impl<K> ArrayAppendNulls for FixedSizeBinaryDictionaryBuilder<K>
@@ -172,11 +195,22 @@ mod test {
         CheckedArrayAppend::append_value(&mut fsb_builder, &b"1234".to_vec()).unwrap();
         CheckedArrayAppend::append_value(&mut fsb_builder, &b"5678".to_vec()).unwrap();
         CheckedArrayAppend::append_value(&mut fsb_builder, &b"9012".to_vec()).unwrap();
+        CheckedArrayAppendSlice::append_slice(&mut fsb_builder, b"4180").unwrap();
+        CheckedArrayAppendSlice::append_slice_n(&mut fsb_builder, b"5140", 2).unwrap();
+
         let result = ArrayBuilder::finish(&mut fsb_builder);
         assert_eq!(result.data_type(), &DataType::FixedSizeBinary(4));
 
         let expected = FixedSizeBinaryArray::try_from_iter(
-            [b"1234".to_vec(), b"5678".to_vec(), b"9012".to_vec()].iter(),
+            [
+                b"1234".to_vec(),
+                b"5678".to_vec(),
+                b"9012".to_vec(),
+                b"4180".to_vec(),
+                b"5140".to_vec(),
+                b"5140".to_vec(),
+            ]
+            .iter(),
         )
         .unwrap();
 
@@ -201,6 +235,11 @@ mod test {
         let index =
             CheckedDictionaryArrayAppend::append_value(&mut dict_builder, &b"b".to_vec()).unwrap();
         assert_eq!(index, 1);
+        let index = CheckedDictionaryAppendSlice::append_slice(&mut dict_builder, b"c").unwrap();
+        assert_eq!(index, 2);
+        let index =
+            CheckedDictionaryAppendSlice::append_slice_n(&mut dict_builder, b"d", 2).unwrap();
+        assert_eq!(index, 3);
 
         let result = DictionaryBuilder::finish(&mut dict_builder);
 
@@ -215,7 +254,10 @@ mod test {
         let mut expected_dict_values = FixedSizeBinaryBuilder::new(1);
         assert!(expected_dict_values.append_value(b"a").is_ok());
         assert!(expected_dict_values.append_value(b"b").is_ok());
-        let expected_dict_keys = UInt8Array::from_iter_values(vec![0, 0, 1]);
+        assert!(expected_dict_values.append_value(b"c").is_ok());
+        assert!(expected_dict_values.append_value(b"d").is_ok());
+        assert!(expected_dict_values.append_value(b"d").is_ok());
+        let expected_dict_keys = UInt8Array::from_iter_values(vec![0, 0, 1, 2, 3, 3]);
         let expected =
             UInt8DictionaryArray::new(expected_dict_keys, Arc::new(expected_dict_values.finish()));
 

rust/otel-arrow-rust/src/encode/record/logs.rs

@@ -131,17 +131,23 @@ impl LogsRecordBatchBuilder {
     }
 
     /// append a value to the `time_unix_nano` array
-    pub fn append_time_unix_nano(&mut self, val: Option<i64>) {
-        let val = val.unwrap_or(0);
+    #[inline]
+    pub fn append_time_unix_nano(&mut self, val: i64) {
         self.time_unix_nano.append_value(&val);
     }
 
     /// append a value to the `observed_time_unix_nano` array
-    pub fn append_observed_time_unix_nano(&mut self, val: Option<i64>) {
-        let val = val.unwrap_or(0);
+    #[inline]
+    pub fn append_observed_time_unix_nano(&mut self, val: i64) {
         self.observed_time_unix_nano.append_value(&val);
     }
 
+    /// append a value to the `observed_time_unix_nano` array `n` times
+    #[inline]
+    pub fn append_observed_time_unix_nano_n(&mut self, val: i64, n: usize) {
+        self.observed_time_unix_nano.append_value_n(&val, n);
+    }
+
     /// append a value to the `severity_number` array
     pub fn append_severity_number(&mut self, val: Option<i32>) {
         if let Some(val) = val {
@@ -183,6 +189,11 @@ impl LogsRecordBatchBuilder {
         self.dropped_attributes_count.append_value(&val);
     }
 
+    /// append a value to the `dropped_attributes_count` array n times
+    pub fn append_dropped_attributes_count_n(&mut self, val: u32, n: usize) {
+        self.dropped_attributes_count.append_value_n(&val, n);
+    }
+
     /// append a value to the `flags` array
     pub fn append_flags(&mut self, val: Option<u32>) {
         if let Some(val) = val {
@@ -192,6 +203,15 @@ impl LogsRecordBatchBuilder {
         }
     }
 
+    /// append a value to the `flags` array n times
+    pub fn append_flags_n(&mut self, val: Option<u32>, n: usize) {
+        if let Some(val) = val {
+            self.flags.append_value_n(&val, n);
+        } else {
+            self.flags.append_nulls(n);
+        }
+    }
+
     /// append a value to the `trace_id` array
     pub fn append_trace_id(&mut self, val: Option<&TraceId>) -> Result<(), ArrowError> {
         if let Some(val) = val {
@@ -202,6 +222,16 @@ impl LogsRecordBatchBuilder {
         }
     }
 
+    /// append a value to the `trace_id` array `n` times
+    pub fn append_trace_id_n(&mut self, val: Option<&TraceId>, n: usize) -> Result<(), ArrowError> {
+        if let Some(val) = val {
+            self.trace_id.append_slice_n(val, n)
+        } else {
+            self.trace_id.append_nulls(n);
+            Ok(())
+        }
+    }
+
     /// append a value to the `span_id` array
     pub fn append_span_id(&mut self, val: Option<&SpanId>) -> Result<(), ArrowError> {
         if let Some(val) = val {
@@ -212,6 +242,16 @@ impl LogsRecordBatchBuilder {
         }
     }
 
+    /// append a value to the `span_id` array `n` times
+    pub fn append_span_id_n(&mut self, val: Option<&SpanId>, n: usize) -> Result<(), ArrowError> {
+        if let Some(val) = val {
+            self.span_id.append_slice_n(val, n)
+        } else {
+            self.span_id.append_nulls(n);
+            Ok(())
+        }
+    }
+
     /// append a value to the `event_name` array
     pub fn append_event_name(&mut self, val: Option<&[u8]>) {
         if let Some(val) = val {