Add filtering benchmarks (#1499)

Adds benchmarks for the filtering code used in the filter processor and
for the columnar query engine, with similar filtering criteria.

My primary goal was to confirm that we get similar performance using the
both approaches. The filter processor code uses arrow's compute kernels,
whereas the columnar query engine uses a hybrid approach of datafusion's
physical expression and compute kernels.

**filter pattern** | **batch_size** | **pdata filter code (µs)** |
**col. query engine (µs)**
:--- | :---: | :---: | :---:
`severity_text == "WARN"` | 32 | 8.5488 | 7.6645
"  | 1024 | 37.295 | 22.341
" | 8192 | 233.7 | 153.21
`attrs["code.namespace"] == "main"` | 32 | 9.9075 | 10.202
" | 1024 | 52.496 | 43.148
" | 8192 | 408.4 | 331.72
`attrs["code.namespace"] == "main" or attrs["code.line.number"] == 2` |
32 | 11.666 | 13.387
" | 1024 | 70.035 | 63.779
" | 8192 | 572.73 | 510.23
`severity_text == "WARN" and attrs["code.namespace"] == "main"` | 32 |
11.705 | 10.719
" | 1024 | 48.919 | 35.641
" | 8192 | 377.17 | 274.38

The results leads me to believe the approach the columnar query engine
is taking for filtering is on the right track.

My secondary goal here was simply to get some benchmarks in place for
the columnar query engine's filtering code, so we have a baseline for
which to measure the effect of future performance optimizations and to
ensure we're not introducing regressions as we add more sophistication
to support additional filtering use cases.

Author: albertlockett

rust/experimental/query_engine/Cargo.toml

@@ -26,6 +26,7 @@ datafusion = { version = "51.0.0", default-features = false }
 bytes = "1.10.1"
 caseless = "0.2.2"
 chrono = "0.4.41"
+criterion = "0.7.0"
 futures-core = "0.3"
 hex = "0.4.3"
 opentelemetry-proto = "0.31.0"

rust/experimental/query_engine/engine-columnar/Cargo.toml

@@ -21,8 +21,20 @@ otap-df-pdata = { path = "../../../otap-dataflow/crates/pdata" }
 data_engine_expressions = { path = "../expressions" }
 data_engine_kql_parser = { path = "../kql-parser" }
 
-
 [dev-dependencies]
+criterion = { workspace = true, features = ["async_tokio"] }
 pretty_assertions = { workspace = true }
 prost = { workspace = true }
-tokio = { workspace = true }
+tokio = { workspace = true }
+
+[[bench]]
+name = "filter"
+harness = false
+
+[profile.bench]
+opt-level = 3
+debug = false
+incremental = false
+lto = "fat"
+codegen-units = 1
+panic = "abort"

rust/experimental/query_engine/engine-columnar/benches/filter.rs

@@ -0,0 +1,100 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+use std::time::Instant;
+
+use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
+use data_engine_columnar::pipeline::Pipeline;
+use data_engine_kql_parser::{KqlParser, Parser};
+use otap_df_pdata::OtapArrowRecords;
+use otap_df_pdata::proto::OtlpProtoMessage;
+use otap_df_pdata::testing::fixtures::logs_with_varying_attributes_and_properties;
+use otap_df_pdata::testing::round_trip::otlp_to_otap;
+use tokio::runtime::Runtime;
+
+fn generate_logs_batch(batch_size: usize) -> OtapArrowRecords {
+    let logs_data = logs_with_varying_attributes_and_properties(batch_size);
+    otlp_to_otap(&OtlpProtoMessage::Logs(logs_data))
+}
+
+fn bench_log_pipeline(
+    c: &mut Criterion,
+    rt: &Runtime,
+    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| {
+            b.iter_custom(|iters| {
+                let batch = generate_logs_batch(**batch_size);
+                let query = KqlParser::parse(bench_pipeline_kql).expect("can parse pipeline");
+                let mut pipeline = Pipeline::new(query);
+                rt.block_on(async move {
+                    // execute the query once to initiate planning
+                    pipeline.execute(batch.clone()).await.unwrap();
+
+                    let start = Instant::now();
+                    for _ in 0..iters {
+                        let result = pipeline.execute(batch.clone()).await.unwrap();
+                        std::hint::black_box(result);
+                    }
+                    start.elapsed()
+                })
+            });
+        });
+    }
+    group.finish();
+}
+
+fn bench_filter_pipelines(c: &mut Criterion) {
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("can build tokio single threaded runtime");
+
+    let batch_sizes = [32, 1024, 8192];
+    bench_log_pipeline(
+        c,
+        &rt,
+        &batch_sizes,
+        "simple_field_filter",
+        "logs | where severity_text == \"WARN\"",
+    );
+    bench_log_pipeline(
+        c,
+        &rt,
+        &batch_sizes,
+        "simple_attr_filter",
+        "logs | where attributes[\"code.namespace\"] == \"main\"",
+    );
+    bench_log_pipeline(
+        c,
+        &rt,
+        &batch_sizes,
+        "attr_or_filter",
+        "logs | where attributes[\"code.namespace\"] == \"main\" or attributes[\"code.line.number\"] == 2",
+    );
+    bench_log_pipeline(
+        c,
+        &rt,
+        &batch_sizes,
+        "attr_and_prop_filter",
+        "logs | where attributes[\"code.namespace\"] == \"main\" and severity_text == \"WARN\"",
+    );
+}
+
+#[allow(missing_docs)]
+mod benches {
+    use super::*;
+
+    criterion_group!(
+        name = benches;
+        config = Criterion::default();
+        targets = bench_filter_pipelines
+    );
+}
+
+criterion_main!(benches::benches);

