use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use datafusion::common::{DataFusionError, Result};
use datafusion::prelude::SessionContext;
use crate::common::LixTimestamp;
use crate::functions::FunctionProviderHandle;
#[derive(Default)]
pub(crate) struct ExecutionSlots {
values: Mutex<ExecutionSlotValues>,
}
#[derive(Default)]
struct ExecutionSlotValues {
active_account_id: Option<String>,
active_branch_id: Option<String>,
active_branch_commit_id: Option<String>,
functions: Option<FunctionProviderHandle>,
current_timestamp: Option<LixTimestamp>,
}
impl std::fmt::Debug for ExecutionSlots {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter
.debug_struct("ExecutionSlots")
.finish_non_exhaustive()
}
}
impl ExecutionSlots {
pub(crate) fn bind(
&self,
functions: FunctionProviderHandle,
active_account_id: &str,
active_branch_id: Option<&str>,
active_branch_commit_id: Option<&str>,
) {
let mut values = self.lock();
assign(&mut values.active_account_id, Some(active_account_id));
assign(&mut values.active_branch_id, active_branch_id);
assign(&mut values.active_branch_commit_id, active_branch_commit_id);
values.functions = Some(functions);
values.current_timestamp = None;
}
pub(crate) fn active_account_id(&self) -> Option<String> {
self.lock().active_account_id.clone()
}
pub(crate) fn active_branch_id(&self) -> Option<String> {
self.lock().active_branch_id.clone()
}
pub(crate) fn active_branch_commit_id(&self) -> Option<String> {
self.lock().active_branch_commit_id.clone()
}
pub(crate) fn functions(&self) -> Result<FunctionProviderHandle> {
self.lock().functions.clone().ok_or_else(|| {
DataFusionError::Internal(
"Lix SQL execution functions were invoked on an unbound session".to_string(),
)
})
}
pub(crate) fn current_timestamp(&self) -> Result<LixTimestamp> {
let mut values = self.lock();
if let Some(timestamp) = values.current_timestamp {
return Ok(timestamp);
}
let functions = values.functions.clone().ok_or_else(|| {
DataFusionError::Internal(
"Lix SQL execution functions were invoked on an unbound session".to_string(),
)
})?;
let timestamp = functions.call_timestamp();
values.current_timestamp = Some(timestamp);
Ok(timestamp)
}
fn lock(&self) -> MutexGuard<'_, ExecutionSlotValues> {
self.values.lock().unwrap_or_else(PoisonError::into_inner)
}
}
fn assign(slot: &mut Option<String>, value: Option<&str>) {
match (slot.as_mut(), value) {
(Some(existing), Some(value)) => {
if existing != value {
existing.clear();
existing.push_str(value);
}
}
(_, Some(value)) => *slot = Some(value.to_string()),
(_, None) => *slot = None,
}
}
pub(crate) fn execution_slots(session: &SessionContext) -> Arc<ExecutionSlots> {
session
.state_ref()
.read()
.config()
.get_extension::<ExecutionSlots>()
.expect("Lix SQL sessions are created with execution-function slots")
}