Wrap resulting PipelineExpression in ParserResult (open-telemetry#1521)

Superficial change for futureproofing. Doing some initial exploration
about tracking `ParserMapSchema` changes during parser executions - if
it is possible to return modified output schema to the caller IN
ADDITION TO `PipelineExpression`, this abstraction makes it easier to
add.

Author: drewrelmas

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

@@ -30,8 +30,9 @@ fn bench_log_pipeline(
         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);
+                let parser_result =
+                    KqlParser::parse(bench_pipeline_kql).expect("can parse pipeline");
+                let mut pipeline = Pipeline::new(parser_result.pipeline);
                 rt.block_on(async move {
                     // execute the query once to initiate planning
                     pipeline.execute(batch.clone()).await.unwrap();

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

@@ -1177,8 +1177,8 @@ mod test {
 
     pub async fn exec_logs_pipeline(kql_expr: &str, logs_data: LogsData) -> LogsData {
         let otap_batch = otlp_to_otap(&OtlpProtoMessage::Logs(logs_data));
-        let pipeline_expr = KqlParser::parse(kql_expr).unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse(kql_expr).unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         otap_to_logs_data(result)
     }
@@ -1200,8 +1200,8 @@ mod test {
                 .finish(),
         ]);
 
-        let pipeline_expr = KqlParser::parse("logs | where severity_text == \"ERROR\"").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse("logs | where severity_text == \"ERROR\"").unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         let expected = to_otap(vec![
             LogRecord::build()
@@ -1212,8 +1212,8 @@ mod test {
         assert_eq!(result, expected);
 
         // test same filter where the literal is on the left and column name on the right
-        let pipeline_expr = KqlParser::parse("logs | where \"ERROR\" == severity_text").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse("logs | where \"ERROR\" == severity_text").unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         assert_eq!(result, expected);
     }
@@ -1242,8 +1242,8 @@ mod test {
                 .finish(),
         ];
 
-        let pipeline_expr = KqlParser::parse("logs | where attributes[\"x\"] == \"b\"").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse("logs | where attributes[\"x\"] == \"b\"").unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         let result_otlp = otap_to_logs_data(result);
         pretty_assertions::assert_eq!(
@@ -1252,8 +1252,8 @@ mod test {
         );
 
         // test same filter where the literal is on the left and the attribute is on the right
-        let pipeline_expr = KqlParser::parse("logs | where \"b\" == attributes[\"x\"]").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse("logs | where \"b\" == attributes[\"x\"]").unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         let result_otlp = otap_to_logs_data(result);
         pretty_assertions::assert_eq!(
@@ -1449,10 +1449,10 @@ mod test {
         let otap_batch = to_otap(log_records.clone());
 
         // check simple filter "and" properties
-        let pipeline_expr =
+        let parser_result =
             KqlParser::parse("logs | where severity_text == \"ERROR\" and event_name == \"2\"")
                 .unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         let result_otlp = otap_to_logs_data(result);
         pretty_assertions::assert_eq!(
@@ -1461,11 +1461,11 @@ mod test {
         );
 
         // check simple filter "and" with attributes
-        let pipeline_expr = KqlParser::parse(
+        let parser_result = KqlParser::parse(
             "logs | where severity_text == \"ERROR\" and attributes[\"x\"] == \"c\"",
         )
         .unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         let result_otlp = otap_to_logs_data(result);
         pretty_assertions::assert_eq!(
@@ -1474,11 +1474,11 @@ mod test {
         );
 
         // check simple filter "and" two attributes
-        let pipeline_expr = KqlParser::parse(
+        let parser_result = KqlParser::parse(
             "logs | where attributes[\"y\"] == \"d\" and attributes[\"x\"] == \"a\"",
         )
         .unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         let result_otlp = otap_to_logs_data(result);
         pretty_assertions::assert_eq!(
@@ -1518,11 +1518,11 @@ mod test {
         let otap_batch = to_otap(log_records.clone());
 
         // check simple filter "or" with properties predicates
-        let pipeline_expr = KqlParser::parse(
+        let parser_result = KqlParser::parse(
             "logs | where severity_text == \"INFO\" or severity_text == \"ERROR\"",
         )
         .unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         let result_otlp = otap_to_logs_data(result);
         pretty_assertions::assert_eq!(
@@ -1531,11 +1531,11 @@ mod test {
         );
 
         // check simple filter "or" with mixed attributes/properties predicates
-        let pipeline_expr = KqlParser::parse(
+        let parser_result = KqlParser::parse(
             "logs | where severity_text == \"ERROR\" or attributes[\"x\"] == \"c\"",
         )
         .unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         let result_otlp = otap_to_logs_data(result);
         pretty_assertions::assert_eq!(
@@ -1544,11 +1544,11 @@ mod test {
         );
 
         // check simple filter "or" two attributes predicates
-        let pipeline_expr = KqlParser::parse(
+        let parser_result = KqlParser::parse(
             "logs | where attributes[\"x\"] == \"a\" or attributes[\"y\"] == \"e\"",
         )
         .unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_batch.clone()).await.unwrap();
         let result_otlp = otap_to_logs_data(result);
         pretty_assertions::assert_eq!(
@@ -1835,8 +1835,8 @@ mod test {
                 .finish(),
         ];
 
-        let pipeline_expr = KqlParser::parse("logs | where event_name == \"5\"").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse("logs | where event_name == \"5\"").unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline
             .execute(to_otap(log_records.clone()))
             .await
@@ -1845,8 +1845,8 @@ mod test {
         assert_eq!(result, OtapArrowRecords::Logs(Logs::default()));
 
         // assert we have the correct behaviour when filtering by attributes as well
-        let pipeline_expr = KqlParser::parse("logs | where attributes[\"a\"] == \"1234\"").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse("logs | where attributes[\"a\"] == \"1234\"").unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline
             .execute(to_otap(log_records.clone()))
             .await
@@ -1857,8 +1857,8 @@ mod test {
     #[tokio::test]
     async fn test_empty_batch() {
         let input = OtapArrowRecords::Logs(Logs::default());
-        let pipeline_expr = KqlParser::parse("logs | where event_name == \"5\"").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse("logs | where event_name == \"5\"").unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(input.clone()).await.unwrap();
         assert_eq!(result, input);
     }
@@ -1881,29 +1881,29 @@ mod test {
         ];
 
         // check that if there are no attributes to filter by then, we get the empty batch
-        let pipeline_expr = KqlParser::parse("logs | where attributes[\"a\"] == \"1234\"").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse("logs | where attributes[\"a\"] == \"1234\"").unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline
             .execute(to_otap(log_records.clone()))
             .await
             .unwrap();
         assert_eq!(result, OtapArrowRecords::Logs(Logs::default()));
 
         // check that the same result happens when filtering by resource and scope attrs
-        let pipeline_expr =
+        let parser_result =
             KqlParser::parse("logs | where resource.attributes[\"a\"] == \"1234\"").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline
             .execute(to_otap(log_records.clone()))
             .await
             .unwrap();
         assert_eq!(result, OtapArrowRecords::Logs(Logs::default()));
 
         // check that the same result happens when filtering by resource and scope attrs
-        let pipeline_expr =
+        let parser_result =
             KqlParser::parse("logs | where instrumentation_scope.attributes[\"a\"] == \"1234\"")
                 .unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline
             .execute(to_otap(log_records.clone()))
             .await
@@ -1916,8 +1916,8 @@ mod test {
             "logs | where not(resource.attributes[\"a\"] == \"1234\")",
             "logs | where not(instrumentation_scope.attributes[\"a\"] == \"1234\")",
         ] {
-            let pipeline_expr = KqlParser::parse(inverted_attrs_filter).unwrap();
-            let mut pipeline = Pipeline::new(pipeline_expr);
+            let parser_result = KqlParser::parse(inverted_attrs_filter).unwrap();
+            let mut pipeline = Pipeline::new(parser_result.pipeline);
             let input = to_otap(log_records.clone());
             let result = pipeline.execute(input.clone()).await.unwrap();
             assert_eq!(result, input);
@@ -1930,8 +1930,8 @@ mod test {
         // next, then present in the next, etc.
 
         let query = "logs | where attributes[\"a\"] == \"1234\"";
-        let pipeline_expr = KqlParser::parse(query).unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr.clone());
+        let parser_result = KqlParser::parse(query).unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
 
         // no attrs to start
         let batch1 = to_otap(vec![LogRecord::build().event_name("a").finish()]);
@@ -1987,8 +1987,8 @@ mod test {
 
         // assert the behaviour is correct when nothing is filtered out
         let otap_input = to_otap(log_records);
-        let pipeline_expr = KqlParser::parse("logs | where severity_text == \"INFO\"").unwrap();
-        let mut pipeline = Pipeline::new(pipeline_expr);
+        let parser_result = KqlParser::parse("logs | where severity_text == \"INFO\"").unwrap();
+        let mut pipeline = Pipeline::new(parser_result.pipeline);
         let result = pipeline.execute(otap_input.clone()).await.unwrap();
 
         assert_eq!(result, otap_input)

rust/experimental/query_engine/engine-recordset-otlp-bridge/src/bridge.rs

@@ -19,7 +19,9 @@ pub fn parse_kql_query_into_pipeline(
     query: &str,
     options: Option<BridgeOptions>,
 ) -> Result<PipelineExpression, Vec<ParserError>> {
-    KqlParser::parse_with_options(query, build_parser_options(options).map_err(|e| vec![e])?)
+    let result =
+        KqlParser::parse_with_options(query, build_parser_options(options).map_err(|e| vec![e])?)?;
+    Ok(result.pipeline)
 }
 
 pub fn register_pipeline_for_kql_query(
@@ -28,7 +30,8 @@ pub fn register_pipeline_for_kql_query(
 ) -> Result<usize, Vec<ParserError>> {
     let options = build_parser_options(options).map_err(|e| vec![e])?;
 
-    let pipeline = KqlParser::parse_with_options(query, options.clone())?;
+    let result = KqlParser::parse_with_options(query, options.clone())?;
+    let pipeline = result.pipeline;
 
     let mut expressions = EXPRESSIONS.write().unwrap();
     expressions.push((options, pipeline));
@@ -730,7 +733,8 @@ mod tests {
             "source | where gettype(TimeGenerated) == 'datetime'",
             options.clone(),
         )
-        .unwrap();
+        .unwrap()
+        .pipeline;
 
         let (included_records, dropped_records) =
             process_export_logs_service_request_using_pipeline(

rust/experimental/query_engine/kql-parser/src/kql_parser.rs

@@ -1,7 +1,6 @@
 // Copyright The OpenTelemetry Authors
 // SPDX-License-Identifier: Apache-2.0
 
-use data_engine_expressions::*;
 use data_engine_parser_abstractions::*;
 use pest_derive::Parser;
 
@@ -17,8 +16,9 @@ impl Parser for KqlParser {
     fn parse_with_options(
         query: &str,
         options: ParserOptions,
-    ) -> Result<PipelineExpression, Vec<ParserError>> {
-        parse_query(query, options)
+    ) -> Result<ParserResult, Vec<ParserError>> {
+        let pipeline = parse_query(query, options)?;
+        Ok(ParserResult::new(pipeline))
     }
 }
 

rust/experimental/query_engine/kql-parser/src/lib.rs

@@ -28,3 +28,4 @@ pub use data_engine_parser_abstractions::ParserError;
 pub use data_engine_parser_abstractions::ParserMapKeySchema;
 pub use data_engine_parser_abstractions::ParserMapSchema;
 pub use data_engine_parser_abstractions::ParserOptions;
+pub use data_engine_parser_abstractions::ParserResult;

rust/experimental/query_engine/ottl-parser/src/lib.rs

@@ -16,3 +16,4 @@ pub use data_engine_parser_abstractions::ParserError;
 pub use data_engine_parser_abstractions::ParserMapKeySchema;
 pub use data_engine_parser_abstractions::ParserMapSchema;
 pub use data_engine_parser_abstractions::ParserOptions;
+pub use data_engine_parser_abstractions::ParserResult;

rust/experimental/query_engine/parser-abstractions/src/parser.rs

@@ -8,14 +8,14 @@ use data_engine_expressions::*;
 use crate::*;
 
 pub trait Parser {
-    fn parse(query: &str) -> Result<PipelineExpression, Vec<ParserError>> {
+    fn parse(query: &str) -> Result<ParserResult, Vec<ParserError>> {
         Self::parse_with_options(query, ParserOptions::new())
     }
 
     fn parse_with_options(
         query: &str,
         options: ParserOptions,
-    ) -> Result<PipelineExpression, Vec<ParserError>>;
+    ) -> Result<ParserResult, Vec<ParserError>>;
 }
 
 type ParserFunctionDefinition = (
@@ -318,3 +318,18 @@ impl TryFrom<&str> for ParserMapKeySchema {
         }
     }
 }
+
+/// Result returned by parsers, containing the parsed pipeline expression
+/// and any additional metadata that may be useful for consumers.
+#[derive(Debug, Clone, PartialEq)]
+pub struct ParserResult {
+    /// The parsed pipeline expression
+    pub pipeline: PipelineExpression,
+}
+
+impl ParserResult {
+    /// Create a new ParserResult with the given pipeline expression
+    pub fn new(pipeline: PipelineExpression) -> Self {
+        Self { pipeline }
+    }
+}