[query-engine] Add data record and batch definitions to recordset engine (open-telemetry#585)

## Changes

* Defines data record and data record batch in the recordset query
engine

Author: CodeBlanch

rust/experimental/query_engine/engine-recordset/src/data/data_record.rs

@@ -0,0 +1,166 @@
+use core::fmt::Debug;
+use std::{any::Any, time::SystemTime};
+
+use crate::{primitives::any_value::AnyValue, value_path::ValuePath};
+
+use super::data_record_resolver::*;
+
+pub trait DataRecord: Any + Debug {
+    fn get_timestamp(&self) -> Option<SystemTime>;
+
+    fn get_observed_timestamp(&self) -> Option<SystemTime>;
+
+    fn get_any_value_resolver_for_path(path: &ValuePath) -> DataRecordAnyValueResolver<Self>
+    where
+        Self: Sized;
+
+    fn get_summary_id(&self) -> Option<&str>;
+
+    fn set_summary_id(&mut self, summary_id: &str);
+
+    fn clear(&mut self);
+}
+
+pub trait AttachedDataRecords: Debug {
+    fn get_attached_data_record(&self, name: &str) -> Option<&dyn DataRecord>;
+}
+
+pub fn create_string_value_resolver<T: DataRecord, R, M, S>(
+    path: &ValuePath,
+    read_action: &'static R,
+    read_mut_action: &'static M,
+    set_action: &'static S,
+) -> DataRecordAnyValueResolver<T>
+where
+    R: Fn(&T) -> Option<&AnyValue>,
+    M: Fn(&mut T) -> Option<&mut AnyValue>,
+    S: Fn(&mut T, Option<AnyValue>) -> Option<AnyValue>,
+{
+    if !path.is_value_selector() {
+        return DataRecordAnyValueResolver::new_no_op();
+    }
+
+    return DataRecordAnyValueResolver::new(
+        path.clone(),
+        |_, data_record| {
+            let root = read_action(data_record);
+            match root {
+                Some(v) => DataRecordReadAnyValueResult::Found(v),
+                None => DataRecordReadAnyValueResult::NotFound,
+            }
+        },
+        |_, data_record| {
+            let root = read_mut_action(data_record);
+            match root {
+                Some(v) => DataRecordReadMutAnyValueResult::Found(v),
+                None => DataRecordReadMutAnyValueResult::NotFound,
+            }
+        },
+        move |_, data_record, v| {
+            if let AnyValue::StringValue(_) = &v {
+                let old_value = set_action(data_record, Some(v));
+
+                if old_value.is_none() {
+                    return DataRecordSetAnyValueResult::Created;
+                }
+
+                return DataRecordSetAnyValueResult::Updated(old_value.unwrap());
+            } else if let AnyValue::NullValue = &v {
+                let old_value = set_action(data_record, None);
+
+                if old_value.is_none() {
+                    return DataRecordSetAnyValueResult::NotFound;
+                }
+
+                return DataRecordSetAnyValueResult::Updated(old_value.unwrap());
+            }
+
+            return DataRecordSetAnyValueResult::NotSupported("Value was not a String");
+        },
+        move |_, data_record| {
+            let old_value = set_action(data_record, None);
+
+            if old_value.is_none() {
+                return DataRecordRemoveAnyValueResult::NotFound;
+            }
+
+            return DataRecordRemoveAnyValueResult::Removed(old_value.unwrap());
+        },
+    );
+}
+
+pub fn create_map_value_resolver<T: DataRecord, R, M, S>(
+    path: &ValuePath,
+    read_action: &'static R,
+    read_mut_action: &'static M,
+    set_action: &'static S,
+) -> DataRecordAnyValueResolver<T>
+where
+    R: Fn(&T) -> Option<&AnyValue>,
+    M: Fn(&mut T) -> Option<&mut AnyValue>,
+    S: Fn(&mut T, Option<AnyValue>) -> Option<AnyValue>,
+{
+    return DataRecordAnyValueResolver::new(
+        path.clone(),
+        |path, data_record| {
+            let root = read_action(data_record);
+            match root {
+                Some(v) => path.read(v),
+                None => DataRecordReadAnyValueResult::NotFound,
+            }
+        },
+        |path, data_record| {
+            let root = read_mut_action(data_record);
+            match root {
+                Some(v) => path.read_mut(v),
+                None => DataRecordReadMutAnyValueResult::NotFound,
+            }
+        },
+        move |path, data_record, v| {
+            if path.is_value_selector() {
+                if let AnyValue::MapValue(_) = &v {
+                    let old_value = set_action(data_record, Some(v));
+
+                    if old_value.is_none() {
+                        return DataRecordSetAnyValueResult::Created;
+                    }
+
+                    return DataRecordSetAnyValueResult::Updated(old_value.unwrap());
+                } else if let AnyValue::NullValue = &v {
+                    let old_value = set_action(data_record, None);
+
+                    if old_value.is_none() {
+                        return DataRecordSetAnyValueResult::NotFound;
+                    }
+
+                    return DataRecordSetAnyValueResult::Updated(old_value.unwrap());
+                }
+
+                return DataRecordSetAnyValueResult::NotSupported("Value was not a Map");
+            } else {
+                let root = read_mut_action(data_record);
+                match root {
+                    Some(r) => path.set(r, v),
+                    None => DataRecordSetAnyValueResult::NotFound,
+                }
+            }
+        },
+        move |path, data_record| {
+            if path.is_value_selector() {
+                let old_value = set_action(data_record, None);
+
+                if old_value.is_none() {
+                    return DataRecordRemoveAnyValueResult::NotFound;
+                }
+
+                return DataRecordRemoveAnyValueResult::Removed(old_value.unwrap());
+            } else {
+                let root = read_mut_action(data_record);
+                match root {
+                    Some(r) => path.remove(r),
+                    None => DataRecordRemoveAnyValueResult::NotFound,
+                }
+            }
+        },
+    );
+}

rust/experimental/query_engine/engine-recordset/src/data/data_record_batch.rs

@@ -0,0 +1,9 @@
+use core::fmt::Debug;
+
+use crate::{data_engine::DataEngineState, error::Error};
+
+pub trait DataRecordBatch<T>: Debug {
+    fn drain<S: DataEngineState, F>(&mut self, state: &mut S, action: F) -> Result<(), Error>
+    where
+        F: Fn(&mut S, T) -> Result<(), Error>;
+}

rust/experimental/query_engine/engine-recordset/src/data/data_record_resolver.rs

@@ -0,0 +1,202 @@
+use std::{any::Any, cell::RefCell, mem::replace};
+
+use crate::{Error, ValuePath, execution_context::ExecutionContext, primitives::AnyValue};
+
+use super::DataRecord;
+
+pub(crate) trait DynamicDataRecordAnyValueResolver: Any {
+    fn read_value(
+        &self,
+        expression_id: usize,
+        execution_context: &dyn ExecutionContext,
+        path: &ValuePath,
+        data_record: &dyn DataRecord,
+        action: &mut dyn DataRecordAnyValueReadCallback,
+    ) -> Result<(), Error>;
+}
+
+pub struct DataRecordAnyValueResolver<T: DataRecord> {
+    path: ValuePath,
+    read_value_fn:
+        Box<dyn for<'a, 'b> Fn(&'a ValuePath, &'b T) -> DataRecordReadAnyValueResult<'b>>,
+    read_value_mut_fn:
+        Box<dyn for<'a, 'b> Fn(&'a ValuePath, &'b mut T) -> DataRecordReadMutAnyValueResult<'b>>,
+    set_value_fn: Box<dyn Fn(&ValuePath, &mut T, AnyValue) -> DataRecordSetAnyValueResult>,
+    remove_value_fn: Box<dyn Fn(&ValuePath, &mut T) -> DataRecordRemoveAnyValueResult>,
+}
+
+impl<T: DataRecord> DataRecordAnyValueResolver<T> {
+    pub fn new(
+        path: ValuePath,
+        read_value: impl for<'a, 'b> Fn(&'a ValuePath, &'b T) -> DataRecordReadAnyValueResult<'b>
+        + 'static,
+        read_value_mut: impl for<'a, 'b> Fn(
+            &'a ValuePath,
+            &'b mut T,
+        ) -> DataRecordReadMutAnyValueResult<'b>
+        + 'static,
+        set_value: impl Fn(&ValuePath, &mut T, AnyValue) -> DataRecordSetAnyValueResult + 'static,
+        remove_value: impl Fn(&ValuePath, &mut T) -> DataRecordRemoveAnyValueResult + 'static,
+    ) -> DataRecordAnyValueResolver<T> {
+        Self {
+            path,
+            read_value_fn: Box::new(read_value),
+            read_value_mut_fn: Box::new(read_value_mut),
+            set_value_fn: Box::new(set_value),
+            remove_value_fn: Box::new(remove_value),
+        }
+    }
+
+    pub fn new_no_op() -> DataRecordAnyValueResolver<T> {
+        DataRecordAnyValueResolver::new(
+            ValuePath::new("").unwrap(),
+            |_, _| DataRecordReadAnyValueResult::NotFound,
+            |_, _| DataRecordReadMutAnyValueResult::NotFound,
+            |_, _, _| DataRecordSetAnyValueResult::NotFound,
+            |_, _| DataRecordRemoveAnyValueResult::NotFound,
+        )
+    }
+
+    pub(crate) fn read_value<F>(&self, data_record: &RefCell<T>, action: F)
+    where
+        F: FnOnce(DataRecordReadAnyValueResult),
+    {
+        let borrow = data_record.borrow();
+
+        let result = (self.read_value_fn)(&self.path, &borrow);
+
+        action(result);
+    }
+
+    pub(crate) fn read_value_direct<F>(&self, data_record: &T, action: F)
+    where
+        F: FnOnce(DataRecordReadAnyValueResult),
+    {
+        let result = (self.read_value_fn)(&self.path, data_record);
+
+        action(result);
+    }
+
+    pub(crate) fn read_value_mut<F>(&self, data_record: &RefCell<T>, action: F)
+    where
+        F: FnOnce(DataRecordReadMutAnyValueResult),
+    {
+        let mut borrow = data_record.borrow_mut();
+
+        let result = (self.read_value_mut_fn)(&self.path, &mut borrow);
+
+        action(result);
+    }
+
+    pub(crate) fn set_value(
+        &self,
+        data_record: &RefCell<T>,
+        value: AnyValue,
+    ) -> DataRecordSetAnyValueResult {
+        let mut borrow = data_record.borrow_mut();
+
+        return (self.set_value_fn)(&self.path, &mut borrow, value);
+    }
+
+    pub(crate) fn remove_value(&self, data_record: &RefCell<T>) -> DataRecordRemoveAnyValueResult {
+        let mut borrow = data_record.borrow_mut();
+
+        return (self.remove_value_fn)(&self.path, &mut borrow);
+    }
+}
+
+#[derive(Debug)]
+pub enum DataRecordReadAnyValueResult<'a> {
+    NotFound,
+    Found(&'a AnyValue),
+}
+
+#[derive(Debug)]
+pub enum DataRecordReadMutAnyValueResult<'a> {
+    NotFound,
+    NotSupported(&'static str),
+    Found(&'a mut AnyValue),
+}
+
+#[derive(Debug)]
+pub enum DataRecordSetAnyValueResult {
+    NotFound,
+    NotSupported(&'static str),
+    Created,
+    Updated(AnyValue),
+}
+
+#[derive(Debug)]
+pub enum DataRecordRemoveAnyValueResult {
+    NotFound,
+    NotSupported(&'static str),
+    Removed(AnyValue),
+}
+
+pub(crate) trait DataRecordAnyValueReadCallback {
+    fn invoke_once(&mut self, result: DataRecordReadAnyValueResult);
+}
+
+pub(crate) struct DataRecordAnyValueReadClosureCallback<F>
+where
+    F: FnOnce(DataRecordReadAnyValueResult),
+{
+    callback: Option<F>,
+}
+
+impl<F> DataRecordAnyValueReadClosureCallback<F>
+where
+    F: FnOnce(DataRecordReadAnyValueResult),
+{
+    pub fn new(callback: F) -> DataRecordAnyValueReadClosureCallback<F> {
+        Self {
+            callback: Some(callback),
+        }
+    }
+}
+
+impl<F> DataRecordAnyValueReadCallback for DataRecordAnyValueReadClosureCallback<F>
+where
+    F: FnOnce(DataRecordReadAnyValueResult),
+{
+    fn invoke_once(&mut self, result: DataRecordReadAnyValueResult) {
+        let callback = replace(&mut self.callback, None);
+        if !callback.is_none() {
+            (callback.unwrap())(result);
+        }
+    }
+}
+
+pub(crate) trait DataRecordAnyValueReadMutCallback {
+    fn invoke_once(&mut self, result: DataRecordReadMutAnyValueResult);
+}
+
+pub(crate) struct DataRecordAnyValueReadMutClosureCallback<F>
+where
+    F: FnOnce(DataRecordReadMutAnyValueResult),
+{
+    callback: Option<F>,
+}
+
+impl<F> DataRecordAnyValueReadMutClosureCallback<F>
+where
+    F: FnOnce(DataRecordReadMutAnyValueResult),
+{
+    pub fn new(callback: F) -> DataRecordAnyValueReadMutClosureCallback<F> {
+        Self {
+            callback: Some(callback),
+        }
+    }
+}
+
+impl<F> DataRecordAnyValueReadMutCallback for DataRecordAnyValueReadMutClosureCallback<F>
+where
+    F: FnOnce(DataRecordReadMutAnyValueResult),
+{
+    fn invoke_once(&mut self, result: DataRecordReadMutAnyValueResult) {
+        let callback = replace(&mut self.callback, None);
+        if !callback.is_none() {
+            (callback.unwrap())(result);
+        }
+    }
+}