Skip to content

refactor(binder): make flatten_plan return explicit derived-column mappings #19696

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
forsaken628 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

refactor(binder): make flatten_plan return explicit derived-column mappings #19696

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion src/query/sql/src/planner/binder/bind_table_reference/bind_join.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -394,7 +394,7 @@ impl Binder {
// If there are outer columns in right child, then the join is a correlated lateral join
let opt_ctx = OptimizerContext::new(self.ctx.clone(), self.metadata.clone());
let mut decorrelator = SubqueryDecorrelatorOptimizer::new(opt_ctx, Some(self.clone()));
right_child = decorrelator.flatten_plan(
let (flatten_plan, derived_columns) = decorrelator.flatten_plan(
&left_child,
&right_child,
&right_prop.outer_columns,
Expand All @@ -403,10 +403,12 @@ impl Binder {
},
false,
)?;
right_child = flatten_plan;
let original_num_conditions = left_conditions.len();
decorrelator.add_equi_conditions(
None,
&right_prop.outer_columns,
&derived_columns,
&mut right_conditions,
&mut left_conditions,
)?;
Expand Down
99 changes: 55 additions & 44 deletions src/query/sql/src/planner/optimizer/optimizers/operator/decorrelate/decorrelate.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ use databend_common_expression::types::DataType;
use databend_common_expression::types::NumberScalar;
use databend_common_functions::BUILTIN_FUNCTIONS;

use super::DerivedColumnScope;
use crate::ColumnSet;
use crate::binder::ColumnBindingBuilder;
use crate::binder::JoinPredicate;
Expand Down Expand Up @@ -244,17 +245,17 @@ impl SubqueryDecorrelatorOptimizer {
Ok(Some(result))
}

pub fn try_decorrelate_subquery(
pub(crate) fn try_decorrelate_subquery(
&mut self,
outer: &SExpr,
subquery: &SubqueryExpr,
flatten_info: &mut FlattenInfo,
is_conjunctive_predicate: bool,
) -> Result<(SExpr, UnnestResult)> {
) -> Result<(SExpr, UnnestResult, DerivedColumnScope)> {
match subquery.typ {
SubqueryType::Scalar => {
let correlated_columns = &subquery.outer_columns;
let flatten_plan = self.flatten_plan(
let (flatten_plan, derived_columns) = self.flatten_plan(
outer,
&subquery.subquery,
correlated_columns,
Expand All @@ -267,6 +268,7 @@ impl SubqueryDecorrelatorOptimizer {
self.add_equi_conditions(
subquery.span,
correlated_columns,
&derived_columns,
&mut right_conditions,
&mut left_conditions,
)?;
Expand Down Expand Up @@ -304,16 +306,20 @@ impl SubqueryDecorrelatorOptimizer {
Arc::new(outer.clone()),
Arc::new(flatten_plan),
);
Ok((s_expr, UnnestResult::SingleJoin))
Ok((s_expr, UnnestResult::SingleJoin, derived_columns))
}
SubqueryType::Exists | SubqueryType::NotExists => {
if is_conjunctive_predicate {
if let Some(result) = self.try_decorrelate_simple_subquery(outer, subquery)? {
return Ok((result, UnnestResult::SimpleJoin { output_index: None }));
}
if is_conjunctive_predicate
&& let Some(result) = self.try_decorrelate_simple_subquery(outer, subquery)?
{
return Ok((
result,
UnnestResult::SimpleJoin { output_index: None },
Default::default(),
));
}
let correlated_columns = &subquery.outer_columns;
let flatten_plan = self.flatten_plan(
let (flatten_plan, derived_columns) = self.flatten_plan(
outer,
&subquery.subquery,
correlated_columns,
Expand All @@ -326,6 +332,7 @@ impl SubqueryDecorrelatorOptimizer {
self.add_equi_conditions(
subquery.span,
correlated_columns,
&derived_columns,
&mut left_conditions,
&mut right_conditions,
)?;
Expand Down Expand Up @@ -368,11 +375,15 @@ impl SubqueryDecorrelatorOptimizer {
Arc::new(outer.clone()),
Arc::new(flatten_plan),
);
Ok((s_expr, UnnestResult::MarkJoin { marker_index }))
Ok((
s_expr,
UnnestResult::MarkJoin { marker_index },
derived_columns,
))
}
SubqueryType::Any => {
let correlated_columns = &subquery.outer_columns;
let flatten_plan = self.flatten_plan(
let (flatten_plan, derived_columns) = self.flatten_plan(
outer,
&subquery.subquery,
correlated_columns,
Expand All @@ -384,6 +395,7 @@ impl SubqueryDecorrelatorOptimizer {
self.add_equi_conditions(
subquery.span,
correlated_columns,
&derived_columns,
&mut left_conditions,
&mut right_conditions,
)?;
Expand Down Expand Up @@ -453,16 +465,18 @@ impl SubqueryDecorrelatorOptimizer {
Arc::new(flatten_plan),
),
UnnestResult::MarkJoin { marker_index },
derived_columns,
))
}
_ => unreachable!(),
}
}

pub fn add_equi_conditions(
pub(crate) fn add_equi_conditions(
&self,
span: Span,
correlated_columns: &ColumnSet,
derived_columns: &DerivedColumnScope,
left_conditions: &mut Vec<ScalarExpr>,
right_conditions: &mut Vec<ScalarExpr>,
) -> Result<()> {
Expand All @@ -482,15 +496,15 @@ impl SubqueryDecorrelatorOptimizer {
.table_index(column_entry.table_index())
.build(),
});
let Some(derive_column) = self.derived_columns.get(&correlated_column) else {
let Some(derive_column) = derived_columns.resolve(correlated_column) else {
continue;
};
let column_entry = metadata.column(*derive_column);
let column_entry = metadata.column(derive_column);
let left_column = ScalarExpr::BoundColumnRef(BoundColumnRef {
span,
column: ColumnBindingBuilder::new(
column_entry.name(),
*derive_column,
derive_column,
Box::from(column_entry.data_type()),
Visibility::Visible,
)
Expand All @@ -507,43 +521,40 @@ impl SubqueryDecorrelatorOptimizer {
// If correlated_columns only occur in equi-conditions, such as `where t1.a = t.a and t1.b = t.b`(t1 is outer table)
// Then we won't join outer and inner table.
pub(crate) fn join_outer_inner_table(
&mut self,
&self,
filter: &Filter,
correlated_columns: &ColumnSet,
) -> Result<bool> {
Ok(!filter.predicates.iter().all(|predicate| {
) -> Result<(bool, DerivedColumnScope)> {
let mut derived_columns = DerivedColumnScope::default();
let can_reuse_inner_columns = filter.predicates.iter().all(|predicate| {
if predicate
.used_columns()
.iter()
.any(|column| correlated_columns.contains(column))
.all(|column| !correlated_columns.contains(column))
{
if let ScalarExpr::FunctionCall(func) = predicate {
if func.func_name == "eq" {
if let (
ScalarExpr::BoundColumnRef(left),
ScalarExpr::BoundColumnRef(right),
) = (&func.arguments[0], &func.arguments[1])
{
if correlated_columns.contains(&left.column.index)
&& !correlated_columns.contains(&right.column.index)
{
self.derived_columns
.insert(left.column.index, right.column.index);
}
if !correlated_columns.contains(&left.column.index)
&& correlated_columns.contains(&right.column.index)
{
self.derived_columns
.insert(right.column.index, left.column.index);
}
return true;
}
}
return true;
}
if let ScalarExpr::FunctionCall(func) = predicate
&& func.func_name == "eq"
&& let (ScalarExpr::BoundColumnRef(left), ScalarExpr::BoundColumnRef(right)) =
(&func.arguments[0], &func.arguments[1])
{
if correlated_columns.contains(&left.column.index)
&& !correlated_columns.contains(&right.column.index)
{
derived_columns.record(left.column.index, right.column.index);
}
if !correlated_columns.contains(&left.column.index)
&& correlated_columns.contains(&right.column.index)
{
derived_columns.record(right.column.index, left.column.index);
}
return false;
true
} else {
false
}
true
}))
});
Ok((!can_reuse_inner_columns, derived_columns))
}

// Try folding the subquery into a constant value expression,
Expand Down
Loading
Loading