apache / apache/datafusion

Invalid partition scheme in sort if intermediate projection has dropped a partition column

Open
#5,184 2 comments 0 reactions 0 assignees View on GitHub
bug
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

**Describe the bug**
I am hitting this error with benchmark query 3 in ray-sql:

```
physical_plan::to_proto() unsupported expression UnKnownColumn { name: "ps_supplycost" }
```

Here is some debug output that I think shows the root cause. This is from walking down the plan. There is a join that uses `ps_supplycost` as a join key so the partitioning scheme includes `ps_supplycost`. This column is then dropped in a projection, but the `SortExec`'s partition scheme still uses `ps_supplycost`.

```
plan = SortExec: [s_acctbal@0 DESC,n_name@2 ASC NULLS LAST,s_name@1 ASC NULLS LAST,p_partkey@3 ASC NULLS LAST]

partitioning_scheme = Hash([Column { name: "p_partkey", index: 3 }, UnKnownColumn { name: "ps_supplycost" }], 4)
---

plan = ProjectionExec: expr=[s_acctbal@5 as s_acctbal, s_name@2 as s_name, n_name@7 as n_name, p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_address@3 as s_address, s_phone@4 as s_phone, s_comment@6 as s_comment]

partitioning_scheme = Hash([Column { name: "p_partkey", index: 3 }, UnKnownColumn { name: "ps_supplycost" }], 4)
---

plan = ProjectionExec: expr=[p_partkey@0 as p_partkey, p_mfgr@1 as p_mfgr, s_name@8 as s_name, s_address@9 as s_address, s_phone@11 as s_phone, s_acctbal@12 as s_acctbal, s_comment@13 as s_comment, n_name@15 as n_name]

partitioning_scheme = Hash([Column { name: "p_partkey", index: 0 }, UnKnownColumn { name: "ps_supplycost" }], 4)
---

plan = CoalesceBatchesExec: target_batch_size=8192

partitioning_scheme = Hash([Column { name: "p_partkey", index: 0 }, Column { name: "ps_supplycost", index: 6 }], 4)
---

plan = HashJoinExec: mode=Partitioned, join_type=Inner, on=[(Column { name: "p_partkey", index: 0 }, Column { name: "ps_partkey", index: 0 }), (Column { name: "ps_supplycost", index: 6 }, Column { name: "__value", index: 1 })]

partitioning_scheme = Hash([Column { name: "p_partkey", index: 0 }, Column { name: "ps_supplycost", index: 6 }], 4)
---
```

**To Reproduce**
Steps to reproduce the behavior:

**Expected behavior**
A clear and concise description of what you expected to happen.

**Additional context**
Add any other context about the problem here.

Contributor guide

Open the contributing guide

Research direction

Start by reproducing benchmark query 3 in ray-sql and inspect the plan from HashJoinExec through CoalesceBatchesExec, ProjectionExec, and SortExec. Trace the partitioning scheme into physical_plan::to_proto(); done means a dropped ps_supplycost column is not retained in the SortExec partition scheme and serialization no longer reports an UnKnownColumn.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.