Skip to content

feat: Support sort merge join with a join condition #553

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

Merged
viirya merged 19 commits into from
Aug 30, 2024
Merged

feat: Support sort merge join with a join condition #553

Changes from 1 commit
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
7e8af66
Init
viirya Jun 1, 2024
77f3dde
test
viirya Jul 18, 2024
7f5f50d
test
viirya Aug 19, 2024
e9147bd
test
viirya Aug 19, 2024
2bcb004
Use specified commit to test
viirya Aug 21, 2024
1f82e8a
Fix format
viirya Aug 21, 2024
e401a62
fix clippy
viirya Aug 21, 2024
445afab
fix
viirya Aug 21, 2024
d201a19
fix
viirya Aug 21, 2024
5037b97
Fix
viirya Aug 21, 2024
90c0020
Change to SQL syntax
viirya Aug 21, 2024
0743df5
Disable SMJ LeftAnti with join filter
viirya Aug 21, 2024
5b8a430
Fix
viirya Aug 21, 2024
28e3bf9
Add test
viirya Aug 26, 2024
c362f4c
Add test
viirya Aug 26, 2024
97781d1
Update to last DataFusion commit
viirya Aug 26, 2024
8b6ca06
fix format
viirya Aug 26, 2024
a2f3049
fix
viirya Aug 27, 2024
4a05b02
Update diffs
viirya Aug 27, 2024
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
Prev Previous commit
Next Next commit
Fix format
  • Loading branch information
@viirya
viirya committed Aug 29, 2024
commit 1f82e8a0a5ce8cb69c9ece971d00cd45f73f11fe
31 changes: 20 additions & 11 deletions native/core/src/execution/datafusion/planner.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,7 @@ use datafusion::functions_aggregate::bit_and_or_xor::{bit_and_udaf, bit_or_udaf,
use datafusion::functions_aggregate::min_max::max_udaf;
use datafusion::functions_aggregate::min_max::min_udaf;
use datafusion::functions_aggregate::sum::sum_udaf;
use datafusion::physical_expr_functions_aggregate::aggregate::AggregateExprBuilder;
use datafusion::physical_plan::windows::BoundedWindowAggExec;
use datafusion::physical_plan::InputOrderMode;
use datafusion::{
Expand Down Expand Up @@ -105,14 +106,13 @@ use datafusion_common::{
};
use datafusion_expr::expr::find_df_window_func;
use datafusion_expr::{WindowFrame, WindowFrameBound, WindowFrameUnits, WindowFunctionDefinition};
use datafusion_physical_expr::window::WindowExpr;
use datafusion_physical_expr::expressions::Literal;
use datafusion_physical_expr::window::WindowExpr;
use itertools::Itertools;
use jni::objects::GlobalRef;
use num::{BigInt, ToPrimitive};
use std::cmp::max;
use std::{collections::HashMap, sync::Arc};
use datafusion::physical_expr_functions_aggregate::aggregate::AggregateExprBuilder;

// For clippy error on type_complexity.
type ExecResult<T> = Result<T, ExecutionError>;
Expand Down Expand Up @@ -1312,7 +1312,8 @@ impl PhysicalPlanner {
.alias("count")
.with_ignore_nulls(false)
.with_distinct(false)
.build().map_err(|e| ExecutionError::DataFusionError(e.to_string()))
.build()
.map_err(|e| ExecutionError::DataFusionError(e.to_string()))
}
AggExprStruct::Min(expr) => {
let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
Expand All @@ -1324,7 +1325,8 @@ impl PhysicalPlanner {
.alias("min")
.with_ignore_nulls(false)
.with_distinct(false)
.build().map_err(|e| ExecutionError::DataFusionError(e.to_string()))
.build()
.map_err(|e| ExecutionError::DataFusionError(e.to_string()))
}
AggExprStruct::Max(expr) => {
let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
Expand All @@ -1336,7 +1338,8 @@ impl PhysicalPlanner {
.alias("max")
.with_ignore_nulls(false)
.with_distinct(false)
.build().map_err(|e| ExecutionError::DataFusionError(e.to_string()))
.build()
.map_err(|e| ExecutionError::DataFusionError(e.to_string()))
}
AggExprStruct::Sum(expr) => {
let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
Expand All @@ -1356,7 +1359,8 @@ impl PhysicalPlanner {
.alias("sum")
.with_ignore_nulls(false)
.with_distinct(false)
.build().map_err(|e| ExecutionError::DataFusionError(e.to_string()))
.build()
.map_err(|e| ExecutionError::DataFusionError(e.to_string()))
}
}
}
Expand Down Expand Up @@ -1389,7 +1393,8 @@ impl PhysicalPlanner {
.alias("first")
.with_ignore_nulls(false)
.with_distinct(false)
.build().map_err(|e| e.into())
.build()
.map_err(|e| e.into())
}
AggExprStruct::Last(expr) => {
let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
Expand All @@ -1400,7 +1405,8 @@ impl PhysicalPlanner {
.alias("last")
.with_ignore_nulls(false)
.with_distinct(false)
.build().map_err(|e| e.into())
.build()
.map_err(|e| e.into())
}
AggExprStruct::BitAndAgg(expr) => {
let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
Expand All @@ -1410,7 +1416,8 @@ impl PhysicalPlanner {
.alias("bit_and")
.with_ignore_nulls(false)
.with_distinct(false)
.build().map_err(|e| e.into())
.build()
.map_err(|e| e.into())
}
AggExprStruct::BitOrAgg(expr) => {
let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
Expand All @@ -1420,7 +1427,8 @@ impl PhysicalPlanner {
.alias("bit_or")
.with_ignore_nulls(false)
.with_distinct(false)
.build().map_err(|e| e.into())
.build()
.map_err(|e| e.into())
}
AggExprStruct::BitXorAgg(expr) => {
let child = self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&schema))?;
Expand All @@ -1430,7 +1438,8 @@ impl PhysicalPlanner {
.alias("bit_xor")
.with_ignore_nulls(false)
.with_distinct(false)
.build().map_err(|e| e.into())
.build()
.map_err(|e| e.into())
}
AggExprStruct::Covariance(expr) => {
let child1 =
Expand Down