rust/otap-dataflow/benchmarks/benches/pdata_filter/main.rs

@@ -7,15 +7,115 @@ use std::hint::black_box;
 use std::sync::Arc;
 
 use arrow::array::Array;
-use criterion::{Criterion, criterion_group, criterion_main};
-use otap_df_pdata::otap::filter::build_uint16_id_filter;
+use criterion::{BatchSize, BenchmarkId, Criterion, criterion_group, criterion_main};
+use otap_df_pdata::OtapArrowRecords;
+use otap_df_pdata::otap::filter::logs::{LogFilter, LogMatchProperties};
+use otap_df_pdata::otap::filter::{self, MatchType, build_uint16_id_filter};
 use otap_df_pdata::proto::OtlpProtoMessage;
 use otap_df_pdata::proto::opentelemetry::arrow::v1::ArrowPayloadType;
 use otap_df_pdata::proto::opentelemetry::common::v1::{AnyValue, KeyValue};
 use otap_df_pdata::proto::opentelemetry::logs::v1::{LogRecord, LogsData, ResourceLogs, ScopeLogs};
+use otap_df_pdata::testing::fixtures::logs_with_varying_attributes_and_properties;
 use otap_df_pdata::testing::round_trip::otlp_to_otap;
 use roaring::RoaringBitmap;
 
