Possible hidden schema mismatch for HashJoin in ProjectionExec
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### Describe the bug
The input schema and schema for projection are different in HashJoin cases because we might add additional alias expression for hash join. Since the schema has no *additional alias column* that the `input schema` has, it is possible to cause schema mismatch. In `stats_projection()`, because we ignore the error so we don't encounter such issue but I think we should not eat the error but fix the schema and column index matching issue instead.
https://github.com/apache/datafusion/blob/2464703c84c400a09cc59277018813f0e797bb4e/datafusion/physical-plan/src/projection.rs#L278-L283
### To Reproduce
Take this query for example,
```
statement count 0
create table t1(name varchar, id int) as values ('df', 1), ('rs', 2), ('po', 3);
statement ok
create table t2(product varchar, sid bigint) as values ('er', 1), ('sd', 2), ('ge', 3);
query TT
select t1.name, t2.product from t1 inner join t2 on t1.id = t2.sid;
----
df er
rs sd
po ge
```
If we add the code to `ProjectExec::try_new()`
```rust
// Construct a map from the input expressions to the output expression of the Projection
let projection_mapping = ProjectionMapping::try_new(&expr, &input_schema)?;
let cache =
Self::compute_properties(&input, &projection_mapping, Arc::clone(&schema))?;
println!("schema: {:?}", schema);
println!("input schema: {:?}", input_schema);
for (e, _) in expr.iter() {
// panic here
let _dt = e.data_type(&schema).unwrap();
}
Ok(Self {
expr,
schema,
input,
metrics: ExecutionPlanMetricsSet::new(),
cache,
})
```
we can see that `input_schema` has additional column `CAST(t1.id AS Int64)`
```
Schema {
fields: [
Field {
name: "name",
data_type: Utf8,
nullable: true,
dict_id: 0,
dict_is_ordered: false,
metadata: {}
},
Field {
name: "id",
data_type: Int32,
nullable: true,
dict_id: 0,
dict_is_ordered: false,
metadata: {}
},
Field {
name: "product",
data_type: Utf8,
nullable: true,
dict_id: 0,
dict_is_ordered: false,
metadata: {}
},
Field {
name: "sid",
data_type: Int64,
nullable: true,
dict_id: 0,
dict_is_ordered: false,
metadata: {}
}
],
metadata: {}
}
Input Schema {
fields: [
Field {
name: "name",
data_type: Utf8,
nullable: true,
dict_id: 0,
dict_is_ordered: false,
metadata: {}
},
Field {
name: "id",
data_type: Int32,
nullable: true,
dict_id: 0,
dict_is_ordered: false,
metadata: {}
},
Field {
name: "CAST(t1.id AS Int64)",
data_type: Int64,
nullable: true,
dict_id: 0,
dict_is_ordered: false,
metadata: {}
},
Field {
name: "product",
data_type: Utf8,
nullable: true,
dict_id: 0,
dict_is_ordered: false,
metadata: {}
},
Field {
name: "sid",
data_type: Int64,
nullable: true,
dict_id: 0,
dict_is_ordered: false,
metadata: {}
}
],
metadata: {}
}
```
If we use the `schema` for projection, we have the column mismatch index that could cause Column's `bounds_check`.
The error is because of the out of bound index. `sid` is created as index 4 for `input_schema` but the schema we used in projection has only 4 columns.
```
thread 'tokio-runtime-worker' panicked at datafusion/physical-plan/src/projection.rs:102:44:
called `Result::unwrap()` on an `Err` value: Internal("PhysicalExpr Column references column 'sid' at index 4 (zero-based) but input schema only has 4 columns: [\"name\", \"id\", \"product\", \"sid\"]")
```
### Expected behavior
I think we might either recreate expressions with the new index that match the `schema` or use the `input_schema` for projection. I'm not sure which is the right choice yet.
### Additional context
_No response_
Contributor guide
Research direction
Start in datafusion/physical-plan/src/projection.rs, especially stats_projection() and ProjectExec::try_new(), then run the provided HashJoin reproduction query. Compare the projection schema with input_schema and trace the column index for sid; done means the reproduction no longer produces the out-of-bounds schema mismatch while the projection error is handled correctly.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust, sql
- Domain
- backend, databases
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 38/100