Skip to content
This repository has been archived by the owner on Sep 28, 2021. It is now read-only.

Commit

Permalink
Merge 62fcd43 into 39f0b5a
Browse files Browse the repository at this point in the history
  • Loading branch information
@alex-dukhno
alex-dukhno committed May 18, 2021
2 parents 39f0b5a + 62fcd43 commit fca6f15
Show file tree
Hide file tree
Showing 32 changed files with 765 additions and 1,365 deletions.
132 changes: 58 additions & 74 deletions node_engine/src/transaction_manager.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,8 +14,8 @@

use catalog::CatalogHandler;
use data_manipulation::{
DynamicTypedTree, QueryPlan, StaticTypedTree, TypedDeleteQuery, TypedInsertQuery, TypedQuery, TypedSelectQuery,
TypedUpdateQuery, UntypedInsertQuery, UntypedQuery, UntypedUpdateQuery,
QueryPlan, TypedDeleteQuery, TypedInsertQuery, TypedQuery, TypedSelectQuery, TypedTree, TypedUpdateQuery,
UntypedInsertQuery, UntypedQuery, UntypedUpdateQuery,
};
use definition::ColumnDef;
use definition_planner::DefinitionPlanner;
Expand Down Expand Up @@ -112,36 +112,28 @@ impl<'t> TransactionContext<'t> {
) -> Result<TypedQuery, QueryError> {
match untyped_query {
UntypedQuery::Insert(insert) => {
let type_checked = insert
let type_coerced = insert
.values
.into_iter()
.map(|values| {
values
.into_iter()
.map(|value| value.map(|v| self.type_inference.infer_static(v, &param_types)))
.collect::<Vec<Option<StaticTypedTree>>>()
.map(|value| value.map(|v| self.type_inference.infer_type(v, &param_types)))
.collect::<Vec<Option<TypedTree>>>()
})
.map(|values| {
values
.into_iter()
.map(|value| value.map(|v| self.type_checker.check_static(v)))
.map(|value| value.map(|v| self.type_checker.type_check(v)))
.collect::<Vec<Option<TypedTree>>>()
})
.map(|values| {
values
.into_iter()
.map(|value| value.map(|c| self.type_coercion.coerce(c)))
.collect()
})
.collect::<Vec<Vec<Option<StaticTypedTree>>>>();
let table_info = self
.catalog
.table_definition(insert.full_table_name.clone())
.unwrap()
.unwrap();
let table_columns = table_info.columns();
let mut type_coerced = vec![];
for checked in type_checked {
let mut row = vec![];
for (index, c) in checked.into_iter().enumerate() {
row.push(c.map(|c| self.type_coercion.coerce_static(c, table_columns[index].sql_type())));
}
type_coerced.push(row);
}
.collect::<Vec<Vec<Option<TypedTree>>>>();
Ok(TypedQuery::Insert(TypedInsertQuery {
full_table_name: insert.full_table_name,
values: type_coerced,
Expand All @@ -151,20 +143,20 @@ impl<'t> TransactionContext<'t> {
let typed_values = select
.projection_items
.into_iter()
.map(|value| self.type_inference.infer_dynamic(value, &param_types));
.map(|value| self.type_inference.infer_type(value, &param_types));
let type_checked_values = typed_values
.into_iter()
.map(|value| self.type_checker.check_dynamic(value));
.map(|value| self.type_checker.type_check(value));
let type_coerced_values = type_checked_values
.into_iter()
.map(|value| self.type_coercion.coerce_dynamic(value))
.collect::<Vec<DynamicTypedTree>>();
.map(|value| self.type_coercion.coerce(value))
.collect::<Vec<TypedTree>>();

let typed_filter = select
.filter
.map(|value| self.type_inference.infer_dynamic(value, &param_types));
let type_checked_filter = typed_filter.map(|value| self.type_checker.check_dynamic(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce_dynamic(value));
.map(|value| self.type_inference.infer_type(value, &param_types));
let type_checked_filter = typed_filter.map(|value| self.type_checker.type_check(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce(value));

Ok(TypedQuery::Select(TypedSelectQuery {
projection_items: type_coerced_values,
Expand All @@ -176,20 +168,20 @@ impl<'t> TransactionContext<'t> {
let typed_values = update
.assignments
.into_iter()
.map(|value| value.map(|value| self.type_inference.infer_dynamic(value, &param_types)));
.map(|value| value.map(|value| self.type_inference.infer_type(value, &param_types)));
let type_checked = typed_values
.into_iter()
.map(|value| value.map(|value| self.type_checker.check_dynamic(value)));
.map(|value| value.map(|value| self.type_checker.type_check(value)));
let type_coerced = type_checked
.into_iter()
.map(|value| value.map(|value| self.type_coercion.coerce_dynamic(value)))
.collect::<Vec<Option<DynamicTypedTree>>>();
.map(|value| value.map(|value| self.type_coercion.coerce(value)))
.collect::<Vec<Option<TypedTree>>>();

let typed_filter = update
.filter
.map(|value| self.type_inference.infer_dynamic(value, &param_types));
let type_checked_filter = typed_filter.map(|value| self.type_checker.check_dynamic(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce_dynamic(value));
.map(|value| self.type_inference.infer_type(value, &param_types));
let type_checked_filter = typed_filter.map(|value| self.type_checker.type_check(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce(value));

Ok(TypedQuery::Update(TypedUpdateQuery {
full_table_name: update.full_table_name,
Expand All @@ -200,9 +192,9 @@ impl<'t> TransactionContext<'t> {
UntypedQuery::Delete(delete) => {
let typed_filter = delete
.filter
.map(|value| self.type_inference.infer_dynamic(value, &param_types));
let type_checked_filter = typed_filter.map(|value| self.type_checker.check_dynamic(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce_dynamic(value));
.map(|value| self.type_inference.infer_type(value, &param_types));
let type_checked_filter = typed_filter.map(|value| self.type_checker.type_check(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce(value));

Ok(TypedQuery::Delete(TypedDeleteQuery {
full_table_name: delete.full_table_name,
Expand All @@ -222,36 +214,28 @@ impl<'t> TransactionContext<'t> {
pub fn process(&self, query: Query, param_types: Vec<SqlTypeFamily>) -> Result<TypedQuery, QueryError> {
match self.query_analyzer.analyze(query)? {
UntypedQuery::Insert(insert) => {
let type_checked = insert
let type_coerced = insert
.values
.into_iter()
.map(|values| {
values
.into_iter()
.map(|value| value.map(|v| self.type_inference.infer_static(v, &param_types)))
.collect::<Vec<Option<StaticTypedTree>>>()
.map(|value| value.map(|v| self.type_inference.infer_type(v, &param_types)))
.collect::<Vec<Option<TypedTree>>>()
})
.map(|values| {
values
.into_iter()
.map(|value| value.map(|v| self.type_checker.type_check(v)))
.collect::<Vec<Option<TypedTree>>>()
})
.map(|values| {
values
.into_iter()
.map(|value| value.map(|v| self.type_checker.check_static(v)))
.map(|value| value.map(|v| self.type_coercion.coerce(v)))
.collect()
})
.collect::<Vec<Vec<Option<StaticTypedTree>>>>();
let table_info = self
.catalog
.table_definition(insert.full_table_name.clone())
.unwrap()
.unwrap();
let table_columns = table_info.columns();
let mut type_coerced = vec![];
for checked in type_checked {
let mut row = vec![];
for (index, c) in checked.into_iter().enumerate() {
row.push(c.map(|c| self.type_coercion.coerce_static(c, table_columns[index].sql_type())));
}
type_coerced.push(row);
}
.collect::<Vec<Vec<Option<TypedTree>>>>();
Ok(TypedQuery::Insert(TypedInsertQuery {
full_table_name: insert.full_table_name,
values: type_coerced,
Expand All @@ -261,18 +245,18 @@ impl<'t> TransactionContext<'t> {
let typed_values = select
.projection_items
.into_iter()
.map(|value| self.type_inference.infer_dynamic(value, &[]));
.map(|value| self.type_inference.infer_type(value, &[]));
let type_checked_values = typed_values
.into_iter()
.map(|value| self.type_checker.check_dynamic(value));
.map(|value| self.type_checker.type_check(value));
let type_coerced_values = type_checked_values
.into_iter()
.map(|value| self.type_coercion.coerce_dynamic(value))
.collect::<Vec<DynamicTypedTree>>();
.map(|value| self.type_coercion.coerce(value))
.collect::<Vec<TypedTree>>();

let typed_filter = select.filter.map(|value| self.type_inference.infer_dynamic(value, &[]));
let type_checked_filter = typed_filter.map(|value| self.type_checker.check_dynamic(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce_dynamic(value));
let typed_filter = select.filter.map(|value| self.type_inference.infer_type(value, &[]));
let type_checked_filter = typed_filter.map(|value| self.type_checker.type_check(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce(value));

Ok(TypedQuery::Select(TypedSelectQuery {
projection_items: type_coerced_values,
Expand All @@ -284,18 +268,18 @@ impl<'t> TransactionContext<'t> {
let typed_values = update
.assignments
.into_iter()
.map(|value| value.map(|value| self.type_inference.infer_dynamic(value, &[])));
.map(|value| value.map(|value| self.type_inference.infer_type(value, &[])));
let type_checked = typed_values
.into_iter()
.map(|value| value.map(|value| self.type_checker.check_dynamic(value)));
.map(|value| value.map(|value| self.type_checker.type_check(value)));
let type_coerced = type_checked
.into_iter()
.map(|value| value.map(|value| self.type_coercion.coerce_dynamic(value)))
.collect::<Vec<Option<DynamicTypedTree>>>();
.map(|value| value.map(|value| self.type_coercion.coerce(value)))
.collect::<Vec<Option<TypedTree>>>();

let typed_filter = update.filter.map(|value| self.type_inference.infer_dynamic(value, &[]));
let type_checked_filter = typed_filter.map(|value| self.type_checker.check_dynamic(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce_dynamic(value));
let typed_filter = update.filter.map(|value| self.type_inference.infer_type(value, &[]));
let type_checked_filter = typed_filter.map(|value| self.type_checker.type_check(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce(value));

Ok(TypedQuery::Update(TypedUpdateQuery {
full_table_name: update.full_table_name,
Expand All @@ -304,9 +288,9 @@ impl<'t> TransactionContext<'t> {
}))
}
UntypedQuery::Delete(delete) => {
let typed_filter = delete.filter.map(|value| self.type_inference.infer_dynamic(value, &[]));
let type_checked_filter = typed_filter.map(|value| self.type_checker.check_dynamic(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce_dynamic(value));
let typed_filter = delete.filter.map(|value| self.type_inference.infer_type(value, &[]));
let type_checked_filter = typed_filter.map(|value| self.type_checker.type_check(value));
let type_coerced_filter = type_checked_filter.map(|value| self.type_coercion.coerce(value));

Ok(TypedQuery::Delete(TypedDeleteQuery {
full_table_name: delete.full_table_name,
Expand Down
28 changes: 14 additions & 14 deletions sql_engine/data_manipulation/query_plan/src/lib.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@
use bigdecimal::{BigDecimal, FromPrimitive};
use binary::BinaryValue;
use data_manipulation_query_result::QueryExecutionError;
use data_manipulation_typed_tree::{DynamicTypedTree, StaticTypedTree};
use data_manipulation_typed_tree::TypedTree;
use definition::ColumnDef;
use query_response::QueryEvent;
use scalar::ScalarValue;
Expand Down Expand Up @@ -87,28 +87,28 @@ pub trait Flow {
fn next_tuple(&mut self, param_values: &[ScalarValue]) -> Result<Option<Self::Output>, QueryExecutionError>;
}

pub struct StaticValues(Box<dyn Iterator<Item = Vec<Option<StaticTypedTree>>>>);
pub struct StaticValues(Box<dyn Iterator<Item = Vec<Option<TypedTree>>>>);

impl StaticValues {
pub fn new(values: Vec<Vec<Option<StaticTypedTree>>>) -> Box<StaticValues> {
pub fn new(values: Vec<Vec<Option<TypedTree>>>) -> Box<StaticValues> {
Box::new(StaticValues(Box::new(values.into_iter())))
}
}

impl Flow for StaticValues {
type Output = Vec<Option<StaticTypedTree>>;
type Output = Vec<Option<TypedTree>>;

fn next_tuple(&mut self, _param_values: &[ScalarValue]) -> Result<Option<Self::Output>, QueryExecutionError> {
Ok(self.0.next())
}
}

pub struct StaticExpressionEval {
source: Box<dyn Flow<Output = Vec<Option<StaticTypedTree>>>>,
source: Box<dyn Flow<Output = Vec<Option<TypedTree>>>>,
}

impl StaticExpressionEval {
pub fn new(source: Box<dyn Flow<Output = Vec<Option<StaticTypedTree>>>>) -> Box<StaticExpressionEval> {
pub fn new(source: Box<dyn Flow<Output = Vec<Option<TypedTree>>>>) -> Box<StaticExpressionEval> {
Box::new(StaticExpressionEval { source })
}
}
Expand All @@ -122,7 +122,7 @@ impl Flow for StaticExpressionEval {
for value in tuple {
let typed_value = match value {
None => None,
Some(value) => match value.eval(param_values) {
Some(value) => match value.eval(param_values, &[]) {
Err(error) => return Err(error),
Ok(value) => Some(value),
},
Expand Down Expand Up @@ -248,13 +248,13 @@ impl InsertQueryPlan {

pub struct Filter {
source: Box<dyn Flow<Output = (Vec<ScalarValue>, Vec<ScalarValue>)>>,
predicate: Option<DynamicTypedTree>,
predicate: Option<TypedTree>,
}

impl Filter {
pub fn new(
source: Box<dyn Flow<Output = (Vec<ScalarValue>, Vec<ScalarValue>)>>,
predicate: Option<DynamicTypedTree>,
predicate: Option<TypedTree>,
) -> Box<Filter> {
Box::new(Filter { source, predicate })
}
Expand Down Expand Up @@ -400,31 +400,31 @@ impl DeleteQueryPlan {
}

pub struct Repeater {
source: Vec<Option<DynamicTypedTree>>,
source: Vec<Option<TypedTree>>,
}

impl Repeater {
pub fn new(source: Vec<Option<DynamicTypedTree>>) -> Box<Repeater> {
pub fn new(source: Vec<Option<TypedTree>>) -> Box<Repeater> {
Box::new(Repeater { source })
}
}

impl Flow for Repeater {
type Output = Vec<Option<DynamicTypedTree>>;
type Output = Vec<Option<TypedTree>>;

fn next_tuple(&mut self, _param_values: &[ScalarValue]) -> Result<Option<Self::Output>, QueryExecutionError> {
Ok(Some(self.source.clone()))
}
}

pub struct DynamicValues {
source: Box<dyn Flow<Output = Vec<Option<DynamicTypedTree>>>>,
source: Box<dyn Flow<Output = Vec<Option<TypedTree>>>>,
records: Box<dyn Flow<Output = (Vec<ScalarValue>, Vec<ScalarValue>)>>,
}

impl DynamicValues {
pub fn new(
source: Box<dyn Flow<Output = Vec<Option<DynamicTypedTree>>>>,
source: Box<dyn Flow<Output = Vec<Option<TypedTree>>>>,
records: Box<dyn Flow<Output = (Vec<ScalarValue>, Vec<ScalarValue>)>>,
) -> Box<DynamicValues> {
Box::new(DynamicValues { source, records })
Expand Down
14 changes: 7 additions & 7 deletions sql_engine/data_manipulation/typed_queries/src/lib.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,33 +12,33 @@
// See the License for the specific language governing permissions and
// limitations under the License.

use data_manipulation_typed_tree::{DynamicTypedTree, StaticTypedTree};
use data_manipulation_typed_tree::TypedTree;
use definition::FullTableName;

#[derive(Debug, PartialEq, Clone)]
pub struct TypedInsertQuery {
pub full_table_name: FullTableName,
pub values: Vec<Vec<Option<StaticTypedTree>>>,
pub values: Vec<Vec<Option<TypedTree>>>,
}

#[derive(Debug, PartialEq, Clone)]
pub struct TypedDeleteQuery {
pub full_table_name: FullTableName,
pub filter: Option<DynamicTypedTree>,
pub filter: Option<TypedTree>,
}

#[derive(Debug, PartialEq, Clone)]
pub struct TypedUpdateQuery {
pub full_table_name: FullTableName,
pub assignments: Vec<Option<DynamicTypedTree>>,
pub filter: Option<DynamicTypedTree>,
pub assignments: Vec<Option<TypedTree>>,
pub filter: Option<TypedTree>,
}

#[derive(Debug, PartialEq, Clone)]
pub struct TypedSelectQuery {
pub full_table_name: FullTableName,
pub projection_items: Vec<DynamicTypedTree>,
pub filter: Option<DynamicTypedTree>,
pub projection_items: Vec<TypedTree>,
pub filter: Option<TypedTree>,
}

#[derive(Debug, PartialEq, Clone)]
Expand Down

0 comments on commit fca6f15

Please sign in to comment.