+fn generate_logs_batch(batch_size: usize) -> OtapArrowRecords {
+    let logs_data = logs_with_varying_attributes_and_properties(batch_size);
+    otlp_to_otap(&OtlpProtoMessage::Logs(logs_data))
+}
+
+fn bench_log_filter(
+    c: &mut Criterion,
+    batch_sizes: &[usize],
+    bench_group_name: &str,
+    include: Option<LogMatchProperties>,
+    exclude: Option<LogMatchProperties>,
+) {
+    let mut group = c.benchmark_group(format!("log_filter/{bench_group_name}"));
+    for batch_size in batch_sizes {
+        let benchmark_id = BenchmarkId::new("batch_size", batch_size);
+
+        let filter = LogFilter::new(include.clone(), exclude.clone(), Vec::new());
+        let batch = generate_logs_batch(*batch_size);
+        _ = group.bench_with_input(benchmark_id, &(batch, filter), |b, input| {
+            b.iter_batched(
+                || input,
+                |input| {
+                    let (batch, filter) = &input;
+                    let (result, _, _) = filter.filter(batch.clone()).expect("shouldn't fail");
+                    black_box(result)
+                },
+                BatchSize::SmallInput,
+            );
+        })
+    }
+
+    group.finish();
+}
+
+fn bench_filter(c: &mut Criterion) {
+    let batch_sizes = [32, 1024, 8092];
+
+    let include = LogMatchProperties::new(
+        MatchType::Strict,
+        Vec::new(),          // no resource attr filter,
+        Vec::new(),          // no record attrs filter,
+        vec!["WARN".into()], // severity_text = "WARN",
+        None,                // no severity number filter,
+        Vec::new(),          // no bodies filter
+    );
+    bench_log_filter(c, &batch_sizes, "simple_field_filter", Some(include), None);
+
+    let include = LogMatchProperties::new(
+        MatchType::Strict,
+        Vec::new(), // no resource attr filter,
+        vec![
+            // attrs["code.namespace"] == "main"
+            filter::KeyValue::new(
+                "code.namespace".into(),
+                filter::AnyValue::String("main".into()),
+            ),
+        ],
+        Vec::new(), // no severity text filter
+        None,       // no severity number filter,
+        Vec::new(), // no bodies filter
+    );
+    bench_log_filter(c, &batch_sizes, "simple_attrs_filter", Some(include), None);
+
+    let include = LogMatchProperties::new(
+        MatchType::Strict,
+        Vec::new(),
+        vec![
+            // attrs["code.namespace"] == "main" or attrs["code.line.number"] == 2
+            filter::KeyValue::new(
+                "code.namespace".into(),
+                filter::AnyValue::String("main".into()),
+            ),
+            filter::KeyValue::new("code.line.number".into(), filter::AnyValue::Int(2)),
+        ],
+        Vec::new(), // no severity text filter
+        None,       // no severity number filter,
+        Vec::new(), // no bodies filter
+    );
+    bench_log_filter(c, &batch_sizes, "attrs_or_filter", Some(include), None);
+
+    let include = LogMatchProperties::new(
+        MatchType::Strict,
+        Vec::new(),
+        vec![
+            // attrs["code.namespace"] == "main"
+            filter::KeyValue::new(
+                "code.namespace".into(),
+                filter::AnyValue::String("main".into()),
+            ),
+        ],
+        vec!["WARN".into()], // severity_text == "WARN"
+        None,                // no severity number filter,
+        Vec::new(),          // no bodies filter
+    );
+    bench_log_filter(c, &batch_sizes, "attr_and_prop_filter", Some(include), None);
+}
+
 /// Benchmark for [`build_uint16_id_filter`]
 ///
 /// # Motivation:
@@ -155,7 +255,7 @@ mod benches {
     criterion_group!(
         name = benches;
         config = Criterion::default();
-        targets = bench_build_uint16_id_filter
+        targets = bench_filter, bench_build_uint16_id_filter
     );
 }
 

rust/otap-dataflow/crates/pdata/src/testing/fixtures.rs

@@ -273,6 +273,59 @@ pub fn logs_multiple_resources_mixed_content() -> LogsData {
     ])
 }
 
+/// Generate logs with varying attributes and properties that follow some semantic
+/// conventions. This can be used to generate somewhat realistic set of records that
+/// of various batch sizes that could be used to test transformations such as filtering
+#[must_use]
+pub fn logs_with_varying_attributes_and_properties(batch_size: usize) -> LogsData {
+    let log_records = (0..batch_size)
+        .map(|i| {
+            // generate some log attributes that somewhat follow semantic conventions
+            let attrs = vec![
+                KeyValue::new(
+                    "code.namespace",
+                    AnyValue::new_string(match i % 3 {
+                        0 => "main",
+                        1 => "otap_dataflow_engine",
+                        _ => "arrow::array",
+                    }),
+                ),
+                KeyValue::new("code.line.number", AnyValue::new_int((i % 5) as i64)),
+            ];
+
+            // cycle through severity numbers
+            // 5 = DEBUG, 9 = INFO, 13 = WARN, 17 = ERROR
+            let severity_number =
+                SeverityNumber::try_from(((i % 4) * 4 + 1) as i32).expect("valid severity_number");
+            let severity_text = severity_number
+                .as_str_name()
+                .split("_") // Note: this splitting something like SEVERITY_NUMBER_INFO
+                .nth(2)
+                .expect("can parse severity_text");
+            let event_name = format!("event {}", i);
+            let time_unix_nano = i as u64;
+
+            LogRecord::build()
+                .attributes(attrs)
+                .event_name(event_name)
+                .severity_number(severity_number)
+                .severity_text(severity_text)
+                .time_unix_nano(time_unix_nano)
+                .finish()
+        })
+        .collect::<Vec<_>>();
+
+    LogsData {
+        resource_logs: vec![ResourceLogs {
+            scope_logs: vec![ScopeLogs {
+                log_records,
+                ..Default::default()
+            }],
+            ..Default::default()
+        }],
+    }
+}
+
 //
 // Traces Fixtures
 //