[query-engine] OTLP RecordSet bridge benchmarks (open-telemetry#2484)

# Changes

* Adds some benchmarks to OTLP RecordSet bridge

# Details

I was measuring `toint(unknown)` vs `coalesce(toint(unknown), null)` for
@drewrelmas and @utpilla just checking in the work so we can build up
more benchmarks in the future (if needed). Modeled after what
@albertlockett is doing in
[query-engine](https://github.com/open-telemetry/otel-arrow/blob/main/rust/otap-dataflow/crates/query-engine/benches/filter.rs).

---------

Co-authored-by: Drew Relmas <drewrelmas@gmail.com>

Author: CodeBlanch

rust/experimental/query_engine/Cargo.toml

@@ -22,6 +22,7 @@ rust-version = "1.86.0"
 bytes = "1.10.1"
 caseless = "0.2.2"
 chrono = "0.4.41"
+criterion = "0.8.0"
 hex = "0.4.3"
 opentelemetry-proto = "0.31.0"
 pest = "2.8"

rust/experimental/query_engine/engine-recordset-otlp-bridge/Cargo.toml

@@ -20,5 +20,12 @@ data_engine_kql_parser = { path = "../kql-parser" }
 data_engine_recordset = { path = "../engine-recordset" }
 
 [dev-dependencies]
+criterion = { workspace = true }
 opentelemetry-proto = { workspace = true }
-prost = { workspace = true }
+prost = { workspace = true }
+
+otap-df-pdata = { path = "../../../otap-dataflow/crates/pdata", features = ["testing"] }
+
+[[bench]]
+name = "extend"
+harness = false

rust/experimental/query_engine/engine-recordset-otlp-bridge/benches/extend.rs

@@ -0,0 +1,70 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use data_engine_recordset_otlp_bridge::*;
+use otap_df_pdata::testing::fixtures::logs_with_varying_attributes_and_properties;
+use prost::Message;
+
+fn generate_logs_batch(batch_size: usize) -> Vec<u8> {
+    let logs_data = logs_with_varying_attributes_and_properties(batch_size);
+    logs_data.encode_to_vec()
+}
+
+fn bench_log_pipeline(
+    c: &mut Criterion,
+    batch_sizes: &[usize],
+    bench_group_name: &str,
+    bench_pipeline_kql: &str,
+) {
+    let mut group = c.benchmark_group(bench_group_name);
+    for batch_size in batch_sizes {
+        let benchmark_id = BenchmarkId::new("batch_size", batch_size);
+        let _ = group.bench_with_input(benchmark_id, batch_size, |b, batch_size| {
+            let batch = generate_logs_batch(*batch_size);
+            let pipeline = parse_kql_query_into_pipeline(bench_pipeline_kql, None)
+                .expect("can parse pipeline");
+            b.iter(|| {
+                process_protobuf_otlp_export_logs_service_request_using_pipeline(
+                    &pipeline,
+                    RecordSetEngineDiagnosticLevel::Warn,
+                    &batch,
+                )
+                .expect("doesn't fail")
+            });
+        });
+    }
+    group.finish();
+}
+
+fn bench_extend_pipelines(c: &mut Criterion) {
+    let batch_sizes = [32, 1024, 8192];
+
+    // Note: toint() for an unknown field generates a warning
+    bench_log_pipeline(
+        c,
+        &batch_sizes,
+        "toint_unknown_field",
+        "source | extend a = toint(unknown)",
+    );
+
+    // Note: coalesce with inner toint() for an unknown field generates an info
+    bench_log_pipeline(
+        c,
+        &batch_sizes,
+        "coalesce_toint_unknown_field",
+        "source | extend a = coalesce(toint(unknown), null)",
+    );
+}
+
+mod benches {
+    use super::*;
+
+    criterion_group!(
+        name = benches;
+        config = Criterion::default();
+        targets = bench_extend_pipelines
+    );
+}
+
+criterion_main!(benches::benches);