Skip to content
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

infer right side nullability for LEFT join #5748

Merged
merged 6 commits into from
Mar 29, 2023
+50 −19
Merged

infer right side nullability for LEFT join #5748

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d135f69
weaken schema check on mem table
comphead Mar 27, 2023
1674be4
weaken schema check on mem table
comphead Mar 27, 2023
3302426
fix LEFT join column null inference
comphead Mar 28, 2023
b34e1b5
fix LEFT join column null inference. fix test
comphead Mar 28, 2023
bf8ed7b
fix LEFT join column null inference. fix test
comphead Mar 28, 2023
b8af34d
fix LEFT join column null inference. rollback some code
comphead Mar 28, 2023
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
7 changes: 7 additions & 0 deletions datafusion/core/tests/sqllogictests/test_files/join.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -530,6 +530,13 @@ FROM t1
----
11 a 55

# test create table from query with LEFT join
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It would be great to add tests like this for Right, Full, and all the other flavors of joins.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

#5748 (comment)

statement ok
create table temp as
with t1 as (select 1 as col1, 'asd' as col2),
t2 as (select 1 as col3, 'sdf' as col4)
select col2, col4 from t1 left join t2 on col1 = col3

statement ok
drop table IF EXISTS t1;

Expand Down
36 changes: 30 additions & 6 deletions datafusion/expr/src/logical_plan/builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1041,20 +1041,44 @@ pub fn build_join_schema(
right: &DFSchema,
join_type: &JoinType,
) -> Result<DFSchema> {
let right_fields = right.fields();
let left_fields = left.fields();

let fields: Vec<DFField> = match join_type {
JoinType::Inner | JoinType::Left | JoinType::Full | JoinType::Right => {
let right_fields = right.fields().iter();
let left_fields = left.fields().iter();
JoinType::Inner | JoinType::Full | JoinType::Right => {
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right should be treated symmetrically opposite of Left.

Full should be nullified on both sides.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

#5748 (comment)

// left then right
left_fields.chain(right_fields).cloned().collect()
left_fields
.iter()
.chain(right_fields.iter())
.cloned()
.collect()
}
JoinType::Left => {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll fix other types of joins on separate PR

// left then right, right set to nullable in case of not matched scenario
let right_fields_nullable: Vec<DFField> = right_fields
.iter()
.map(|f| {
let field = f.field().clone().with_nullable(true);
if let Some(q) = f.qualifier() {
DFField::from_qualified(q, field)
} else {
DFField::from(field)
}
})
.collect();
left_fields
.iter()
.chain(&right_fields_nullable)
.cloned()
.collect()
}
JoinType::LeftSemi | JoinType::LeftAnti => {
// Only use the left side for the schema
left.fields().clone()
left_fields.clone()
}
JoinType::RightSemi | JoinType::RightAnti => {
// Only use the right side for the schema
right.fields().clone()
right_fields.clone()
}
};

Expand Down
20 changes: 10 additions & 10 deletions datafusion/optimizer/src/extract_equijoin_predicate.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,7 @@ mod tests {
Some(col("t1.a").eq(col("t2.a"))),
)?
.build()?;
let expected = "Left Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
let expected = "Left Join: t1.a = t2.a [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]\
\n TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]";

Expand All @@ -199,7 +199,7 @@ mod tests {
Some((col("t1.a") + lit(10i64)).eq(col("t2.a") * lit(2u32))),
)?
.build()?;
let expected = "Left Join: t1.a + Int64(10) = t2.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
let expected = "Left Join: t1.a + Int64(10) = t2.a * UInt32(2) [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]\
\n TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]";

Expand All @@ -223,7 +223,7 @@ mod tests {
),
)?
.build()?;
let expected = "Left Join: Filter: t1.a + Int64(10) >= t2.a * UInt32(2) AND t1.b < Int32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
let expected = "Left Join: Filter: t1.a + Int64(10) >= t2.a * UInt32(2) AND t1.b < Int32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]\
\n TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]";

Expand All @@ -250,7 +250,7 @@ mod tests {
),
)?
.build()?;
let expected = "Left Join: t1.a + UInt32(11) = t2.a * UInt32(2), t1.a + Int64(10) = t2.a * UInt32(2) Filter: t1.b < Int32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
let expected = "Left Join: t1.a + UInt32(11) = t2.a * UInt32(2), t1.a + Int64(10) = t2.a * UInt32(2) Filter: t1.b < Int32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]\
\n TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]";

