You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by dh...@apache.org on 2024/10/18 21:00:28 UTC
(datafusion) branch main updated: Remove logical cross join in planning (#12985)
This is an automated email from the ASF dual-hosted git repository.
dheres pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 34bd8237d2 Remove logical cross join in planning (#12985)
34bd8237d2 is described below
commit 34bd8237d2189eca5b560c034d15e63d97a15fa0
Author: Daniël Heres <da...@gmail.com>
AuthorDate: Fri Oct 18 23:00:24 2024 +0200
Remove logical cross join in planning (#12985)
* Remove logical cross join in planning
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* WIP
* Implement some more substrait pieces
* Update datafusion/core/src/physical_planner.rs
Co-authored-by: Oleks V <co...@users.noreply.github.com>
* Remove incorrect comment
---------
Co-authored-by: Oleks V <co...@users.noreply.github.com>
---
datafusion/core/src/physical_planner.rs | 22 ++++----
datafusion/expr/src/logical_plan/builder.rs | 11 ++--
datafusion/expr/src/logical_plan/plan.rs | 6 +++
datafusion/optimizer/src/eliminate_cross_join.rs | 25 +++++----
datafusion/optimizer/src/eliminate_join.rs | 26 +--------
datafusion/optimizer/src/push_down_filter.rs | 4 +-
datafusion/optimizer/src/push_down_limit.rs | 7 ++-
datafusion/sql/src/relation/join.rs | 4 +-
datafusion/sql/tests/cases/plan_to_sql.rs | 2 +-
datafusion/sql/tests/sql_integration.rs | 30 +++++------
datafusion/sqllogictest/test_files/cte.slt | 2 +-
datafusion/sqllogictest/test_files/group_by.slt | 2 +-
datafusion/sqllogictest/test_files/join.slt | 4 +-
datafusion/sqllogictest/test_files/joins.slt | 2 +-
datafusion/sqllogictest/test_files/select.slt | 2 +-
datafusion/sqllogictest/test_files/update.slt | 4 +-
datafusion/substrait/src/logical_plan/consumer.rs | 12 ++++-
.../substrait/tests/cases/consumer_integration.rs | 62 +++++++++++-----------
18 files changed, 117 insertions(+), 110 deletions(-)
diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs
index a4dffd3d02..918ebccbeb 100644
--- a/datafusion/core/src/physical_planner.rs
+++ b/datafusion/core/src/physical_planner.rs
@@ -78,7 +78,7 @@ use datafusion_expr::expr::{
use datafusion_expr::expr_rewriter::unnormalize_cols;
use datafusion_expr::logical_plan::builder::wrap_projection_for_join_if_necessary;
use datafusion_expr::{
- DescribeTable, DmlStatement, Extension, Filter, RecursiveQuery, SortExpr,
+ DescribeTable, DmlStatement, Extension, Filter, JoinType, RecursiveQuery, SortExpr,
StringifiedPlan, WindowFrame, WindowFrameBound, WriteOp,
};
use datafusion_physical_expr::aggregate::{AggregateExprBuilder, AggregateFunctionExpr};
@@ -1045,14 +1045,18 @@ impl DefaultPhysicalPlanner {
session_state.config_options().optimizer.prefer_hash_join;
let join: Arc<dyn ExecutionPlan> = if join_on.is_empty() {
- // there is no equal join condition, use the nested loop join
- // TODO optimize the plan, and use the config of `target_partitions` and `repartition_joins`
- Arc::new(NestedLoopJoinExec::try_new(
- physical_left,
- physical_right,
- join_filter,
- join_type,
- )?)
+ if join_filter.is_none() && matches!(join_type, JoinType::Inner) {
+ // cross join if there is no join conditions and no join filter set
+ Arc::new(CrossJoinExec::new(physical_left, physical_right))
+ } else {
+ // there is no equal join condition, use the nested loop join
+ Arc::new(NestedLoopJoinExec::try_new(
+ physical_left,
+ physical_right,
+ join_filter,
+ join_type,
+ )?)
+ }
} else if session_state.config().target_partitions() > 1
&& session_state.config().repartition_joins()
&& !prefer_hash_join
diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs
index da2a96327c..6ab50440ec 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -30,8 +30,8 @@ use crate::expr_rewriter::{
rewrite_sort_cols_by_aggs,
};
use crate::logical_plan::{
- Aggregate, Analyze, CrossJoin, Distinct, DistinctOn, EmptyRelation, Explain, Filter,
- Join, JoinConstraint, JoinType, Limit, LogicalPlan, Partitioning, PlanType, Prepare,
+ Aggregate, Analyze, Distinct, DistinctOn, EmptyRelation, Explain, Filter, Join,
+ JoinConstraint, JoinType, Limit, LogicalPlan, Partitioning, PlanType, Prepare,
Projection, Repartition, Sort, SubqueryAlias, TableScan, Union, Unnest, Values,
Window,
};
@@ -950,9 +950,14 @@ impl LogicalPlanBuilder {
pub fn cross_join(self, right: LogicalPlan) -> Result<Self> {
let join_schema =
build_join_schema(self.plan.schema(), right.schema(), &JoinType::Inner)?;
- Ok(Self::new(LogicalPlan::CrossJoin(CrossJoin {
+ Ok(Self::new(LogicalPlan::Join(Join {
left: self.plan,
right: Arc::new(right),
+ on: vec![],
+ filter: None,
+ join_type: JoinType::Inner,
+ join_constraint: JoinConstraint::On,
+ null_equals_null: false,
schema: DFSchemaRef::new(join_schema),
})))
}
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index 9bd57d2212..10a99c9e78 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -222,6 +222,7 @@ pub enum LogicalPlan {
Join(Join),
/// Apply Cross Join to two logical plans.
/// This is used to implement SQL `CROSS JOIN`
+ /// Deprecated: use [LogicalPlan::Join] instead with empty `on` / no filter
CrossJoin(CrossJoin),
/// Repartitions the input based on a partitioning scheme. This is
/// used to add parallelism and is sometimes referred to as an
@@ -1873,6 +1874,11 @@ impl LogicalPlan {
.as_ref()
.map(|expr| format!(" Filter: {expr}"))
.unwrap_or_else(|| "".to_string());
+ let join_type = if filter.is_none() && keys.is_empty() && matches!(join_type, JoinType::Inner) {
+ "Cross".to_string()
+ } else {
+ join_type.to_string()
+ };
match join_constraint {
JoinConstraint::On => {
write!(
diff --git a/datafusion/optimizer/src/eliminate_cross_join.rs b/datafusion/optimizer/src/eliminate_cross_join.rs
index 550728ddd3..bce5c77ca6 100644
--- a/datafusion/optimizer/src/eliminate_cross_join.rs
+++ b/datafusion/optimizer/src/eliminate_cross_join.rs
@@ -25,7 +25,7 @@ use datafusion_common::tree_node::{Transformed, TreeNode};
use datafusion_common::{internal_err, Result};
use datafusion_expr::expr::{BinaryExpr, Expr};
use datafusion_expr::logical_plan::{
- CrossJoin, Filter, Join, JoinConstraint, JoinType, LogicalPlan, Projection,
+ Filter, Join, JoinConstraint, JoinType, LogicalPlan, Projection,
};
use datafusion_expr::utils::{can_hash, find_valid_equijoin_key_pair};
use datafusion_expr::{build_join_schema, ExprSchemable, Operator};
@@ -51,7 +51,7 @@ impl EliminateCrossJoin {
/// Looks like this:
/// ```text
/// Filter(a.x = b.y AND b.xx = 100)
-/// CrossJoin
+/// Cross Join
/// TableScan a
/// TableScan b
/// ```
@@ -351,10 +351,15 @@ fn find_inner_join(
&JoinType::Inner,
)?);
- Ok(LogicalPlan::CrossJoin(CrossJoin {
+ Ok(LogicalPlan::Join(Join {
left: Arc::new(left_input),
right: Arc::new(right),
schema: join_schema,
+ on: vec![],
+ filter: None,
+ join_type: JoinType::Inner,
+ join_constraint: JoinConstraint::On,
+ null_equals_null: false,
}))
}
@@ -513,7 +518,7 @@ mod tests {
let expected = vec![
"Filter: t1.a = t2.a OR t2.b = t1.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
- " CrossJoin: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
+ " Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]",
];
@@ -601,7 +606,7 @@ mod tests {
let expected = vec![
"Filter: t1.a = t2.a AND t2.c < UInt32(15) OR t1.b = t2.b AND t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
- " CrossJoin: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
+ " Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]",
];
@@ -627,7 +632,7 @@ mod tests {
let expected = vec![
"Filter: t1.a = t2.a AND t2.c < UInt32(15) OR t1.a = t2.a OR t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
- " CrossJoin: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
+ " Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]",
];
@@ -843,7 +848,7 @@ mod tests {
let expected = vec![
"Filter: t3.a = t1.a AND t4.c < UInt32(15) OR t3.a = t1.a OR t4.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
- " CrossJoin: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
+ " Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" Filter: t2.c < UInt32(15) OR t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" Inner Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]",
@@ -924,7 +929,7 @@ mod tests {
" TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]",
" Filter: t3.a = t4.a AND t4.c < UInt32(15) OR t3.a = t4.a AND t3.c = UInt32(688) OR t3.a = t4.a OR t3.b = t4.b [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
- " CrossJoin: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
+ " Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t4 [a:UInt32, b:UInt32, c:UInt32]",
];
@@ -999,7 +1004,7 @@ mod tests {
"Filter: t4.c < UInt32(15) OR t4.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" Inner Join: t1.a = t3.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" Filter: t1.a = t2.a OR t2.c < UInt32(15) OR t1.a = t2.a AND t2.c = UInt32(688) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
- " CrossJoin: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
+ " Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]",
" Filter: t4.c < UInt32(15) OR t3.c = UInt32(688) OR t3.b = t4.b [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
@@ -1238,7 +1243,7 @@ mod tests {
let expected = vec![
"Filter: t1.a + UInt32(100) = t2.a * UInt32(2) OR t2.b = t1.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
- " CrossJoin: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
+ " Cross Join: [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]",
" TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]",
];
diff --git a/datafusion/optimizer/src/eliminate_join.rs b/datafusion/optimizer/src/eliminate_join.rs
index f9b79e036f..789235595d 100644
--- a/datafusion/optimizer/src/eliminate_join.rs
+++ b/datafusion/optimizer/src/eliminate_join.rs
@@ -23,7 +23,7 @@ use datafusion_common::{Result, ScalarValue};
use datafusion_expr::JoinType::Inner;
use datafusion_expr::{
logical_plan::{EmptyRelation, LogicalPlan},
- CrossJoin, Expr,
+ Expr,
};
/// Eliminates joins when join condition is false.
@@ -54,13 +54,6 @@ impl OptimizerRule for EliminateJoin {
match plan {
LogicalPlan::Join(join) if join.join_type == Inner && join.on.is_empty() => {
match join.filter {
- Some(Expr::Literal(ScalarValue::Boolean(Some(true)))) => {
- Ok(Transformed::yes(LogicalPlan::CrossJoin(CrossJoin {
- left: join.left,
- right: join.right,
- schema: join.schema,
- })))
- }
Some(Expr::Literal(ScalarValue::Boolean(Some(false)))) => Ok(
Transformed::yes(LogicalPlan::EmptyRelation(EmptyRelation {
produce_one_row: false,
@@ -105,21 +98,4 @@ mod tests {
let expected = "EmptyRelation";
assert_optimized_plan_equal(plan, expected)
}
-
- #[test]
- fn join_on_true() -> Result<()> {
- let plan = LogicalPlanBuilder::empty(false)
- .join_on(
- LogicalPlanBuilder::empty(false).build()?,
- Inner,
- Some(lit(true)),
- )?
- .build()?;
-
- let expected = "\
- CrossJoin:\
- \n EmptyRelation\
- \n EmptyRelation";
- assert_optimized_plan_equal(plan, expected)
- }
}
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
index 6e2cc0cbdb..2e3bca5b0b 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -1727,7 +1727,7 @@ mod tests {
.build()?;
let expected = "Projection: test.a, test1.d\
- \n CrossJoin:\
+ \n Cross Join: \
\n Projection: test.a, test.b, test.c\
\n TableScan: test, full_filters=[test.a = Int32(1)]\
\n Projection: test1.d, test1.e, test1.f\
@@ -1754,7 +1754,7 @@ mod tests {
.build()?;
let expected = "Projection: test.a, test1.a\
- \n CrossJoin:\
+ \n Cross Join: \
\n Projection: test.a, test.b, test.c\
\n TableScan: test, full_filters=[test.a = Int32(1)]\
\n Projection: test1.a, test1.b, test1.c\
diff --git a/datafusion/optimizer/src/push_down_limit.rs b/datafusion/optimizer/src/push_down_limit.rs
index 47fce64ae0..6ed7738704 100644
--- a/datafusion/optimizer/src/push_down_limit.rs
+++ b/datafusion/optimizer/src/push_down_limit.rs
@@ -254,10 +254,9 @@ fn push_down_join(mut join: Join, limit: usize) -> Transformed<Join> {
let (left_limit, right_limit) = if is_no_join_condition(&join) {
match join.join_type {
- Left | Right | Full => (Some(limit), Some(limit)),
+ Left | Right | Full | Inner => (Some(limit), Some(limit)),
LeftAnti | LeftSemi => (Some(limit), None),
RightAnti | RightSemi => (None, Some(limit)),
- Inner => (None, None),
}
} else {
match join.join_type {
@@ -1116,7 +1115,7 @@ mod test {
.build()?;
let expected = "Limit: skip=0, fetch=1000\
- \n CrossJoin:\
+ \n Cross Join: \
\n Limit: skip=0, fetch=1000\
\n TableScan: test, fetch=1000\
\n Limit: skip=0, fetch=1000\
@@ -1136,7 +1135,7 @@ mod test {
.build()?;
let expected = "Limit: skip=1000, fetch=1000\
- \n CrossJoin:\
+ \n Cross Join: \
\n Limit: skip=0, fetch=2000\
\n TableScan: test, fetch=2000\
\n Limit: skip=0, fetch=2000\
diff --git a/datafusion/sql/src/relation/join.rs b/datafusion/sql/src/relation/join.rs
index 409533a3ea..3f34608e37 100644
--- a/datafusion/sql/src/relation/join.rs
+++ b/datafusion/sql/src/relation/join.rs
@@ -151,7 +151,9 @@ impl<'a, S: ContextProvider> SqlToRel<'a, S> {
.build()
}
}
- JoinConstraint::None => not_impl_err!("NONE constraint is not supported"),
+ JoinConstraint::None => LogicalPlanBuilder::from(left)
+ .join_on(right, join_type, [])?
+ .build(),
}
}
}
diff --git a/datafusion/sql/tests/cases/plan_to_sql.rs b/datafusion/sql/tests/cases/plan_to_sql.rs
index 74abdf075f..2a3c5b5f6b 100644
--- a/datafusion/sql/tests/cases/plan_to_sql.rs
+++ b/datafusion/sql/tests/cases/plan_to_sql.rs
@@ -243,7 +243,7 @@ fn roundtrip_crossjoin() -> Result<()> {
.unwrap();
let expected = "Projection: j1.j1_id, j2.j2_string\
- \n Inner Join: Filter: Boolean(true)\
+ \n Cross Join: \
\n TableScan: j1\
\n TableScan: j2";
diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs
index 19f3d31321..edb614493b 100644
--- a/datafusion/sql/tests/sql_integration.rs
+++ b/datafusion/sql/tests/sql_integration.rs
@@ -898,7 +898,7 @@ fn natural_right_join() {
fn natural_join_no_common_becomes_cross_join() {
let sql = "SELECT * FROM person a NATURAL JOIN lineitem b";
let expected = "Projection: *\
- \n CrossJoin:\
+ \n Cross Join: \
\n SubqueryAlias: a\
\n TableScan: person\
\n SubqueryAlias: b\
@@ -2744,8 +2744,8 @@ fn cross_join_not_to_inner_join() {
"select person.id from person, orders, lineitem where person.id = person.age;";
let expected = "Projection: person.id\
\n Filter: person.id = person.age\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: person\
\n TableScan: orders\
\n TableScan: lineitem";
@@ -2842,11 +2842,11 @@ fn exists_subquery_schema_outer_schema_overlap() {
\n Subquery:\
\n Projection: person.first_name\
\n Filter: person.id = p2.id AND person.last_name = outer_ref(p.last_name) AND person.state = outer_ref(p.state)\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: person\
\n SubqueryAlias: p2\
\n TableScan: person\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: person\
\n SubqueryAlias: p\
\n TableScan: person";
@@ -2934,10 +2934,10 @@ fn scalar_subquery_reference_outer_field() {
\n Projection: count(*)\
\n Aggregate: groupBy=[[]], aggr=[[count(*)]]\
\n Filter: outer_ref(j2.j2_id) = j1.j1_id AND j1.j1_id = j3.j3_id\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: j1\
\n TableScan: j3\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: j1\
\n TableScan: j2";
@@ -3123,7 +3123,7 @@ fn join_on_complex_condition() {
fn lateral_constant() {
let sql = "SELECT * FROM j1, LATERAL (SELECT 1) AS j2";
let expected = "Projection: *\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: j1\
\n SubqueryAlias: j2\
\n Subquery:\
@@ -3138,7 +3138,7 @@ fn lateral_comma_join() {
j1, \
LATERAL (SELECT * FROM j2 WHERE j1_id < j2_id) AS j2";
let expected = "Projection: j1.j1_string, j2.j2_string\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: j1\
\n SubqueryAlias: j2\
\n Subquery:\
@@ -3154,7 +3154,7 @@ fn lateral_comma_join_referencing_join_rhs() {
\n j1 JOIN (j2 JOIN j3 ON(j2_id = j3_id - 2)) ON(j1_id = j2_id),\
\n LATERAL (SELECT * FROM j3 WHERE j3_string = j2_string) as j4;";
let expected = "Projection: *\
- \n CrossJoin:\
+ \n Cross Join: \
\n Inner Join: Filter: j1.j1_id = j2.j2_id\
\n TableScan: j1\
\n Inner Join: Filter: j2.j2_id = j3.j3_id - Int64(2)\
@@ -3178,12 +3178,12 @@ fn lateral_comma_join_with_shadowing() {
) as j2\
) as j2;";
let expected = "Projection: *\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: j1\
\n SubqueryAlias: j2\
\n Subquery:\
\n Projection: *\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: j1\
\n SubqueryAlias: j2\
\n Subquery:\
@@ -3215,7 +3215,7 @@ fn lateral_nested_left_join() {
j1, \
(j2 LEFT JOIN LATERAL (SELECT * FROM j3 WHERE j1_id + j2_id = j3_id) AS j3 ON(true))";
let expected = "Projection: *\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: j1\
\n Left Join: Filter: Boolean(true)\
\n TableScan: j2\
@@ -4281,7 +4281,7 @@ fn test_table_alias() {
let expected = "Projection: *\
\n SubqueryAlias: f\
- \n CrossJoin:\
+ \n Cross Join: \
\n SubqueryAlias: t1\
\n Projection: person.id\
\n TableScan: person\
@@ -4299,7 +4299,7 @@ fn test_table_alias() {
let expected = "Projection: *\
\n SubqueryAlias: f\
\n Projection: t1.id AS c1, t2.age AS c2\
- \n CrossJoin:\
+ \n Cross Join: \
\n SubqueryAlias: t1\
\n Projection: person.id\
\n TableScan: person\
diff --git a/datafusion/sqllogictest/test_files/cte.slt b/datafusion/sqllogictest/test_files/cte.slt
index e9fcf07e77..6056980332 100644
--- a/datafusion/sqllogictest/test_files/cte.slt
+++ b/datafusion/sqllogictest/test_files/cte.slt
@@ -722,7 +722,7 @@ logical_plan
03)----Projection: Int64(1) AS val
04)------EmptyRelation
05)----Projection: Int64(2) AS val
-06)------CrossJoin:
+06)------Cross Join:
07)--------Filter: recursive_cte.val < Int64(2)
08)----------TableScan: recursive_cte
09)--------SubqueryAlias: sub_cte
diff --git a/datafusion/sqllogictest/test_files/group_by.slt b/datafusion/sqllogictest/test_files/group_by.slt
index 8202b806a7..4f2778b5c0 100644
--- a/datafusion/sqllogictest/test_files/group_by.slt
+++ b/datafusion/sqllogictest/test_files/group_by.slt
@@ -4050,7 +4050,7 @@ EXPLAIN SELECT lhs.c, rhs.c, lhs.sum1, rhs.sum1
----
logical_plan
01)Projection: lhs.c, rhs.c, lhs.sum1, rhs.sum1
-02)--CrossJoin:
+02)--Cross Join:
03)----SubqueryAlias: lhs
04)------Projection: multiple_ordered_table_with_pk.c, sum(multiple_ordered_table_with_pk.d) AS sum1
05)--------Aggregate: groupBy=[[multiple_ordered_table_with_pk.c]], aggr=[[sum(CAST(multiple_ordered_table_with_pk.d AS Int64))]]
diff --git a/datafusion/sqllogictest/test_files/join.slt b/datafusion/sqllogictest/test_files/join.slt
index 519fbb887c..fe9ceaa790 100644
--- a/datafusion/sqllogictest/test_files/join.slt
+++ b/datafusion/sqllogictest/test_files/join.slt
@@ -671,7 +671,7 @@ query TT
explain select * from t1 inner join t2 on true;
----
logical_plan
-01)CrossJoin:
+01)Cross Join:
02)--TableScan: t1 projection=[t1_id, t1_name, t1_int]
03)--TableScan: t2 projection=[t2_id, t2_name, t2_int]
physical_plan
@@ -905,7 +905,7 @@ JOIN department AS d
ON (e.name = 'Alice' OR e.name = 'Bob');
----
logical_plan
-01)CrossJoin:
+01)Cross Join:
02)--SubqueryAlias: e
03)----Filter: employees.name = Utf8("Alice") OR employees.name = Utf8("Bob")
04)------TableScan: employees projection=[emp_id, name]
diff --git a/datafusion/sqllogictest/test_files/joins.slt b/datafusion/sqllogictest/test_files/joins.slt
index be9321ddb9..558a9170c7 100644
--- a/datafusion/sqllogictest/test_files/joins.slt
+++ b/datafusion/sqllogictest/test_files/joins.slt
@@ -4050,7 +4050,7 @@ query TT
explain select t1_id, t1_name, i from join_t1 t1 cross join lateral (select * from unnest(generate_series(1, t1_int))) as series(i);
----
logical_plan
-01)CrossJoin:
+01)Cross Join:
02)--SubqueryAlias: t1
03)----TableScan: join_t1 projection=[t1_id, t1_name]
04)--SubqueryAlias: series
diff --git a/datafusion/sqllogictest/test_files/select.slt b/datafusion/sqllogictest/test_files/select.slt
index 0fef56aeea..9910ca8da7 100644
--- a/datafusion/sqllogictest/test_files/select.slt
+++ b/datafusion/sqllogictest/test_files/select.slt
@@ -558,7 +558,7 @@ EXPLAIN SELECT * FROM ((SELECT column1 FROM foo) "T1" CROSS JOIN (SELECT column2
----
logical_plan
01)SubqueryAlias: F
-02)--CrossJoin:
+02)--Cross Join:
03)----SubqueryAlias: T1
04)------TableScan: foo projection=[column1]
05)----SubqueryAlias: T2
diff --git a/datafusion/sqllogictest/test_files/update.slt b/datafusion/sqllogictest/test_files/update.slt
index 59133379d4..aaba6998ee 100644
--- a/datafusion/sqllogictest/test_files/update.slt
+++ b/datafusion/sqllogictest/test_files/update.slt
@@ -67,7 +67,7 @@ logical_plan
01)Dml: op=[Update] table=[t1]
02)--Projection: t1.a AS a, t2.b AS b, CAST(t2.a AS Float64) AS c, CAST(Int64(1) AS Int32) AS d
03)----Filter: t1.a = t2.a AND t1.b > Utf8("foo") AND t2.c > Float64(1)
-04)------CrossJoin:
+04)------Cross Join:
05)--------TableScan: t1
06)--------TableScan: t2
@@ -86,7 +86,7 @@ logical_plan
01)Dml: op=[Update] table=[t1]
02)--Projection: t.a AS a, t2.b AS b, CAST(t.a AS Float64) AS c, CAST(Int64(1) AS Int32) AS d
03)----Filter: t.a = t2.a AND t.b > Utf8("foo") AND t2.c > Float64(1)
-04)------CrossJoin:
+04)------Cross Join:
05)--------SubqueryAlias: t
06)----------TableScan: t1
07)--------TableScan: t2
diff --git a/datafusion/substrait/src/logical_plan/consumer.rs b/datafusion/substrait/src/logical_plan/consumer.rs
index 08e54166d3..5f1824bc4b 100644
--- a/datafusion/substrait/src/logical_plan/consumer.rs
+++ b/datafusion/substrait/src/logical_plan/consumer.rs
@@ -780,7 +780,17 @@ pub async fn from_substrait_rel(
)?
.build()
}
- None => plan_err!("JoinRel without join condition is not allowed"),
+ None => {
+ let on: Vec<String> = vec![];
+ left.join_detailed(
+ right.build()?,
+ join_type,
+ (on.clone(), on),
+ None,
+ false,
+ )?
+ .build()
+ }
}
}
Some(RelType::Cross(cross)) => {
diff --git a/datafusion/substrait/tests/cases/consumer_integration.rs b/datafusion/substrait/tests/cases/consumer_integration.rs
index fffa29df1d..bc38ef8297 100644
--- a/datafusion/substrait/tests/cases/consumer_integration.rs
+++ b/datafusion/substrait/tests/cases/consumer_integration.rs
@@ -73,17 +73,17 @@ mod tests {
\n Aggregate: groupBy=[[]], aggr=[[min(PARTSUPP.PS_SUPPLYCOST)]]\
\n Projection: PARTSUPP.PS_SUPPLYCOST\
\n Filter: PARTSUPP.PS_PARTKEY = PARTSUPP.PS_PARTKEY AND SUPPLIER.S_SUPPKEY = PARTSUPP.PS_SUPPKEY AND SUPPLIER.S_NATIONKEY = NATION.N_NATIONKEY AND NATION.N_REGIONKEY = REGION.R_REGIONKEY AND REGION.R_NAME = Utf8(\"EUROPE\")\
- \n CrossJoin:\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: PARTSUPP\
\n TableScan: SUPPLIER\
\n TableScan: NATION\
\n TableScan: REGION\
- \n CrossJoin:\
- \n CrossJoin:\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: PART\
\n TableScan: SUPPLIER\
\n TableScan: PARTSUPP\
@@ -105,8 +105,8 @@ mod tests {
\n Aggregate: groupBy=[[LINEITEM.L_ORDERKEY, ORDERS.O_ORDERDATE, ORDERS.O_SHIPPRIORITY]], aggr=[[sum(LINEITEM.L_EXTENDEDPRICE * Int32(1) - LINEITEM.L_DISCOUNT)]]\
\n Projection: LINEITEM.L_ORDERKEY, ORDERS.O_ORDERDATE, ORDERS.O_SHIPPRIORITY, LINEITEM.L_EXTENDEDPRICE * (CAST(Int32(1) AS Decimal128(15, 2)) - LINEITEM.L_DISCOUNT)\
\n Filter: CUSTOMER.C_MKTSEGMENT = Utf8(\"BUILDING\") AND CUSTOMER.C_CUSTKEY = ORDERS.O_CUSTKEY AND LINEITEM.L_ORDERKEY = ORDERS.O_ORDERKEY AND ORDERS.O_ORDERDATE < CAST(Utf8(\"1995-03-15\") AS Date32) AND LINEITEM.L_SHIPDATE > CAST(Utf8(\"1995-03-15\") AS Date32)\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: LINEITEM\
\n TableScan: CUSTOMER\
\n TableScan: ORDERS"
@@ -142,11 +142,11 @@ mod tests {
\n Aggregate: groupBy=[[NATION.N_NAME]], aggr=[[sum(LINEITEM.L_EXTENDEDPRICE * Int32(1) - LINEITEM.L_DISCOUNT)]]\
\n Projection: NATION.N_NAME, LINEITEM.L_EXTENDEDPRICE * (CAST(Int32(1) AS Decimal128(15, 2)) - LINEITEM.L_DISCOUNT)\
\n Filter: CUSTOMER.C_CUSTKEY = ORDERS.O_CUSTKEY AND LINEITEM.L_ORDERKEY = ORDERS.O_ORDERKEY AND LINEITEM.L_SUPPKEY = SUPPLIER.S_SUPPKEY AND CUSTOMER.C_NATIONKEY = SUPPLIER.S_NATIONKEY AND SUPPLIER.S_NATIONKEY = NATION.N_NATIONKEY AND NATION.N_REGIONKEY = REGION.R_REGIONKEY AND REGION.R_NAME = Utf8(\"ASIA\") AND ORDERS.O_ORDERDATE >= CAST(Utf8(\"1994-01-01\") AS Date32) AND ORDERS.O_ORDERDATE < CAST(Utf8(\"1995-01-01\") AS Date32)\
- \n CrossJoin:\
- \n CrossJoin:\
- \n CrossJoin:\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
+ \n Cross Join: \
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: CUSTOMER\
\n TableScan: ORDERS\
\n TableScan: LINEITEM\
@@ -206,9 +206,9 @@ mod tests {
\n Aggregate: groupBy=[[CUSTOMER.C_CUSTKEY, CUSTOMER.C_NAME, CUSTOMER.C_ACCTBAL, CUSTOMER.C_PHONE, NATION.N_NAME, CUSTOMER.C_ADDRESS, CUSTOMER.C_COMMENT]], aggr=[[sum(LINEITEM.L_EXTENDEDPRICE * Int32(1) - LINEITEM.L_DISCOUNT)]]\
\n Projection: CUSTOMER.C_CUSTKEY, CUSTOMER.C_NAME, CUSTOMER.C_ACCTBAL, CUSTOMER.C_PHONE, NATION.N_NAME, CUSTOMER.C_ADDRESS, CUSTOMER.C_COMMENT, LINEITEM.L_EXTENDEDPRICE * (CAST(Int32(1) AS Decimal128(15, 2)) - LINEITEM.L_DISCOUNT)\
\n Filter: CUSTOMER.C_CUSTKEY = ORDERS.O_CUSTKEY AND LINEITEM.L_ORDERKEY = ORDERS.O_ORDERKEY AND ORDERS.O_ORDERDATE >= CAST(Utf8(\"1993-10-01\") AS Date32) AND ORDERS.O_ORDERDATE < CAST(Utf8(\"1994-01-01\") AS Date32) AND LINEITEM.L_RETURNFLAG = Utf8(\"R\") AND CUSTOMER.C_NATIONKEY = NATION.N_NATIONKEY\
- \n CrossJoin:\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: CUSTOMER\
\n TableScan: ORDERS\
\n TableScan: LINEITEM\
@@ -230,16 +230,16 @@ mod tests {
\n Aggregate: groupBy=[[]], aggr=[[sum(PARTSUPP.PS_SUPPLYCOST * PARTSUPP.PS_AVAILQTY)]]\
\n Projection: PARTSUPP.PS_SUPPLYCOST * CAST(PARTSUPP.PS_AVAILQTY AS Decimal128(19, 0))\
\n Filter: PARTSUPP.PS_SUPPKEY = SUPPLIER.S_SUPPKEY AND SUPPLIER.S_NATIONKEY = NATION.N_NATIONKEY AND NATION.N_NAME = Utf8(\"JAPAN\")\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: PARTSUPP\
\n TableScan: SUPPLIER\
\n TableScan: NATION\
\n Aggregate: groupBy=[[PARTSUPP.PS_PARTKEY]], aggr=[[sum(PARTSUPP.PS_SUPPLYCOST * PARTSUPP.PS_AVAILQTY)]]\
\n Projection: PARTSUPP.PS_PARTKEY, PARTSUPP.PS_SUPPLYCOST * CAST(PARTSUPP.PS_AVAILQTY AS Decimal128(19, 0))\
\n Filter: PARTSUPP.PS_SUPPKEY = SUPPLIER.S_SUPPKEY AND SUPPLIER.S_NATIONKEY = NATION.N_NATIONKEY AND NATION.N_NAME = Utf8(\"JAPAN\")\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: PARTSUPP\
\n TableScan: SUPPLIER\
\n TableScan: NATION"
@@ -257,7 +257,7 @@ mod tests {
\n Aggregate: groupBy=[[LINEITEM.L_SHIPMODE]], aggr=[[sum(CASE WHEN ORDERS.O_ORDERPRIORITY = Utf8(\"1-URGENT\") OR ORDERS.O_ORDERPRIORITY = Utf8(\"2-HIGH\") THEN Int32(1) ELSE Int32(0) END), sum(CASE WHEN ORDERS.O_ORDERPRIORITY != Utf8(\"1-URGENT\") AND ORDERS.O_ORDERPRIORITY != Utf8(\"2-HIGH\") THEN Int32(1) ELSE Int32(0) END)]]\
\n Projection: LINEITEM.L_SHIPMODE, CASE WHEN ORDERS.O_ORDERPRIORITY = Utf8(\"1-URGENT\") OR ORDERS.O_ORDERPRIORITY = Utf8(\"2-HIGH\") THEN Int32(1) ELSE Int32(0) END, CASE WHEN ORDERS.O_ORDERPRIORITY != Utf8(\"1-URGENT\") AND ORDERS.O_ORDERPRIORITY != Utf8(\"2-HIGH\") THEN Int32(1) ELSE Int32(0) END\
\n Filter: ORDERS.O_ORDERKEY = LINEITEM.L_ORDERKEY AND (LINEITEM.L_SHIPMODE = CAST(Utf8(\"MAIL\") AS Utf8) OR LINEITEM.L_SHIPMODE = CAST(Utf8(\"SHIP\") AS Utf8)) AND LINEITEM.L_COMMITDATE < LINEITEM.L_RECEIPTDATE AND LINEITEM.L_SHIPDATE < LINEITEM.L_COMMITDATE AND LINEITEM.L_RECEIPTDATE >= CAST(Utf8(\"1994-01-01\") AS Date32) AND LINEITEM.L_RECEIPTDATE < CAST(Utf8(\"1995-01-01\") AS Date32)\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: ORDERS\
\n TableScan: LINEITEM"
);
@@ -292,7 +292,7 @@ mod tests {
\n Aggregate: groupBy=[[]], aggr=[[sum(CASE WHEN PART.P_TYPE LIKE Utf8(\"PROMO%\") THEN LINEITEM.L_EXTENDEDPRICE * Int32(1) - LINEITEM.L_DISCOUNT ELSE Decimal128(Some(0),19,4) END), sum(LINEITEM.L_EXTENDEDPRICE * Int32(1) - LINEITEM.L_DISCOUNT)]]\
\n Projection: CASE WHEN PART.P_TYPE LIKE CAST(Utf8(\"PROMO%\") AS Utf8) THEN LINEITEM.L_EXTENDEDPRICE * (CAST(Int32(1) AS Decimal128(15, 2)) - LINEITEM.L_DISCOUNT) ELSE Decimal128(Some(0),19,4) END, LINEITEM.L_EXTENDEDPRICE * (CAST(Int32(1) AS Decimal128(15, 2)) - LINEITEM.L_DISCOUNT)\
\n Filter: LINEITEM.L_PARTKEY = PART.P_PARTKEY AND LINEITEM.L_SHIPDATE >= Date32(\"1995-09-01\") AND LINEITEM.L_SHIPDATE < CAST(Utf8(\"1995-10-01\") AS Date32)\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: LINEITEM\
\n TableScan: PART"
);
@@ -321,7 +321,7 @@ mod tests {
\n Projection: SUPPLIER.S_SUPPKEY\
\n Filter: SUPPLIER.S_COMMENT LIKE CAST(Utf8(\"%Customer%Complaints%\") AS Utf8)\
\n TableScan: SUPPLIER\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: PARTSUPP\
\n TableScan: PART"
);
@@ -353,8 +353,8 @@ mod tests {
\n Aggregate: groupBy=[[LINEITEM.L_ORDERKEY]], aggr=[[sum(LINEITEM.L_QUANTITY)]]\
\n Projection: LINEITEM.L_ORDERKEY, LINEITEM.L_QUANTITY\
\n TableScan: LINEITEM\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: CUSTOMER\
\n TableScan: ORDERS\
\n TableScan: LINEITEM"
@@ -369,7 +369,7 @@ mod tests {
"Aggregate: groupBy=[[]], aggr=[[sum(LINEITEM.L_EXTENDEDPRICE * Int32(1) - LINEITEM.L_DISCOUNT) AS REVENUE]]\
\n Projection: LINEITEM.L_EXTENDEDPRICE * (CAST(Int32(1) AS Decimal128(15, 2)) - LINEITEM.L_DISCOUNT)\
\n Filter: PART.P_PARTKEY = LINEITEM.L_PARTKEY AND PART.P_BRAND = Utf8(\"Brand#12\") AND (PART.P_CONTAINER = CAST(Utf8(\"SM CASE\") AS Utf8) OR PART.P_CONTAINER = CAST(Utf8(\"SM BOX\") AS Utf8) OR PART.P_CONTAINER = CAST(Utf8(\"SM PACK\") AS Utf8) OR PART.P_CONTAINER = CAST(Utf8(\"SM PKG\") AS Utf8)) AND LINEITEM.L_QUANTITY >= CAST(Int32(1) AS Decimal128(15, 2)) AND LINEITEM.L_QUANTITY <= CAST(Int32(1) + Int32(10) AS Decimal128(15, 2)) AND PART.P_SIZE >= Int32(1) AND PART. [...]
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: LINEITEM\
\n TableScan: PART"
);
@@ -398,7 +398,7 @@ mod tests {
\n Filter: LINEITEM.L_PARTKEY = LINEITEM.L_ORDERKEY AND LINEITEM.L_SUPPKEY = LINEITEM.L_PARTKEY AND LINEITEM.L_SHIPDATE >= CAST(Utf8(\"1994-01-01\") AS Date32) AND LINEITEM.L_SHIPDATE < CAST(Utf8(\"1995-01-01\") AS Date32)\
\n TableScan: LINEITEM\
\n TableScan: PARTSUPP\
- \n CrossJoin:\
+ \n Cross Join: \
\n TableScan: SUPPLIER\
\n TableScan: NATION"
);
@@ -422,9 +422,9 @@ mod tests {
\n Subquery:\
\n Filter: LINEITEM.L_ORDERKEY = LINEITEM.L_TAX AND LINEITEM.L_SUPPKEY != LINEITEM.L_LINESTATUS AND LINEITEM.L_RECEIPTDATE > LINEITEM.L_COMMITDATE\
\n TableScan: LINEITEM\
- \n CrossJoin:\
- \n CrossJoin:\
- \n CrossJoin:\
+ \n Cross Join: \
+ \n Cross Join: \
+ \n Cross Join: \
\n TableScan: SUPPLIER\
\n TableScan: LINEITEM\
\n TableScan: ORDERS\
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org