Skip to content

Commit

Permalink
Merge pull request #6043 from xudong963/null
Browse files Browse the repository at this point in the history
  • Loading branch information
@sundy-li
sundy-li authored Jun 18, 2022
2 parents 5b68287 + 82373a9 commit ea6abd9
Show file tree
Hide file tree
Showing 14 changed files with 97 additions and 48 deletions.
3 changes: 3 additions & 0 deletions query/src/pipelines/new/processors/transforms/transform_apply.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -150,19 +150,22 @@ impl OuterRefRewriter {
Scalar::AndExpr(expr) => Ok(AndExpr {
left: Box::new(self.rewrite_scalar(&expr.left)?),
right: Box::new(self.rewrite_scalar(&expr.right)?),
return_type: expr.return_type.clone(),
}
.into()),

Scalar::OrExpr(expr) => Ok(OrExpr {
left: Box::new(self.rewrite_scalar(&expr.left)?),
right: Box::new(self.rewrite_scalar(&expr.right)?),
return_type: expr.return_type.clone(),
}
.into()),

Scalar::ComparisonExpr(expr) => Ok(ComparisonExpr {
op: expr.op.clone(),
left: Box::new(self.rewrite_scalar(&expr.left)?),
right: Box::new(self.rewrite_scalar(&expr.right)?),
return_type: expr.return_type.clone(),
}
.into()),

Expand Down
10 changes: 5 additions & 5 deletions query/src/sql/exec/expression_builder.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,17 +57,17 @@ impl ExpressionBuilder {
Scalar::ConstantExpr(ConstantExpr { value, data_type }) => {
self.build_literal(value, data_type)
}
Scalar::ComparisonExpr(ComparisonExpr { op, left, right }) => {
self.build_binary_operator(left, right, op.to_func_name())
}
Scalar::ComparisonExpr(ComparisonExpr {
op, left, right, ..
}) => self.build_binary_operator(left, right, op.to_func_name()),
Scalar::AggregateFunction(AggregateFunction {
func_name,
distinct,
params,
args,
..
}) => self.build_aggr_function(func_name.clone(), *distinct, params.clone(), args),
Scalar::AndExpr(AndExpr { left, right }) => {
Scalar::AndExpr(AndExpr { left, right, .. }) => {
let left = self.build(&**left)?;
let right = self.build(&**right)?;
Ok(Expression::BinaryExpression {
Expand All @@ -76,7 +76,7 @@ impl ExpressionBuilder {
right: Box::new(right),
})
}
Scalar::OrExpr(OrExpr { left, right }) => {
Scalar::OrExpr(OrExpr { left, right, .. }) => {
let left = self.build(&**left)?;
let right = self.build(&**right)?;
Ok(Expression::BinaryExpression {
Expand Down
5 changes: 5 additions & 0 deletions query/src/sql/exec/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ use common_datavalues::ToDataType;
use common_datavalues::Vu8;
use common_exception::ErrorCode;
use common_exception::Result;
use common_functions::scalars::FunctionFactory;
use common_planners::find_aggregate_exprs;
use common_planners::find_aggregate_exprs_in_expr;
use common_planners::Expression;
Expand Down Expand Up @@ -334,9 +335,13 @@ impl PipelineBuilder {
let eb = ExpressionBuilder::create(self.metadata.clone());
let scalars = &filter.predicates;
let pred = scalars.iter().cloned().reduce(|acc, v| {
let func = FunctionFactory::instance()
.get("and", &[&acc.data_type(), &v.data_type()])
.unwrap();
AndExpr {
left: Box::new(acc),
right: Box::new(v),
return_type: func.return_type(),
}
.into()
});
Expand Down
1 change: 1 addition & 0 deletions query/src/sql/optimizer/rule/rewrite/rule_push_down_filter_join.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ impl<'a> Predicate<'a> {
op: ComparisonOp::Equal,
left,
right,
..
}) = scalar
{
if satisfied_by(left, left_prop) && satisfied_by(right, right_prop) {
Expand Down
3 changes: 3 additions & 0 deletions query/src/sql/planner/binder/aggregate.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,17 +95,20 @@ impl<'a> AggregateRewriter<'a> {
Scalar::AndExpr(scalar) => Ok(AndExpr {
left: Box::new(self.visit(&scalar.left)?),
right: Box::new(self.visit(&scalar.right)?),
return_type: scalar.return_type.clone(),
}
.into()),
Scalar::OrExpr(scalar) => Ok(OrExpr {
left: Box::new(self.visit(&scalar.left)?),
right: Box::new(self.visit(&scalar.right)?),
return_type: scalar.return_type.clone(),
}
.into()),
Scalar::ComparisonExpr(scalar) => Ok(ComparisonExpr {
op: scalar.op.clone(),
left: Box::new(self.visit(&scalar.left)?),
right: Box::new(self.visit(&scalar.right)?),
return_type: scalar.return_type.clone(),
}
.into()),
Scalar::FunctionCall(func) => {
Expand Down
10 changes: 4 additions & 6 deletions query/src/sql/planner/binder/scalar_common.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ where F: Fn(&Scalar) -> bool

pub fn split_conjunctions(scalar: &Scalar) -> Vec<Scalar> {
match scalar {
Scalar::AndExpr(AndExpr { left, right }) => {
Scalar::AndExpr(AndExpr { left, right, .. }) => {
vec![split_conjunctions(left), split_conjunctions(right)].concat()
}
_ => {
Expand All @@ -75,11 +75,9 @@ pub fn split_conjunctions(scalar: &Scalar) -> Vec<Scalar> {

pub fn split_equivalent_predicate(scalar: &Scalar) -> Option<(Scalar, Scalar)> {
match scalar {
Scalar::ComparisonExpr(ComparisonExpr { op, left, right })
if *op == ComparisonOp::Equal =>
{
Some((*left.clone(), *right.clone()))
}
Scalar::ComparisonExpr(ComparisonExpr {
op, left, right, ..
}) if *op == ComparisonOp::Equal => Some((*left.clone(), *right.clone())),
_ => None,
}
}
Expand Down
4 changes: 2 additions & 2 deletions query/src/sql/planner/binder/scalar_visitor.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,11 +61,11 @@ pub trait ScalarVisitor: Sized {
stack.push(RecursionProcessing::Call(&**left));
stack.push(RecursionProcessing::Call(&**right));
}
Scalar::AndExpr(AndExpr { left, right }) => {
Scalar::AndExpr(AndExpr { left, right, .. }) => {
stack.push(RecursionProcessing::Call(&**left));
stack.push(RecursionProcessing::Call(&**right));
}
Scalar::OrExpr(OrExpr { left, right }) => {
Scalar::OrExpr(OrExpr { left, right, .. }) => {
stack.push(RecursionProcessing::Call(&**left));
stack.push(RecursionProcessing::Call(&**right));
}
Expand Down
14 changes: 14 additions & 0 deletions query/src/sql/planner/binder/subquery.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ use common_datavalues::DataValue;
use common_exception::ErrorCode;
use common_exception::Result;
use common_functions::aggregates::AggregateFunctionFactory;
use common_functions::scalars::FunctionFactory;

use crate::sql::binder::ColumnBinding;
use crate::sql::optimizer::ColumnSet;
Expand Down Expand Up @@ -44,6 +45,7 @@ use crate::sql::plans::ScalarItem;
use crate::sql::plans::SubqueryExpr;
use crate::sql::plans::SubqueryType;
use crate::sql::MetadataRef;
use crate::sql::ScalarExpr;

/// Rewrite subquery into `Apply` operator
pub struct SubqueryRewriter {
Expand Down Expand Up @@ -197,6 +199,7 @@ impl SubqueryRewriter {
}
.into(),
),
return_type: agg_func.return_type()?,
};
let eval_scalar = EvalScalar {
items: vec![ScalarItem {
Expand Down Expand Up @@ -262,10 +265,13 @@ impl SubqueryRewriter {
Scalar::AndExpr(expr) => {
let (left, _result_left) = self.try_rewrite_subquery(&expr.left, s_expr)?;
let (right, _result_right) = self.try_rewrite_subquery(&expr.right, s_expr)?;
let func = FunctionFactory::instance()
.get("and", &[&left.data_type(), &right.data_type()])?;
Ok((
AndExpr {
left: Box::new(left),
right: Box::new(right),
return_type: func.return_type(),
}
.into(),
s_expr.clone(),
Expand All @@ -275,10 +281,13 @@ impl SubqueryRewriter {
Scalar::OrExpr(expr) => {
let (left, s_expr) = self.try_rewrite_subquery(&expr.left, s_expr)?;
let (right, s_expr) = self.try_rewrite_subquery(&expr.right, &s_expr)?;
let func = FunctionFactory::instance()
.get("or", &[&left.data_type(), &right.data_type()])?;
Ok((
OrExpr {
left: Box::new(left),
right: Box::new(right),
return_type: func.return_type(),
}
.into(),
s_expr,
Expand All @@ -288,11 +297,16 @@ impl SubqueryRewriter {
Scalar::ComparisonExpr(expr) => {
let (left, s_expr) = self.try_rewrite_subquery(&expr.left, s_expr)?;
let (right, s_expr) = self.try_rewrite_subquery(&expr.right, &s_expr)?;
let func = FunctionFactory::instance().get(expr.op.to_func_name(), &[
&left.data_type(),
&right.data_type(),
])?;
Ok((
ComparisonExpr {
op: expr.op.clone(),
left: Box::new(left),
right: Box::new(right),
return_type: func.return_type(),
}
.into(),
s_expr,
Expand Down
10 changes: 10 additions & 0 deletions query/src/sql/planner/format/display_rel_operator.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
use std::fmt::Display;

use common_datavalues::format_data_type_sql;
use common_functions::scalars::FunctionFactory;
use itertools::Itertools;

use super::FormatTreeNode;
Expand All @@ -37,6 +38,7 @@ use crate::sql::plans::RelOperator;
use crate::sql::plans::Scalar;
use crate::sql::plans::Sort;
use crate::sql::MetadataRef;
use crate::sql::ScalarExpr;

pub struct FormatContext {
metadata: MetadataRef,
Expand Down Expand Up @@ -147,18 +149,26 @@ pub fn format_logical_inner_join(
.iter()
.zip(op.right_conditions.iter())
.map(|(left, right)| {
let func = FunctionFactory::instance()
.get("=", &[&left.data_type(), &right.data_type()])
.unwrap();
ComparisonExpr {
op: ComparisonOp::Equal,
left: Box::new(left.clone()),
right: Box::new(right.clone()),
return_type: func.return_type(),
}
.into()
})
.collect();
let pred: Scalar = preds.iter().fold(preds[0].clone(), |prev, next| {
let func = FunctionFactory::instance()
.get("and", &[&prev.data_type(), &next.data_type()])
.unwrap();
Scalar::AndExpr(AndExpr {
left: Box::new(prev),
right: Box::new(next.clone()),
return_type: func.return_type(),
})
});
write!(f, "LogicalInnerJoin: {}", format_scalar(metadata, &pred))
Expand Down
9 changes: 6 additions & 3 deletions query/src/sql/planner/plans/scalar.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -310,11 +310,12 @@ impl ScalarExpr for ConstantExpr {
pub struct AndExpr {
pub left: Box<Scalar>,
pub right: Box<Scalar>,
pub return_type: DataTypeImpl,
}

impl ScalarExpr for AndExpr {
fn data_type(&self) -> DataTypeImpl {
BooleanType::new_impl()
self.return_type.clone()
}

fn used_columns(&self) -> ColumnSet {
Expand All @@ -332,11 +333,12 @@ impl ScalarExpr for AndExpr {
pub struct OrExpr {
pub left: Box<Scalar>,
pub right: Box<Scalar>,
pub return_type: DataTypeImpl,
}

impl ScalarExpr for OrExpr {
fn data_type(&self) -> DataTypeImpl {
BooleanType::new_impl()
self.return_type.clone()
}

fn used_columns(&self) -> ColumnSet {
Expand Down Expand Up @@ -406,11 +408,12 @@ pub struct ComparisonExpr {
pub op: ComparisonOp,
pub left: Box<Scalar>,
pub right: Box<Scalar>,
pub return_type: DataTypeImpl,
}

impl ScalarExpr for ComparisonExpr {
fn data_type(&self) -> DataTypeImpl {
BooleanType::new_impl()
self.return_type.clone()
}

fn used_columns(&self) -> ColumnSet {
Expand Down
3 changes: 3 additions & 0 deletions query/src/sql/planner/semantic/grouping_check.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -68,17 +68,20 @@ impl<'a> GroupingChecker<'a> {
Scalar::AndExpr(scalar) => Ok(AndExpr {
left: Box::new(self.resolve(&scalar.left)?),
right: Box::new(self.resolve(&scalar.right)?),
return_type: scalar.return_type.clone(),
}
.into()),
Scalar::OrExpr(scalar) => Ok(OrExpr {
left: Box::new(self.resolve(&scalar.left)?),
right: Box::new(self.resolve(&scalar.right)?),
return_type: scalar.return_type.clone(),
}
.into()),
Scalar::ComparisonExpr(scalar) => Ok(ComparisonExpr {
op: scalar.op.clone(),
left: Box::new(self.resolve(&scalar.left)?),
right: Box::new(self.resolve(&scalar.right)?),
return_type: scalar.return_type.clone(),
}
.into()),
Scalar::FunctionCall(func) => {
Expand Down
Loading

0 comments on commit ea6abd9

Please sign in to comment.