Expand All @@ -277,7 +277,7 @@ mod tests {
),
)?
.build()?;
let expected = "Left Join: t1.a = t2.a, t1.b = t2.b Filter: t1.c = t2.c OR t1.a + t1.b > t2.b + t2.c [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
let expected = "Left Join: t1.a = t2.a, t1.b = t2.b Filter: t1.c = t2.c OR t1.a + t1.b > t2.b + t2.c [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]\
\n TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]";

Expand Down Expand Up @@ -314,9 +314,9 @@ mod tests {
),
)?
.build()?;
let expected = "Left Join: t1.a = t2.a Filter: t1.c + t2.c + t3.c < UInt32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
let expected = "Left Join: t1.a = t2.a Filter: t1.c + t2.c + t3.c < UInt32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]\
\n Left Join: t2.a = t3.a Filter: t2.a + t3.b > UInt32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
\n Left Join: t2.a = t3.a Filter: t2.a + t3.b > UInt32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]\
\n TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]";

Expand Down Expand Up @@ -349,9 +349,9 @@ mod tests {
Some(col("t1.a").eq(col("t2.a")).and(col("t2.c").eq(col("t3.c")))),
)?
.build()?;
let expected = "Left Join: t1.a = t2.a Filter: t2.c = t3.c [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
let expected = "Left Join: t1.a = t2.a Filter: t2.c = t3.c [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]\
\n Left Join: t2.a = t3.a Filter: t2.a + t3.b > UInt32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
\n Left Join: t2.a = t3.a Filter: t2.a + t3.b > UInt32(100) [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]\
\n TableScan: t3 [a:UInt32, b:UInt32, c:UInt32]";

Expand Down Expand Up @@ -380,7 +380,7 @@ mod tests {
Some(filter),
)?
.build()?;
let expected = "Left Join: t1.a + CAST(Int64(1) AS UInt32) = t2.a + CAST(Int32(2) AS UInt32) [a:UInt32, b:UInt32, c:UInt32, a:UInt32, b:UInt32, c:UInt32]\
let expected = "Left Join: t1.a + CAST(Int64(1) AS UInt32) = t2.a + CAST(Int32(2) AS UInt32) [a:UInt32, b:UInt32, c:UInt32, a:UInt32;N, b:UInt32;N, c:UInt32;N]\
\n TableScan: t1 [a:UInt32, b:UInt32, c:UInt32]\
\n TableScan: t2 [a:UInt32, b:UInt32, c:UInt32]";

Expand Down
6 changes: 3 additions & 3 deletions datafusion/optimizer/src/push_down_projection.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -733,7 +733,7 @@ mod tests {
vec![
DFField::new(Some("test"), "a", DataType::UInt32, false),
DFField::new(Some("test"), "b", DataType::UInt32, false),
DFField::new(Some("test2"), "c1", DataType::UInt32, false),
DFField::new(Some("test2"), "c1", DataType::UInt32, true),
],
HashMap::new(),
)?,
Expand Down Expand Up @@ -776,7 +776,7 @@ mod tests {
vec![
DFField::new(Some("test"), "a", DataType::UInt32, false),
DFField::new(Some("test"), "b", DataType::UInt32, false),
DFField::new(Some("test2"), "c1", DataType::UInt32, false),
DFField::new(Some("test2"), "c1", DataType::UInt32, true),
],
HashMap::new(),
)?,
Expand Down Expand Up @@ -817,7 +817,7 @@ mod tests {
vec![
DFField::new(Some("test"), "a", DataType::UInt32, false),
DFField::new(Some("test"), "b", DataType::UInt32, false),
DFField::new(Some("test2"), "a", DataType::UInt32, false),
DFField::new(Some("test2"), "a", DataType::UInt32, true),
],
HashMap::new(),
)?,
Expand Down