You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "ygf11 (via GitHub)" <gi...@apache.org> on 2023/02/13 12:14:37 UTC
[GitHub] [arrow-datafusion] ygf11 opened a new pull request, #5264: Support non-tuple expression for exists-subquery to join
ygf11 opened a new pull request, #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #4934.
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
This sql works correctly in datafusion, and will be rewrited to `LeftSemi Join`.
```sql
> explain select * from t1 where exists (select 1 from t2 where t1.t1_id = t2.t2_id);
+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| logical_plan | Projection: t1.t1_id, t1.t1_name, t1.t1_int |
| | LeftSemi Join: t1.t1_id = t2.t2_id |
| | TableScan: t1 projection=[t1_id, t1_name, t1_int] |
| | TableScan: t2 projection=[t2_id]
```
But the following sql will not:
```sql
> select * from t1 where exists (select 1 from t2 where t1.t1_id + 1 = t2.t2_id * 2);
```
These sql should also be rewrited to `LeftSemi Join`.
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
# Are these changes tested?
Yes.
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ursabot commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ursabot (via GitHub)" <gi...@apache.org>.
ursabot commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1435710025
Benchmark runs are scheduled for baseline = 5d5b1a0bd8234465d179adc0b947bb439acc91f7 and contender = 27b15fdcf3234ba050e254ec179f51e5d92e9784. 27b15fdcf3234ba050e254ec179f51e5d92e9784 is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/ec1de7ae689f4747983a442706fe604a...31ee1428c9314acaae4868d85e09ba9d/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] [test-mac-arm](https://conbench.ursa.dev/compare/runs/916eade7fdeb4acd94df3ae267fb80d4...79b7cee3c57e4b3399229b1d99e66e52/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/e04efdc339434d4dbf5f494109ecac58...8289ab572de34715b2c01c1050edeaee/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/938cd0f32dbe48fc8fcfbd51ce47c830...4b04396ae2444a55a13fa54992b153d2/)
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1109915829
##########
datafusion/optimizer/src/decorrelate_where_exists.rs:
##########
@@ -144,55 +142,68 @@ fn optimize_exists(
query_info: &SubqueryInfo,
outer_input: &LogicalPlan,
) -> Result<Option<LogicalPlan>> {
- let subqry_filter = match query_info.query.subquery.as_ref() {
+ let maybe_subqury_filter = match query_info.query.subquery.as_ref() {
LogicalPlan::Distinct(subqry_distinct) => match subqry_distinct.input.as_ref() {
- LogicalPlan::Projection(subqry_proj) => {
- Filter::try_from_plan(&subqry_proj.input)
- }
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
- // Subquery currently only supports distinct or projection
return Ok(None);
}
},
- LogicalPlan::Projection(subqry_proj) => Filter::try_from_plan(&subqry_proj.input),
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
// Subquery currently only supports distinct or projection
return Ok(None);
}
}
- .map_err(|e| context!("cannot optimize non-correlated subquery", e))?;
-
- // split into filters
- let subqry_filter_exprs = split_conjunction(&subqry_filter.predicate);
- verify_not_disjunction(&subqry_filter_exprs)?;
-
- // Grab column names to join on
- let (col_exprs, other_subqry_exprs) =
- find_join_exprs(subqry_filter_exprs, subqry_filter.input.schema())?;
- let (outer_cols, subqry_cols, join_filters) =
- exprs_to_join_cols(&col_exprs, subqry_filter.input.schema(), false)?;
- if subqry_cols.is_empty() || outer_cols.is_empty() {
- // cannot optimize non-correlated subquery
+ .as_ref();
+
+ // extract join filters
+ let (join_filters, subquery_input) = extract_join_filters(maybe_subqury_filter)?;
+ // cannot optimize non-correlated subquery
+ if join_filters.is_empty() {
return Ok(None);
}
- // build subquery side of join - the thing the subquery was querying
- let mut subqry_plan = LogicalPlanBuilder::from(subqry_filter.input.as_ref().clone());
- if let Some(expr) = conjunction(other_subqry_exprs) {
- subqry_plan = subqry_plan.filter(expr)? // if the subquery had additional expressions, restore them
- }
- let subqry_plan = subqry_plan.build()?;
+ let input_schema = subquery_input.schema();
+ let subquery_cols: BTreeSet<Column> =
+ join_filters
+ .iter()
+ .try_fold(BTreeSet::new(), |mut cols, expr| {
+ let using_cols: Vec<Column> = expr
+ .to_columns()?
+ .into_iter()
+ .filter(|col| input_schema.field_from_column(col).is_ok())
+ .collect::<_>();
+
+ cols.extend(using_cols);
+ Result::<_, DataFusionError>::Ok(cols)
+ })?;
+
+ let projection_exprs: Vec<Expr> =
+ subquery_cols.into_iter().map(Expr::Column).collect();
+
+ let right = LogicalPlanBuilder::from(subquery_input)
+ .project(projection_exprs)?
+ .build()?;
- let join_keys = (subqry_cols, outer_cols);
+ let join_filter = conjunction(join_filters);
// join our sub query into the main plan
let join_type = match query_info.negated {
true => JoinType::LeftAnti,
false => JoinType::LeftSemi,
};
+
+ // TODO: add Distinct if the original plan is a Distinct.
Review Comment:
> I will do this in the following pr if it is ok 🤣.
Absolutely it is ok! Thank you for all your work
##########
datafusion/optimizer/src/decorrelate_where_exists.rs:
##########
@@ -144,55 +142,68 @@ fn optimize_exists(
query_info: &SubqueryInfo,
outer_input: &LogicalPlan,
) -> Result<Option<LogicalPlan>> {
- let subqry_filter = match query_info.query.subquery.as_ref() {
+ let maybe_subqury_filter = match query_info.query.subquery.as_ref() {
LogicalPlan::Distinct(subqry_distinct) => match subqry_distinct.input.as_ref() {
- LogicalPlan::Projection(subqry_proj) => {
- Filter::try_from_plan(&subqry_proj.input)
- }
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
- // Subquery currently only supports distinct or projection
return Ok(None);
}
},
- LogicalPlan::Projection(subqry_proj) => Filter::try_from_plan(&subqry_proj.input),
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
// Subquery currently only supports distinct or projection
return Ok(None);
}
}
- .map_err(|e| context!("cannot optimize non-correlated subquery", e))?;
-
- // split into filters
- let subqry_filter_exprs = split_conjunction(&subqry_filter.predicate);
- verify_not_disjunction(&subqry_filter_exprs)?;
-
- // Grab column names to join on
- let (col_exprs, other_subqry_exprs) =
- find_join_exprs(subqry_filter_exprs, subqry_filter.input.schema())?;
- let (outer_cols, subqry_cols, join_filters) =
- exprs_to_join_cols(&col_exprs, subqry_filter.input.schema(), false)?;
- if subqry_cols.is_empty() || outer_cols.is_empty() {
- // cannot optimize non-correlated subquery
+ .as_ref();
+
+ // extract join filters
+ let (join_filters, subquery_input) = extract_join_filters(maybe_subqury_filter)?;
+ // cannot optimize non-correlated subquery
+ if join_filters.is_empty() {
return Ok(None);
}
- // build subquery side of join - the thing the subquery was querying
- let mut subqry_plan = LogicalPlanBuilder::from(subqry_filter.input.as_ref().clone());
- if let Some(expr) = conjunction(other_subqry_exprs) {
- subqry_plan = subqry_plan.filter(expr)? // if the subquery had additional expressions, restore them
- }
- let subqry_plan = subqry_plan.build()?;
+ let input_schema = subquery_input.schema();
+ let subquery_cols: BTreeSet<Column> =
+ join_filters
+ .iter()
+ .try_fold(BTreeSet::new(), |mut cols, expr| {
+ let using_cols: Vec<Column> = expr
+ .to_columns()?
+ .into_iter()
+ .filter(|col| input_schema.field_from_column(col).is_ok())
+ .collect::<_>();
+
+ cols.extend(using_cols);
+ Result::<_, DataFusionError>::Ok(cols)
+ })?;
+
+ let projection_exprs: Vec<Expr> =
+ subquery_cols.into_iter().map(Expr::Column).collect();
+
+ let right = LogicalPlanBuilder::from(subquery_input)
+ .project(projection_exprs)?
+ .build()?;
- let join_keys = (subqry_cols, outer_cols);
+ let join_filter = conjunction(join_filters);
// join our sub query into the main plan
let join_type = match query_info.negated {
true => JoinType::LeftAnti,
false => JoinType::LeftSemi,
};
+
+ // TODO: add Distinct if the original plan is a Distinct.
Review Comment:
> I will do this in the following pr if it is ok 🤣.
Absolutely it is ok! Thank you for all your work on this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1109779911
##########
datafusion/optimizer/src/utils.rs:
##########
@@ -468,6 +469,42 @@ pub fn merge_schema(inputs: Vec<&LogicalPlan>) -> DFSchema {
})
}
+/// Extract join predicates from the correclated subquery.
+/// The join predicate means that the expression references columns
+/// from both the subquery and outer table or only from the outer table.
+///
+/// Returns join predicates and subquery(extracted).
+/// ```
+pub(crate) fn extract_join_filters(
+ maybe_filter: &LogicalPlan,
+) -> Result<(Vec<Expr>, LogicalPlan)> {
+ if let LogicalPlan::Filter(plan_filter) = maybe_filter {
+ let input_schema = plan_filter.input.schema();
+ let subquery_filter_exprs = split_conjunction(&plan_filter.predicate);
+
+ let mut join_filters: Vec<Expr> = vec![];
+ let mut subquery_filters: Vec<Expr> = vec![];
+ for expr in subquery_filter_exprs {
+ let cols = expr.to_columns()?;
+ if check_all_column_from_schema(&cols, input_schema.clone()) {
+ subquery_filters.push(expr.clone());
Review Comment:
I think we need clone here.
`subquery_filter_exprs` is the result of `split_conjunction` function, and the type is `Vec<&Expr>`.
https://github.com/apache/arrow-datafusion/blob/fed4019d556f4afb3156fd12c21608e08b8d7eb6/datafusion/optimizer/src/utils.rs#L65-L67
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1431514368
I will try and find time to review this PR over the next day or two
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1429711292
Please take a look @alamb @jackwener.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1109785623
##########
datafusion/optimizer/src/decorrelate_where_exists.rs:
##########
@@ -144,55 +142,68 @@ fn optimize_exists(
query_info: &SubqueryInfo,
outer_input: &LogicalPlan,
) -> Result<Option<LogicalPlan>> {
- let subqry_filter = match query_info.query.subquery.as_ref() {
+ let maybe_subqury_filter = match query_info.query.subquery.as_ref() {
LogicalPlan::Distinct(subqry_distinct) => match subqry_distinct.input.as_ref() {
- LogicalPlan::Projection(subqry_proj) => {
- Filter::try_from_plan(&subqry_proj.input)
- }
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
- // Subquery currently only supports distinct or projection
return Ok(None);
}
},
- LogicalPlan::Projection(subqry_proj) => Filter::try_from_plan(&subqry_proj.input),
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
// Subquery currently only supports distinct or projection
return Ok(None);
}
}
- .map_err(|e| context!("cannot optimize non-correlated subquery", e))?;
-
- // split into filters
- let subqry_filter_exprs = split_conjunction(&subqry_filter.predicate);
- verify_not_disjunction(&subqry_filter_exprs)?;
-
- // Grab column names to join on
- let (col_exprs, other_subqry_exprs) =
- find_join_exprs(subqry_filter_exprs, subqry_filter.input.schema())?;
- let (outer_cols, subqry_cols, join_filters) =
- exprs_to_join_cols(&col_exprs, subqry_filter.input.schema(), false)?;
- if subqry_cols.is_empty() || outer_cols.is_empty() {
- // cannot optimize non-correlated subquery
+ .as_ref();
+
+ // extract join filters
+ let (join_filters, subquery_input) = extract_join_filters(maybe_subqury_filter)?;
+ // cannot optimize non-correlated subquery
+ if join_filters.is_empty() {
return Ok(None);
}
- // build subquery side of join - the thing the subquery was querying
- let mut subqry_plan = LogicalPlanBuilder::from(subqry_filter.input.as_ref().clone());
- if let Some(expr) = conjunction(other_subqry_exprs) {
- subqry_plan = subqry_plan.filter(expr)? // if the subquery had additional expressions, restore them
- }
- let subqry_plan = subqry_plan.build()?;
+ let input_schema = subquery_input.schema();
+ let subquery_cols: BTreeSet<Column> =
+ join_filters
+ .iter()
+ .try_fold(BTreeSet::new(), |mut cols, expr| {
+ let using_cols: Vec<Column> = expr
+ .to_columns()?
+ .into_iter()
+ .filter(|col| input_schema.field_from_column(col).is_ok())
+ .collect::<_>();
+
+ cols.extend(using_cols);
+ Result::<_, DataFusionError>::Ok(cols)
+ })?;
+
+ let projection_exprs: Vec<Expr> =
+ subquery_cols.into_iter().map(Expr::Column).collect();
+
+ let right = LogicalPlanBuilder::from(subquery_input)
+ .project(projection_exprs)?
+ .build()?;
- let join_keys = (subqry_cols, outer_cols);
+ let join_filter = conjunction(join_filters);
// join our sub query into the main plan
let join_type = match query_info.negated {
true => JoinType::LeftAnti,
false => JoinType::LeftSemi,
};
+
+ // TODO: add Distinct if the original plan is a Distinct.
Review Comment:
Yes, it is still a todo, some distinct cases need more checks. like:
```rust
// SELECT t1.t1_id,
// t1.t1_name,
// t1.t1_int
// FROM t1
// WHERE EXISTS(SELECT DISTINCT t2_int
// FROM t2
// WHERE t2.t2_id > t1.t1_id);
// if we just add back the `DISTINCT`, the result:
Projection: t1.t1_id, t1.t1_name, t1.t1_int
LeftSemi Join: Filter: t2.t2_int > t1.t1_int
TableScan: t1
Distinct:
Projection: t2.t2_int
TableScan: t2
// expected result:
Projection: t1.t1_id, t1.t1_name, t1.t1_int
LeftSemi Join: Filter: t2.t2_int > t1.t1_int
TableScan: t1
Distinct:
Projection: t2.t2_int, t2.t2_id
TableScan: t2
```
`t2_int` will not be in the projection.
The reason is we just consider the columns from join filter as the project items, we should also consider columns from original projection when there is an outer distinct.
I will do this in the following pr if it is ok :rofl:.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] jackwener commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "jackwener (via GitHub)" <gi...@apache.org>.
jackwener commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1437862020
> @ygf11 @jackwener I have a feeling that maybe we should combine the two rules `DecorrelateWhereExists` and `DecorrelateWhereIn` to one rule. How do you think ?
Agree it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1105781873
##########
datafusion/optimizer/src/decorrelate_where_exists.rs:
##########
@@ -241,13 +252,14 @@ mod tests {
.project(vec![col("customer.c_custkey")])?
.build()?;
- let expected = r#"Projection: customer.c_custkey [c_custkey:Int64]
- LeftSemi Join: customer.c_custkey = orders.o_custkey [c_custkey:Int64, c_name:Utf8]
- LeftSemi Join: customer.c_custkey = orders.o_custkey [c_custkey:Int64, c_name:Utf8]
- TableScan: customer [c_custkey:Int64, c_name:Utf8]
- TableScan: orders [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N]
- TableScan: orders [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N]"#;
-
+ let expected = "Projection: customer.c_custkey [c_custkey:Int64]\
+ \n LeftSemi Join: Filter: orders.o_custkey = customer.c_custkey [c_custkey:Int64, c_name:Utf8]\
+ \n LeftSemi Join: Filter: orders.o_custkey = customer.c_custkey [c_custkey:Int64, c_name:Utf8]\
+ \n TableScan: customer [c_custkey:Int64, c_name:Utf8]\
+ \n Projection: orders.o_custkey [o_custkey:Int64]\
+ \n TableScan: orders [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N]\
+ \n Projection: orders.o_custkey [o_custkey:Int64]\
+ \n TableScan: orders [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N]";
Review Comment:
Before this pr, this rule will not keep the projection.
I think keeping projection make it easy to read, and spark also keeps it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb merged pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1105786637
##########
datafusion/optimizer/src/decorrelate_where_exists.rs:
##########
@@ -305,21 +318,21 @@ mod tests {
.project(vec![col("customer.c_custkey")])?
.build()?;
- let expected = r#"Projection: customer.c_custkey [c_custkey:Int64]
- LeftSemi Join: customer.c_custkey = orders.o_custkey [c_custkey:Int64, c_name:Utf8]
- TableScan: customer [c_custkey:Int64, c_name:Utf8]
- Filter: orders.o_orderkey = Int32(1) [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N]
- TableScan: orders [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N]"#;
+ let expected = "Projection: customer.c_custkey [c_custkey:Int64]\
+ \n LeftSemi Join: Filter: customer.c_custkey = orders.o_custkey [c_custkey:Int64, c_name:Utf8]\
+ \n TableScan: customer [c_custkey:Int64, c_name:Utf8]\
+ \n Projection: orders.o_custkey [o_custkey:Int64]\
+ \n Filter: orders.o_orderkey = Int32(1) [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N]\
+ \n TableScan: orders [o_orderkey:Int64, o_custkey:Int64, o_orderstatus:Utf8, o_totalprice:Float64;N]";
assert_plan_eq(&plan, expected)
}
- /// Test for correlated exists subquery with no columns in schema
#[test]
fn exists_subquery_no_cols() -> Result<()> {
let sq = Arc::new(
LogicalPlanBuilder::from(scan_tpch_table("orders"))
- .filter(col("customer.c_custkey").eq(col("customer.c_custkey")))?
+ .filter(col("customer.c_custkey").eq(lit(1u32)))?
.project(vec![col("orders.o_custkey")])?
Review Comment:
`customer.c_custkey = customer.c_custkey` is too specific here, so modify it to `customer.c_custkey = 1`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1105790317
##########
benchmarks/expected-plans/q21.txt:
##########
@@ -14,8 +14,10 @@ Sort: numwait DESC NULLS FIRST, supplier.s_name ASC NULLS LAST
TableScan: orders projection=[o_orderkey, o_orderstatus]
Filter: nation.n_name = Utf8("SAUDI ARABIA")
TableScan: nation projection=[n_nationkey, n_name]
- SubqueryAlias: l2
- TableScan: lineitem projection=[l_orderkey, l_suppkey]
- SubqueryAlias: l3
- Filter: lineitem.l_receiptdate > lineitem.l_commitdate
- TableScan: lineitem projection=[l_orderkey, l_suppkey, l_commitdate, l_receiptdate]
\ No newline at end of file
+ Projection: l2.l_orderkey, l2.l_suppkey
+ SubqueryAlias: l2
+ TableScan: lineitem projection=[l_orderkey, l_suppkey]
+ Projection: l3.l_orderkey, l3.l_suppkey
+ SubqueryAlias: l3
+ Filter: lineitem.l_receiptdate > lineitem.l_commitdate
+ TableScan: lineitem projection=[l_orderkey, l_suppkey, l_commitdate, l_receiptdate]
Review Comment:
The difference is we keep the projection now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1105792126
##########
datafusion/core/tests/sql/joins.rs:
##########
@@ -2187,7 +2187,6 @@ async fn left_anti_join() -> Result<()> {
}
#[tokio::test]
-#[ignore = "Test ignored, will be enabled after fixing the anti join plan bug"]
// https://github.com/apache/arrow-datafusion/issues/4366
Review Comment:
This test passes now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1110987431
##########
datafusion/core/tests/sql/joins.rs:
##########
@@ -2187,7 +2187,6 @@ async fn left_anti_join() -> Result<()> {
}
#[tokio::test]
-#[ignore = "Test ignored, will be enabled after fixing the anti join plan bug"]
// https://github.com/apache/arrow-datafusion/issues/4366
Review Comment:
Linked issue #4366 to this pr.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] alamb commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1108896625
##########
datafusion/optimizer/src/utils.rs:
##########
@@ -468,6 +469,42 @@ pub fn merge_schema(inputs: Vec<&LogicalPlan>) -> DFSchema {
})
}
+/// Extract join predicates from the correclated subquery.
+/// The join predicate means that the expression references columns
+/// from both the subquery and outer table or only from the outer table.
+///
+/// Returns join predicates and subquery(extracted).
+/// ```
+pub(crate) fn extract_join_filters(
+ maybe_filter: &LogicalPlan,
+) -> Result<(Vec<Expr>, LogicalPlan)> {
+ if let LogicalPlan::Filter(plan_filter) = maybe_filter {
+ let input_schema = plan_filter.input.schema();
+ let subquery_filter_exprs = split_conjunction(&plan_filter.predicate);
+
+ let mut join_filters: Vec<Expr> = vec![];
+ let mut subquery_filters: Vec<Expr> = vec![];
+ for expr in subquery_filter_exprs {
+ let cols = expr.to_columns()?;
+ if check_all_column_from_schema(&cols, input_schema.clone()) {
+ subquery_filters.push(expr.clone());
Review Comment:
I wonder if you need to clone the expr here. It seems like it is owned already it could be used directly here and below
Though I see this code was just refactored into a different module
##########
datafusion/optimizer/src/decorrelate_where_exists.rs:
##########
@@ -144,55 +142,68 @@ fn optimize_exists(
query_info: &SubqueryInfo,
outer_input: &LogicalPlan,
) -> Result<Option<LogicalPlan>> {
- let subqry_filter = match query_info.query.subquery.as_ref() {
+ let maybe_subqury_filter = match query_info.query.subquery.as_ref() {
LogicalPlan::Distinct(subqry_distinct) => match subqry_distinct.input.as_ref() {
- LogicalPlan::Projection(subqry_proj) => {
- Filter::try_from_plan(&subqry_proj.input)
- }
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
- // Subquery currently only supports distinct or projection
return Ok(None);
}
},
- LogicalPlan::Projection(subqry_proj) => Filter::try_from_plan(&subqry_proj.input),
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
// Subquery currently only supports distinct or projection
return Ok(None);
}
}
- .map_err(|e| context!("cannot optimize non-correlated subquery", e))?;
-
- // split into filters
- let subqry_filter_exprs = split_conjunction(&subqry_filter.predicate);
- verify_not_disjunction(&subqry_filter_exprs)?;
-
- // Grab column names to join on
- let (col_exprs, other_subqry_exprs) =
- find_join_exprs(subqry_filter_exprs, subqry_filter.input.schema())?;
- let (outer_cols, subqry_cols, join_filters) =
- exprs_to_join_cols(&col_exprs, subqry_filter.input.schema(), false)?;
- if subqry_cols.is_empty() || outer_cols.is_empty() {
- // cannot optimize non-correlated subquery
+ .as_ref();
+
+ // extract join filters
+ let (join_filters, subquery_input) = extract_join_filters(maybe_subqury_filter)?;
+ // cannot optimize non-correlated subquery
+ if join_filters.is_empty() {
return Ok(None);
}
- // build subquery side of join - the thing the subquery was querying
- let mut subqry_plan = LogicalPlanBuilder::from(subqry_filter.input.as_ref().clone());
- if let Some(expr) = conjunction(other_subqry_exprs) {
- subqry_plan = subqry_plan.filter(expr)? // if the subquery had additional expressions, restore them
- }
- let subqry_plan = subqry_plan.build()?;
+ let input_schema = subquery_input.schema();
+ let subquery_cols: BTreeSet<Column> =
+ join_filters
+ .iter()
+ .try_fold(BTreeSet::new(), |mut cols, expr| {
+ let using_cols: Vec<Column> = expr
+ .to_columns()?
+ .into_iter()
+ .filter(|col| input_schema.field_from_column(col).is_ok())
+ .collect::<_>();
+
+ cols.extend(using_cols);
+ Result::<_, DataFusionError>::Ok(cols)
+ })?;
+
+ let projection_exprs: Vec<Expr> =
+ subquery_cols.into_iter().map(Expr::Column).collect();
+
+ let right = LogicalPlanBuilder::from(subquery_input)
+ .project(projection_exprs)?
+ .build()?;
- let join_keys = (subqry_cols, outer_cols);
+ let join_filter = conjunction(join_filters);
// join our sub query into the main plan
let join_type = match query_info.negated {
true => JoinType::LeftAnti,
false => JoinType::LeftSemi,
};
+
+ // TODO: add Distinct if the original plan is a Distinct.
Review Comment:
is this still a todo? I see the code above looking into Distinct children, but the distinct is not added back
##########
datafusion/optimizer/tests/integration-test.rs:
##########
@@ -121,8 +121,9 @@ fn semi_join_with_join_filter() -> Result<()> {
let expected = "Projection: test.col_utf8\
\n LeftSemi Join: test.col_int32 = t2.col_int32 Filter: test.col_uint32 != t2.col_uint32\
\n TableScan: test projection=[col_int32, col_uint32, col_utf8]\
- \n SubqueryAlias: t2\
- \n TableScan: test projection=[col_int32, col_uint32, col_utf8]";
+ \n Projection: t2.col_int32, t2.col_uint32\
+ \n SubqueryAlias: t2\
+ \n TableScan: test projection=[col_int32, col_uint32]";
Review Comment:
this is a nice improvement too -- the unused column `col_utf8` is filtered out
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on a diff in pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on code in PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#discussion_r1109785623
##########
datafusion/optimizer/src/decorrelate_where_exists.rs:
##########
@@ -144,55 +142,68 @@ fn optimize_exists(
query_info: &SubqueryInfo,
outer_input: &LogicalPlan,
) -> Result<Option<LogicalPlan>> {
- let subqry_filter = match query_info.query.subquery.as_ref() {
+ let maybe_subqury_filter = match query_info.query.subquery.as_ref() {
LogicalPlan::Distinct(subqry_distinct) => match subqry_distinct.input.as_ref() {
- LogicalPlan::Projection(subqry_proj) => {
- Filter::try_from_plan(&subqry_proj.input)
- }
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
- // Subquery currently only supports distinct or projection
return Ok(None);
}
},
- LogicalPlan::Projection(subqry_proj) => Filter::try_from_plan(&subqry_proj.input),
+ LogicalPlan::Projection(subqry_proj) => &subqry_proj.input,
_ => {
// Subquery currently only supports distinct or projection
return Ok(None);
}
}
- .map_err(|e| context!("cannot optimize non-correlated subquery", e))?;
-
- // split into filters
- let subqry_filter_exprs = split_conjunction(&subqry_filter.predicate);
- verify_not_disjunction(&subqry_filter_exprs)?;
-
- // Grab column names to join on
- let (col_exprs, other_subqry_exprs) =
- find_join_exprs(subqry_filter_exprs, subqry_filter.input.schema())?;
- let (outer_cols, subqry_cols, join_filters) =
- exprs_to_join_cols(&col_exprs, subqry_filter.input.schema(), false)?;
- if subqry_cols.is_empty() || outer_cols.is_empty() {
- // cannot optimize non-correlated subquery
+ .as_ref();
+
+ // extract join filters
+ let (join_filters, subquery_input) = extract_join_filters(maybe_subqury_filter)?;
+ // cannot optimize non-correlated subquery
+ if join_filters.is_empty() {
return Ok(None);
}
- // build subquery side of join - the thing the subquery was querying
- let mut subqry_plan = LogicalPlanBuilder::from(subqry_filter.input.as_ref().clone());
- if let Some(expr) = conjunction(other_subqry_exprs) {
- subqry_plan = subqry_plan.filter(expr)? // if the subquery had additional expressions, restore them
- }
- let subqry_plan = subqry_plan.build()?;
+ let input_schema = subquery_input.schema();
+ let subquery_cols: BTreeSet<Column> =
+ join_filters
+ .iter()
+ .try_fold(BTreeSet::new(), |mut cols, expr| {
+ let using_cols: Vec<Column> = expr
+ .to_columns()?
+ .into_iter()
+ .filter(|col| input_schema.field_from_column(col).is_ok())
+ .collect::<_>();
+
+ cols.extend(using_cols);
+ Result::<_, DataFusionError>::Ok(cols)
+ })?;
+
+ let projection_exprs: Vec<Expr> =
+ subquery_cols.into_iter().map(Expr::Column).collect();
+
+ let right = LogicalPlanBuilder::from(subquery_input)
+ .project(projection_exprs)?
+ .build()?;
- let join_keys = (subqry_cols, outer_cols);
+ let join_filter = conjunction(join_filters);
// join our sub query into the main plan
let join_type = match query_info.negated {
true => JoinType::LeftAnti,
false => JoinType::LeftSemi,
};
+
+ // TODO: add Distinct if the original plan is a Distinct.
Review Comment:
Yes, it is still a todo, some distinct cases need more checks. like:
```rust
// SELECT t1.t1_id,
// t1.t1_name,
// t1.t1_int
// FROM t1
// WHERE EXISTS(SELECT DISTINCT t2_int
// FROM t2
// WHERE t2.t2_id > t1.t1_id);
// if we just add back the `DISTINCT`, the result:
Projection: t1.t1_id, t1.t1_name, t1.t1_int
LeftSemi Join: Filter: t2.t2_int > t1.t1_int
TableScan: t1
Distinct:
Projection: t2.t2_int
TableScan: t2
// expected result:
Projection: t1.t1_id, t1.t1_name, t1.t1_int
LeftSemi Join: Filter: t2.t2_int > t1.t1_int
TableScan: t1
Distinct:
Projection: t2.t2_int, t2.t2_id
TableScan: t2
```
`t2_id` will not be in the projection.
The reason is we just consider the columns from join filter as the project items, we should also consider columns from original projection when there is an outer distinct.
I will do this in the following pr if it is ok :rofl:.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "mingmwang (via GitHub)" <gi...@apache.org>.
mingmwang commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1436503204
@ygf11
Thanks for taking care of this. I didn't get chance to provide a fix for in/exists subqueries. I will take a closer look at your PRs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "mingmwang (via GitHub)" <gi...@apache.org>.
mingmwang commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1436638329
@ygf11 @jackwener
I have a feeling that maybe we should combine the two rules `DecorrelateWhereExists` and `DecorrelateWhereIn` to one rule. How do you think ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1436902566
> I have a feeling that maybe we should combine the two rules DecorrelateWhereExists and DecorrelateWhereIn to one rule. How do you think ?
Agree, most logics are same, we should combine them. I can take this task if others do not work on :smile:.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "mingmwang (via GitHub)" <gi...@apache.org>.
mingmwang commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1437079370
BTW, this PR looks really nice and fixed couple of issues related to the Exist subquery.
Thanks you @ygf11.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] mingmwang commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "mingmwang (via GitHub)" <gi...@apache.org>.
mingmwang commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1437070399
One more thing, looks like the SubqueryAlias logic is not consistent between the two rules `DecorrelateWhereExists` and `DecorrelateWhereIn`
https://github.com/apache/arrow-datafusion/pull/4767
I think you can make them consistent in the refactoring work.
```rust
let sql = "SELECT t1_id, t1_name FROM t1 WHERE t1_id IN (SELECT t2_id FROM t2) ORDER BY t1_id";
let msg = format!("Creating logical plan for '{sql}'");
let dataframe = ctx.sql(&("explain ".to_owned() + sql)).await.expect(&msg);
let plan = dataframe.into_optimized_plan()?;
let expected = vec![
"Explain [plan_type:Utf8, plan:Utf8]",
" Sort: t1.t1_id ASC NULLS LAST [t1_id:UInt32;N, t1_name:Utf8;N]",
" Projection: t1.t1_id, t1.t1_name [t1_id:UInt32;N, t1_name:Utf8;N]",
" LeftSemi Join: t1.t1_id = __correlated_sq_1.t2_id [t1_id:UInt32;N, t1_name:Utf8;N]",
" TableScan: t1 projection=[t1_id, t1_name] [t1_id:UInt32;N, t1_name:Utf8;N]",
" SubqueryAlias: __correlated_sq_1 [t2_id:UInt32;N]",
" Projection: t2.t2_id AS t2_id [t2_id:UInt32;N]",
" TableScan: t2 projection=[t2_id] [t2_id:UInt32;N]",
];
```
```rust
let sql = "SELECT t1_id, t1_name FROM t1 WHERE NOT EXISTS (SELECT 1 FROM t2 WHERE t1_id = t2_id and t1_id > 11) ORDER BY t1_id";
let msg = format!("Creating logical plan for '{sql}'");
let dataframe = ctx.sql(&("explain ".to_owned() + sql)).await.expect(&msg);
let plan = dataframe.into_optimized_plan()?;
let expected = vec![
"Explain [plan_type:Utf8, plan:Utf8]",
" Sort: t1.t1_id ASC NULLS LAST [t1_id:UInt32;N, t1_name:Utf8;N]",
" Projection: t1.t1_id, t1.t1_name [t1_id:UInt32;N, t1_name:Utf8;N]",
" LeftAnti Join: t1.t1_id = t2.t2_id Filter: t1.t1_id > UInt32(11) [t1_id:UInt32;N, t1_name:Utf8;N]",
" TableScan: t1 projection=[t1_id, t1_name] [t1_id:UInt32;N, t1_name:Utf8;N]",
" Projection: t2.t2_id [t2_id:UInt32;N]",
" TableScan: t2 projection=[t2_id] [t2_id:UInt32;N]",
];
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-datafusion] ygf11 commented on pull request #5264: Support non-tuple expression for exists-subquery to join
Posted by "ygf11 (via GitHub)" <gi...@apache.org>.
ygf11 commented on PR #5264:
URL: https://github.com/apache/arrow-datafusion/pull/5264#issuecomment-1435650735
Thanks for reviewing @alamb. I fixed the conflict.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org