[query-engine] Support binding functions to external implementations in RecordSet engine (open-telemetry#1629)

Relates to open-telemetry#1479

## Changes

* Support binding of functions to external implementations in RecordSet
engine

Author: CodeBlanch

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

@@ -22,6 +22,8 @@ use crate::{
 pub struct RecordSetEngineOptions {
     pub(crate) diagnostic_level: RecordSetEngineDiagnosticLevel,
     pub(crate) summary_cardinality_limit: usize,
+    pub(crate) external_function_implementations:
+        HashMap<Box<str>, Box<dyn RecordSetEngineFunctionCallback>>,
 }
 
 impl Default for RecordSetEngineOptions {
@@ -35,6 +37,7 @@ impl RecordSetEngineOptions {
         Self {
             diagnostic_level: RecordSetEngineDiagnosticLevel::Warn,
             summary_cardinality_limit: 8192,
+            external_function_implementations: HashMap::new(),
         }
     }
 
@@ -53,11 +56,22 @@ impl RecordSetEngineOptions {
         self.summary_cardinality_limit = summary_cardinality_limit;
         self
     }
+
+    pub fn with_external_function_implementation<F: RecordSetEngineFunctionCallback + 'static>(
+        mut self,
+        name: &str,
+        callback: F,
+    ) -> RecordSetEngineOptions {
+        self.external_function_implementations
+            .insert(name.into(), Box::new(callback));
+        self
+    }
 }
 
 pub struct RecordSetEngine {
     diagnostic_level: RecordSetEngineDiagnosticLevel,
     summary_cardinality_limit: usize,
+    external_function_implementations: HashMap<Box<str>, Box<dyn RecordSetEngineFunctionCallback>>,
 }
 
 impl Default for RecordSetEngine {
@@ -75,6 +89,7 @@ impl RecordSetEngine {
         Self {
             diagnostic_level: options.diagnostic_level,
             summary_cardinality_limit: options.summary_cardinality_limit,
+            external_function_implementations: options.external_function_implementations,
         }
     }
 
@@ -121,6 +136,7 @@ impl<'a, 'b, TRecord: Record + 'static> RecordSetEngineBatch<'a, 'b, TRecord> {
 
         let execution_context = ExecutionContext::<TRecord>::new(
             self.engine.diagnostic_level.clone(),
+            &self.engine.external_function_implementations,
             self.pipeline,
             &self.global_variables,
             &self.summaries,
@@ -186,6 +202,7 @@ impl<'a, 'b, TRecord: Record + 'static> RecordSetEngineBatch<'a, 'b, TRecord> {
             self.diagnostics,
             process_summaries(
                 self.engine.diagnostic_level.clone(),
+                &self.engine.external_function_implementations,
                 &self.global_variables,
                 self.pipeline,
                 &self.summaries,
@@ -206,6 +223,7 @@ impl<'a, 'b, TRecord: Record + 'static> RecordSetEngineBatch<'a, 'b, TRecord> {
 
         let execution_context = ExecutionContext::new(
             diagnostic_level,
+            &self.engine.external_function_implementations,
             self.pipeline,
             &self.global_variables,
             &self.summaries,
@@ -298,6 +316,7 @@ fn process_record<'a, TRecord: Record + 'static>(
 
 fn process_summaries<'a>(
     diagnostic_level: RecordSetEngineDiagnosticLevel,
+    external_function_implementations: &HashMap<Box<str>, Box<dyn RecordSetEngineFunctionCallback>>,
     global_variables: &RefCell<MapValueStorage<OwnedValue>>,
     pipeline: &'a PipelineExpression,
     summaries: &Summaries<'a>,
@@ -317,6 +336,7 @@ fn process_summaries<'a>(
 
             let execution_context = ExecutionContext::new(
                 diagnostic_level.clone(),
+                external_function_implementations,
                 pipeline,
                 global_variables,
                 &summaries,
@@ -350,6 +370,7 @@ fn process_summaries<'a>(
 
             let results = process_summaries(
                 diagnostic_level.clone(),
+                external_function_implementations,
                 global_variables,
                 pipeline,
                 &summaries,