columnar query engine support basic filtering of spans and metrics (open-telemetry#1560)

part of open-telemetry#1508

Adds support to the columnar query engine to filter metrics and traces.
The thing that needed to be implemented to support this was just to
traverse down the tree of payload types and remove rows whose parent was
filtered out.

Because attributes w/ 32 bit parent IDs can have this column dictionary
encoded, I decided to expose the arrays::MaybeDictionaryAccessor from
the OTAP Dataflow Pdata crate. I wasn't ready to make everything in this
module public, so many methods and types are changed to `pub(crate)` in
this module.

Note - only basic filtering support is added currently. There's no
capability to say, filter metrics by datapoint attributes, filter
datapoint attributes themselves, and likewise no ability to filter
span's spanevents/span links. All this can be added in future

Author: albertlockett

rust/experimental/query_engine/engine-columnar/src/pipeline.rs

@@ -257,6 +257,12 @@ mod test {
     use datafusion::logical_expr::{col, lit};
     use otap_df_pdata::proto::OtlpProtoMessage;
     use otap_df_pdata::proto::opentelemetry::arrow::v1::ArrowPayloadType;
+    use otap_df_pdata::proto::opentelemetry::metrics::v1::{
+        Metric, MetricsData, ResourceMetrics, ScopeMetrics,
+    };
+    use otap_df_pdata::proto::opentelemetry::trace::v1::{
+        ResourceSpans, ScopeSpans, Span, TracesData,
+    };
 
     use otap_df_pdata::proto::opentelemetry::logs::v1::{
         LogRecord, LogsData, ResourceLogs, ScopeLogs,
@@ -278,19 +284,55 @@ mod test {
         }
     }
 
+    /// helper function for converting [`Metric`]s to [`MetricsData`]
+    pub fn to_metrics_data(metrics: Vec<Metric>) -> MetricsData {
+        MetricsData {
+            resource_metrics: vec![ResourceMetrics {
+                scope_metrics: vec![ScopeMetrics {
+                    metrics,
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        }
+    }
+
+    /// helper function for converting [`Span`]s to [`TracesData`]
+    pub fn to_traces_data(spans: Vec<Span>) -> TracesData {
+        TracesData {
+            resource_spans: vec![ResourceSpans {
+                scope_spans: vec![ScopeSpans {
+                    spans,
+                    ..Default::default()
+                }],
+                ..Default::default()
+            }],
+        }
+    }
+
     /// helper function for converting OTLP logs to OTAP batch
-    pub fn to_otap(log_records: Vec<LogRecord>) -> OtapArrowRecords {
+    pub fn to_otap_logs(log_records: Vec<LogRecord>) -> OtapArrowRecords {
         otlp_to_otap(&OtlpProtoMessage::Logs(to_logs_data(log_records)))
     }
 
+    /// helper function for converting OTLP spans to OTAP batch
+    pub fn to_otap_traces(spans: Vec<Span>) -> OtapArrowRecords {
+        otlp_to_otap(&OtlpProtoMessage::Traces(to_traces_data(spans)))
+    }
+
+    /// helper function for converting OTLP metrics to OTAP batch
+    pub fn to_otap_metrics(metrics: Vec<Metric>) -> OtapArrowRecords {
+        otlp_to_otap(&OtlpProtoMessage::Metrics(to_metrics_data(metrics)))
+    }
+
     #[tokio::test]
     async fn test_pipeline_execute_multi_batch() {
         // TODO eventually we might want to drive this test from a pipeline expression, which we
         // can do once we have the query planning implemented. For now, we are manually creating
         // the `PlannedPipeline` and its `PipelineStage`s and any additional datafusion context
         // they need
 
-        let otap_batch1 = to_otap(vec![
+        let otap_batch1 = to_otap_logs(vec![
             LogRecord {
                 severity_text: "ERROR".into(),
                 event_name: "1".into(),
@@ -308,7 +350,7 @@ mod test {
             },
         ]);
 
-        let otap_batch2 = to_otap(vec![
+        let otap_batch2 = to_otap_logs(vec![
             LogRecord {
                 severity_text: "DEBUG".into(),
                 event_name: "4".into(),