jayzhan211 opened a new issue, #25296:
URL: https://github.com/apache/datafusion/issues/25296
## Describe the bug
**Physical filter pushdown resolves columns by name, so when a schema
contains two columns with the same name, a filter over the second one is pushed
to the first.** With a TopK (`ORDER BY ... LIMIT n`) dynamic filter this
returns wrong results under the default configuration.
Trigger: two joined tables that share a column name. Below, `orders` and
`payments` both have `amount`, so the lower join's output schema is `[amount,
id, amount]`. A filter over `payments.amount` is looked up as `"amount"` in
that schema, matches the *first* `amount`, which is `orders.amount`, and is
pushed to the wrong table. If either column were renamed, the query would be
correct.
Query: the order with the smallest payment.
```sql
SELECT o.amount, p.amount
FROM orders o
JOIN payments p ON o.id = p.id
JOIN customers c ON p.id = c.id
ORDER BY p.amount LIMIT 1;
```
How the filter travels down the physical plan (the lower join is built on
`payments` because `orders` is larger, and a projection restores the written
column order):
```text
SortExec TopK(fetch=1), ORDER BY amount@1 dynamic filter: amount@1
< 30
│ ("amount" at output index
1 = p.amount, correct)
▼
HashJoinExec (upper) output [o.amount, p.amount]
│ index 1 is on the right child, so the filter may go there.
│ Position inside the child is resolved BY NAME: index_of("amount")
▼
HashJoinExec (lower) output [amount, id, amount] <-- two columns
named "amount"
▲ ▲
│ └── p.amount the column the
filter is about
└──────────────── o.amount the column
index_of("amount") returns:
same name, first
match wins
│
├── DataSourceExec payments receives nothing (should have
received amount < 30)
└── DataSourceExec orders receives amount < 30 WRONG TABLE
```
Why that returns the wrong row. The joined rows are (100, 30), (200, 20),
(300, 10), and the answer should be (300, 10). `orders` streams through the
lower join one row per batch (`batch_size = 1`, one row per row group), and
after each batch the TopK publishes "only rows with `amount` below my current
best can still matter":
| step | `orders` scan | row reaching the TopK | TopK best so far | filter
published by the TopK |
|---|---|---|---|---|
| 1 | reads order 1 (amount 100) | (100, 30) | 30 | `amount < 30` |
| 2 | order 2 (amount 200): `200 < 30` is false, row group pruned | nothing
| 30 | unchanged |
| 3 | order 3 (amount 300): `300 < 30` is false, pruned | nothing | 30 |
unchanged |
| 4-6 | orders 4 to 6 pruned the same way | nothing | 30 | unchanged |
The threshold 30 was computed from `payments.amount` (values 30, 20, 10). On
that column `amount < 30` keeps payments 20 and 10, the ones that could still
beat the current best. But it is applied to `orders.amount` (values 100, 200,
..., 600), where no value is below 30. The parquet reader checks each remaining
row group's min/max against the filter before decoding it, finds `200 < 30`,
`300 < 30`, ... impossible, and skips them all. The rows (200, 20) and (300,
10) never reach the TopK, which reports the only row it ever saw:
```text
expected actual
o.amount, p.amount 300, 10 100, 30
```
Root cause: `FilterRemapper::try_remap` in
`datafusion/physical-plan/src/filter_pushdown.rs` maps a parent filter into a
child's schema with `child_schema.index_of(col.name())`. Arrow's `index_of`
returns the *first* field with that name, so with duplicated names the lookup
is ambiguous and silently picks the wrong column.
`HashJoinExec::gather_filters_for_pushdown` reaches it through
`ChildFilterDescription::from_child_with_allowed_indices`; the index check
there only decides *whether* a column belongs to the child, the *position*
inside the child is still looked up by name. Physical schemas are not required
to have unique names, and a join of two tables that share a column name
produces one every time.
## To Reproduce
```sql
set datafusion.execution.target_partitions = 1;
-- one row per batch, so the TopK filter is updated while the `orders` scan
still has row groups to read
set datafusion.execution.batch_size = 1;
CREATE TABLE orders_src(id INT, amount INT) AS VALUES (1, 100), (2, 200),
(3, 300), (4, 400), (5, 500), (6, 600);
CREATE TABLE payments_src(id INT, amount INT) AS VALUES (1, 30), (2, 20),
(3, 10);
CREATE TABLE customers_src(id INT) AS VALUES (1), (2), (3);
COPY orders_src TO '/tmp/dup/orders.parquet' STORED AS PARQUET OPTIONS
('format.max_row_group_size' '1');
COPY payments_src TO '/tmp/dup/payments.parquet' STORED AS PARQUET;
COPY customers_src TO '/tmp/dup/customers.parquet' STORED AS PARQUET;
CREATE EXTERNAL TABLE orders(id INT, amount INT) STORED AS PARQUET LOCATION
'/tmp/dup/orders.parquet';
CREATE EXTERNAL TABLE payments(id INT, amount INT) STORED AS PARQUET
LOCATION '/tmp/dup/payments.parquet';
CREATE EXTERNAL TABLE customers(id INT) STORED AS PARQUET LOCATION
'/tmp/dup/customers.parquet';
SELECT o.amount, p.amount
FROM orders o
JOIN payments p ON o.id = p.id
JOIN customers c ON p.id = c.id
ORDER BY p.amount LIMIT 1;
```
```text
+--------+--------+
| amount | amount |
+--------+--------+
| 100 | 30 |
+--------+--------+
```
The same wrong row comes back with
`datafusion.execution.parquet.pushdown_filters = true`. With
`datafusion.optimizer.enable_topk_dynamic_filter_pushdown = false` the result
is correct.
## Expected behavior
```text
+--------+--------+
| amount | amount |
+--------+--------+
| 300 | 10 |
+--------+--------+
```
## Additional context
`EXPLAIN` of the query:
```text
SortExec: TopK(fetch=1), expr=[amount@1 ASC NULLS LAST],
preserve_partitioning=[false]
--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@1)],
projection=[amount@1, amount@3]
----DataSourceExec: file_groups={1 group: [[.../customers.parquet]]},
projection=[id], file_type=parquet
----HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)],
projection=[amount@3, id@0, amount@1]
------DataSourceExec: file_groups={1 group: [[.../payments.parquet]]},
projection=[id, amount], file_type=parquet, predicate=DynamicFilter [ empty ],
dynamic_rg_pruning=eligible
------DataSourceExec: file_groups={1 group: [[.../orders.parquet]]},
projection=[id, amount], file_type=parquet, predicate=DynamicFilter [ empty ]
AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible
```
`EXPLAIN ANALYZE` shows the misplaced filter on the `orders` scan:
`predicate=DynamicFilter [ ... ] AND DynamicFilter [ amount@1 < 30 ]`.
Why the reproducer needs this shape: the name lookup only misfires when a
join's child is itself a join (two same-named columns in one schema), and the
wrongly targeted table must be the streaming probe side, so that the filter
still has row groups left to prune when the TopK updates it. With the default
batch size the lower join coalesces its output into one batch and the bug is
masked; `batch_size = 1` and one row per row group make it deterministic.
Found while reviewing #25255, which removes one caller of the name-based
semi-join routing but does not change `try_remap`. Reproduced on `main` at
a407990b4 and on that PR branch. The static-predicate path is less exposed only
because the logical optimizer usually pushes static predicates to the scans
before the physical pass.
A fix probably needs the remap to carry positions rather than names:
`HashJoinExec` already knows which parent output index maps to which child
index (`column_indices`), so `from_child_with_allowed_indices` could take a
`HashMap<usize, usize>` from parent index to child index instead of a
`HashSet<usize>` plus a name lookup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
