parse to new 'GetRecordType' ScalarExpression

Author: albertlockett

rust/experimental/query_engine/engine-recordset/src/scalars/scalar_expressions.rs

@@ -326,6 +326,12 @@ where
         ScalarExpression::Math(m) => {
             return execute_math_scalar_expression(execution_context, m);
         }
+        ScalarExpression::GetRecordType(r) => {
+            return Err(ExpressionError::NotSupported(
+                r.get_query_location().clone(),
+                format!("{} not yet supported  in RecordSet engine", r.get_name()),
+            ));
+        }
         ScalarExpression::GetType(g) => {
             let value_type =
                 execute_scalar_expression(execution_context, g.get_value())?.get_value_type();

rust/experimental/query_engine/expressions/src/scalars/scalar_expressions.rs

@@ -43,6 +43,9 @@ pub enum ScalarExpression {
     /// Returns the type string for the inner expression.
     GetType(GetTypeScalarExpression),
 
+    /// Returns the type string for the record.
+    GetRecordType(GetRecordTypeScalarExpression),
+
     /// Invoke a user-defined function.
     InvokeFunction(InvokeFunctionScalarExpression),
 
@@ -101,6 +104,7 @@ impl ScalarExpression {
             ScalarExpression::Constant(c) => c.try_resolve_value_type(scope),
             ScalarExpression::Collection(c) => c.try_resolve_value_type(scope),
             ScalarExpression::GetType(_) => Ok(Some(ValueType::String)),
+            ScalarExpression::GetRecordType(_) => Ok(Some(ValueType::String)),
             ScalarExpression::Logical(_) => Ok(Some(ValueType::Boolean)),
             ScalarExpression::Coalesce(c) => c.try_resolve_value_type(scope),
             ScalarExpression::Conditional(c) => c.try_resolve_value_type(scope),
@@ -205,6 +209,7 @@ impl ScalarExpression {
             ScalarExpression::Case(c) => c.try_resolve_static(scope),
             ScalarExpression::Convert(c) => c.try_resolve_static(scope),
             ScalarExpression::GetType(g) => g.try_resolve_static(scope),
+            ScalarExpression::GetRecordType(g) => g.try_resolve_static(scope),
             ScalarExpression::Length(l) => l.try_resolve_static(scope),
             ScalarExpression::Slice(s) => s.try_resolve_static(scope),
             ScalarExpression::Parse(p) => p.try_resolve_static(scope),
@@ -231,6 +236,7 @@ impl Expression for ScalarExpression {
             ScalarExpression::Constant(c) => c.get_query_location(),
             ScalarExpression::Collection(c) => c.get_query_location(),
             ScalarExpression::GetType(g) => g.get_query_location(),
+            ScalarExpression::GetRecordType(g) => g.get_query_location(),
             ScalarExpression::Logical(l) => l.get_query_location(),
             ScalarExpression::Coalesce(c) => c.get_query_location(),
             ScalarExpression::Conditional(c) => c.get_query_location(),
@@ -256,6 +262,7 @@ impl Expression for ScalarExpression {
             ScalarExpression::Static(s) => s.get_name(),
             ScalarExpression::Collection(_) => "ScalarExpression(Collection)",
             ScalarExpression::GetType(_) => "ScalarExpression(GetType)",
+            ScalarExpression::GetRecordType(_) => "ScalarExpression(GetRecordType)",
             ScalarExpression::Logical(_) => "ScalarExpression(Logical)",
             ScalarExpression::Coalesce(_) => "ScalarExpression(Coalesce)",
             ScalarExpression::Conditional(_) => "ScalarExpression(Conditional)",
@@ -284,6 +291,7 @@ impl Expression for ScalarExpression {
             ScalarExpression::Conditional(c) => c.fmt_with_indent(f, indent),
             ScalarExpression::Convert(c) => c.fmt_with_indent(f, indent),
             ScalarExpression::GetType(g) => g.fmt_with_indent(f, indent),
+            ScalarExpression::GetRecordType(g) => g.fmt_with_indent(f, indent),
             ScalarExpression::Temporal(t) => t.fmt_with_indent(f, indent),
             ScalarExpression::Length(l) => l.fmt_with_indent(f, indent),
             ScalarExpression::Logical(l) => l.fmt_with_indent(f, indent),
@@ -1307,6 +1315,39 @@ impl Expression for GetTypeScalarExpression {
     }
 }
 
+#[derive(Debug, Clone, PartialEq)]
+pub struct GetRecordTypeScalarExpression {
+    query_location: QueryLocation,
+}
+
+impl GetRecordTypeScalarExpression {
+    pub fn new(query_location: QueryLocation) -> GetRecordTypeScalarExpression {
+        Self { query_location }
+    }
+
+    pub(crate) fn try_resolve_static(
+        &mut self,
+        _scope: &PipelineResolutionScope,
+    ) -> ScalarStaticResolutionResult<'_> {
+        ScalarStaticResolutionResult::Ok(None)
+    }
+}
+
+impl Expression for GetRecordTypeScalarExpression {
+    fn get_query_location(&self) -> &QueryLocation {
+        &self.query_location
+    }
+
+    fn get_name(&self) -> &'static str {
+        "GetRecordTypeScalarExpression"
+    }
+
+    fn fmt_with_indent(&self, f: &mut std::fmt::Formatter<'_>, _indent: &str) -> std::fmt::Result {
+        writeln!(f, "GetRecordTypeScalarExpression")?;
+        Ok(())
+    }
+}
+
 #[derive(Debug, Clone, PartialEq)]
 pub struct SelectScalarExpression {
     query_location: QueryLocation,

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

@@ -659,6 +659,7 @@ pub(crate) fn try_resolve_identifier(
         ScalarExpression::Slice(_) => Ok(None),
         ScalarExpression::Static(_) => Ok(None),
         ScalarExpression::Text(_) => Ok(None),
+        ScalarExpression::GetRecordType(_) => Ok(None),
         ScalarExpression::GetType(g) => {
             if let Some(mut i) = try_resolve_identifier(g.get_value(), scope)? {
                 i.insert(0, "type".into());
@@ -667,6 +668,7 @@ pub(crate) fn try_resolve_identifier(
 
             Ok(None)
         }
+
         ScalarExpression::Select(s) => {
             if let Some(mut value) = try_resolve_identifier(s.get_value(), scope)?
                 && let ScalarExpression::Static(StaticScalarExpression::Array(selectors)) =

rust/otap-dataflow/crates/opl/src/parser/expression.rs

@@ -7,9 +7,9 @@ use data_engine_expressions::{
     AndLogicalExpression, BinaryMathematicalScalarExpression, BooleanScalarExpression,
     CaptureTextScalarExpression, CollectionScalarExpression, CombineScalarExpression,
     ContainsLogicalExpression, DateTimeScalarExpression, DoubleScalarExpression, DoubleValue,
-    EqualToLogicalExpression, Expression, GetTypeScalarExpression, GreaterThanLogicalExpression,
-    GreaterThanOrEqualToLogicalExpression, IntegerScalarExpression, IntegerValue,
-    InvokeFunctionArgument, InvokeFunctionScalarExpression, JoinTextScalarExpression,
+    EqualToLogicalExpression, Expression, GetRecordTypeScalarExpression,
+    GreaterThanLogicalExpression, GreaterThanOrEqualToLogicalExpression, IntegerScalarExpression,
+    IntegerValue, InvokeFunctionArgument, InvokeFunctionScalarExpression, JoinTextScalarExpression,
     ListScalarExpression, LogicalExpression, MatchesLogicalExpression, MathScalarExpression,
     NotLogicalExpression, NullScalarExpression, OrLogicalExpression, QueryLocation,
     RegexScalarExpression, ReplaceTextScalarExpression, ScalarExpression, SliceScalarExpression,
@@ -215,13 +215,9 @@ pub(crate) fn parse_type_check_expression(
         // IF there are two rules, we have an expression like (is <Type>) meaning we're checking
         // that an element of the stream is some type
         2 => {
-            let type_check_expr = ScalarExpression::GetType(GetTypeScalarExpression::new(
-                type_check_rule_query_location.clone(),
-                ScalarExpression::Source(SourceScalarExpression::new(
-                    type_check_rule_query_location.clone(),
-                    ValueAccessor::new_with_selectors(Vec::new()),
-                )),
-            ));
+            let type_check_expr = ScalarExpression::GetRecordType(
+                GetRecordTypeScalarExpression::new(type_check_rule_query_location.clone()),
+            );
 
             let type_name_rule = inner_rules.nth(1).expect("two rules");
             let type_check_expected_type = ScalarExpression::Static(

rust/otap-dataflow/crates/query-engine/src/pipeline/filter.rs

@@ -479,46 +479,25 @@ impl FilterPlan {
     ) -> Result<Option<Self>> {
         match (left_expr, binary_op, right_expr) {
             (
-                ScalarExpression::GetType(get_type_expr),
+                ScalarExpression::GetRecordType(_),
                 Operator::Eq,
                 ScalarExpression::Static(StaticScalarExpression::String(typename_expr)),
             ) => {
-                let source_value_accessor = match get_type_expr.get_value() {
-                    ScalarExpression::Source(source_scalar_expr) => {
-                        source_scalar_expr.get_value_accessor()
-                    }
-                    _ => {
-                        // the source for the get type expression isn't something that we can check
-                        // determine to statically be the identifier of a type, defer to expression
-                        // evaluation for this filter
-                        return Ok(None);
-                    }
-                };
+                // since the source accessor has no selectors, it means we're checking the type
+                // of elements of the stream for this pipeline. We'll try to determine if the
+                // type name is a stream type that is handled by this query engine ...
+                let type_name = typename_expr.get_value();
+                let stream_element_type =
+                    StreamElementType::from_str(type_name).ok_or_else(|| {
+                        Error::InvalidPipelineError {
+                            cause: format!("Unknown stream type name {type_name}"),
+                            query_location: Some(right_expr.get_query_location().clone()),
+                        }
+                    })?;
 
-                if !source_value_accessor.has_selectors() {
-                    // since the source accessor has no selectors, it means we're checking the type
-                    // of elements of the stream for this pipeline. We'll try to determine if the
-                    // type name is a stream type that is handled by this query engine ...
-                    let type_name = typename_expr.get_value();
-                    let stream_element_type =
-                        StreamElementType::from_str(type_name).ok_or_else(|| {
-                            Error::InvalidPipelineError {
-                                cause: format!("Unknown stream type name {type_name}"),
-                                query_location: Some(right_expr.get_query_location().clone()),
-                            }
-                        })?;
-
-                    Ok(Some(FilterPlan::from(ElementTypeFilter::new(
-                        stream_element_type,
-                    ))))
-                } else {
-                    // TODO - we will support this soon
-                    Err(Error::NotYetSupportedError {
-                        message:
-                            "Checking if a field of stream element is a type is not yet supported"
-                                .into(),
-                    })
-                }
+                Ok(Some(FilterPlan::from(ElementTypeFilter::new(
+                    stream_element_type,
+                ))))
             }
             _ => Ok(None),
         }