use crate::constants::SystemTables;
use crate::engine::objects::types::BaseSqlTypes;
use crate::engine::objects::{ConstraintMapper, SqlTuple};
use super::io::{ConstraintManager, ConstraintManagerError};
use super::objects::types::SqlTypeDefinition;
use super::objects::{ParseTree, Plan, PlannedStatement, SqlTupleError, Table};
use super::transactions::TransactionId;
use async_stream::try_stream;
use futures::stream::Stream;
use std::convert::TryFrom;
use std::num::TryFromIntError;
use std::pin::Pin;
use std::sync::Arc;
use thiserror::Error;
use uuid::Uuid;
#[derive(Clone)]
pub struct Executor {
cons_man: ConstraintManager,
}
impl Executor {
pub fn new(cons_man: ConstraintManager) -> Executor {
Executor { cons_man }
}
pub fn execute(
self,
tran_id: TransactionId,
plan_tree: PlannedStatement,
) -> Pin<Box<dyn Stream<Item = Result<SqlTuple, ExecutorError>> + Send>> {
self.execute_plans(tran_id, plan_tree.plan)
}
fn execute_plans(
self,
tran_id: TransactionId,
plan: Arc<Plan>,
) -> Pin<Box<dyn Stream<Item = Result<SqlTuple, ExecutorError>> + Send>> {
match plan.as_ref() {
Plan::CartesianJoin(cp) => {
self.cartesian_join(tran_id, cp.left.clone(), cp.right.clone())
}
Plan::FullTableScan(fts) => {
self.full_table_scan(tran_id, fts.src_table.clone(), fts.target_type.clone())
}
Plan::ModifyTable(mt) => self.modify_table(tran_id, &mt.table, mt.source.clone()),
Plan::StaticData(sd) => self.static_data(sd.clone()),
}
}
fn cartesian_join(
self,
tran_id: TransactionId,
left: Arc<Plan>,
right: Arc<Plan>,
) -> Pin<Box<impl Stream<Item = Result<SqlTuple, ExecutorError>>>> {
let s = try_stream! {
for await left_data in self.clone().execute_plans(tran_id, left) {
let left_data = left_data?;
for await right_data in self.clone().execute_plans(tran_id, right.clone()) {
let right_data = right_data?;
yield SqlTuple::merge(&left_data, &right_data);
}
}
};
Box::pin(s)
}
fn full_table_scan(
self,
tran_id: TransactionId,
src_table: Arc<Table>,
target_type: Arc<SqlTypeDefinition>,
) -> Pin<Box<impl Stream<Item = Result<SqlTuple, ExecutorError>>>> {
let s = try_stream! {
let vis = self.cons_man;
for await row in vis.get_stream(tran_id, src_table.clone()) {
let data = row?.user_data.clone();
let requested_row = data.filter_map(&src_table.sql_type, &target_type)?;
yield requested_row;
}
};
Box::pin(s)
}
fn modify_table(
self,
tran_id: TransactionId,
table: &Arc<Table>,
source: Arc<Plan>,
) -> Pin<Box<impl Stream<Item = Result<SqlTuple, ExecutorError>>>> {
let vis = self.clone().cons_man;
let table = table.clone();
let s = try_stream! {
for await val in self.clone().execute_plans(tran_id, source) {
let unwrapped_val = val?;
vis.clone()
.insert_row(tran_id, &table, unwrapped_val.clone())
.await?;
yield unwrapped_val;
}
};
Box::pin(s)
}
fn static_data(
self,
rows: Arc<Vec<SqlTuple>>,
) -> Pin<Box<impl Stream<Item = Result<SqlTuple, ExecutorError>>>> {
let s = try_stream! {
for row in rows.as_ref().iter() {
yield row.clone();
}
};
Box::pin(s)
}
pub async fn execute_utility(
&mut self,
tran_id: TransactionId,
parse_tree: ParseTree,
) -> Result<Vec<SqlTuple>, ExecutorError> {
let mut cm = self.cons_man.clone();
let create_table = match parse_tree {
ParseTree::CreateTable(t) => t,
_ => return Err(ExecutorError::NotUtility()),
};
let table_id = Uuid::new_v4();
let pg_class = SystemTables::PgClass.value();
let table_row = SqlTuple(vec![
Some(BaseSqlTypes::Uuid(table_id)),
Some(BaseSqlTypes::Text(create_table.table_name.clone())),
]);
cm.insert_row(tran_id, &pg_class, table_row).await?;
let mut primary_key_cols = vec![];
let pg_attribute = SystemTables::PgAttribute.value();
for i in 0..create_table.provided_columns.len() {
let cm = self.cons_man.clone();
let i_u32 = u32::try_from(i).map_err(ExecutorError::ConversionError)?;
let table_row = SqlTuple(vec![
Some(BaseSqlTypes::Uuid(table_id)),
Some(BaseSqlTypes::Text(
create_table.provided_columns[i].name.clone(),
)),
Some(BaseSqlTypes::Text(
create_table.provided_columns[i].sql_type.clone(),
)),
Some(BaseSqlTypes::Integer(i_u32)),
Some(BaseSqlTypes::Bool(create_table.provided_columns[i].null)),
]);
cm.clone()
.insert_row(tran_id, &pg_attribute, table_row)
.await?;
if create_table.provided_columns[i].primary_key {
primary_key_cols.push(BaseSqlTypes::Integer(i_u32));
}
}
if !primary_key_cols.is_empty() {
let pk_id = Uuid::new_v4();
let primary_key_index = SqlTuple(vec![
Some(BaseSqlTypes::Uuid(pk_id)),
Some(BaseSqlTypes::Uuid(table_id)),
Some(BaseSqlTypes::Text(format!(
"{}_primary_key_index",
create_table.table_name
))),
Some(BaseSqlTypes::Array(primary_key_cols)),
Some(BaseSqlTypes::Bool(true)),
]);
let pg_index = SystemTables::PgIndex.value();
self.cons_man
.clone()
.insert_row(tran_id, &pg_index, primary_key_index)
.await?;
let primary_key_constraint = SqlTuple(vec![
Some(BaseSqlTypes::Uuid(Uuid::new_v4())),
Some(BaseSqlTypes::Uuid(table_id)),
Some(BaseSqlTypes::Uuid(pk_id)),
Some(BaseSqlTypes::Text(format!(
"{}_primary_key",
create_table.table_name
))),
Some(BaseSqlTypes::Text(ConstraintMapper::PrimaryKey.to_string())),
]);
let pg_constraint = SystemTables::PgConstraint.value();
self.cons_man
.clone()
.insert_row(tran_id, &pg_constraint, primary_key_constraint)
.await?;
}
Ok(vec![])
}
}
#[derive(Debug, Error)]
pub enum ExecutorError {
#[error("Not a utility statement")]
NotUtility(),
#[error(transparent)]
SqlTupleError(#[from] SqlTupleError),
#[error(transparent)]
ConstraintManagerError(#[from] ConstraintManagerError),
#[error("Unable to convert usize to u32")]
ConversionError(#[from] TryFromIntError),
#[error("Recursive Plans Not Allowed")]
RecursionNotAllowed(),
#[error("Unknown")]
Unknown(),
}