ndscoder opened a new issue, #26779:
URL: https://github.com/apache/datafusion/issues/26779
### Describe the bug
With a filter such as `WHERE k / 10 = 5`, the physical plan treats `k`
itself as constant: the sort for `ORDER BY k, id` becomes `SortExec:
expr=[id]`, and a `GROUP BY k` runs as if its input were sorted by `k`. The
rows come back unordered and the groups come back unordered, with no error. In
a plan with more partitions, one group can be split into several rows.
`k / 10 = 5` holds for `k` from 50 to 59, so `k` is not constant. It looks
as if the filter's interval analysis turns `f(k) = literal` into a single-point
interval for `k` when `f` is monotonic but not injective (integer division,
`CAST(ts AS DATE)`), and that point then becomes a constant in the equivalence
properties.
Seen with:
- `k / 10 = 5`
- `k / 10 IN (5)`
- `k / 10 BETWEEN 5 AND 5`
- `k / 10 >= 5 AND k / 10 <= 5`
- `k / 10 = j AND j = 5` (but not `j = 5 AND k / 10 = j`)
- `CAST(ts AS DATE) = DATE '2024-01-06'` with `ORDER BY ts`
Not seen with `k + 1 = 51`, `abs(k) = 50`, `floor(k / 10.0) = 5` or `k =
50`. It does not depend on Parquet, unions or `collect_statistics`: one
in-memory partition shows it.
### To Reproduce
With datafusion-python 55.0.0:
```python
import pyarrow as pa
from datafusion import SessionContext
keys = [(row * 7) % 120 for row in range(400)]
context = SessionContext()
table = pa.table({"k": pa.array(keys, pa.int64()), "id":
pa.array(range(400), pa.int64())})
context.register_record_batches("t", [table.to_batches()])
rows = context.sql("SELECT k, id FROM t WHERE k / 10 = 5 ORDER BY k,
id").to_pylist()
print("sorted:", [(r["k"], r["id"]) for r in rows] == sorted((r["k"],
r["id"]) for r in rows))
print("first keys:", [r["k"] for r in rows[:8]])
groups = context.sql("SELECT k, count(*) AS n FROM t WHERE k / 10 = 5 GROUP
BY k ORDER BY k").to_pylist()
print("groups:", [g["k"] for g in groups])
print(context.sql("EXPLAIN SELECT k, id FROM t WHERE k / 10 = 5 ORDER BY k,
id").to_pylist()[-1]["plan"])
```
Output (the order of `groups` varies between runs; it is never sorted):
```
sorted: False
first keys: [56, 55, 54, 53, 52, 59, 51, 58]
groups: [58, 59, 53, 54, 55, 57, 56, 50, 52, 51]
SortExec: expr=[id@1 ASC NULLS LAST], preserve_partitioning=[false]
FilterExec: k@0 / 10 = 5
DataSourceExec: partitions=1, partition_sizes=[1]
```
### Expected behavior
`ORDER BY k, id` sorted by `k` and then `id`, and one row per `k` from
`GROUP BY k`, in order of `k`, whatever the filter. The plan should keep `k` in
the sort: `SortExec: expr=[k@0 ASC NULLS LAST, id@1 ASC NULLS LAST]`.
### Additional context
A query comes back in the wrong order, or with one group's rows split across
several groups, and nothing signals it. Filtering timestamps by their day
(`CAST(ts AS DATE) = ...`) and ordering by the timestamp is a common query
shape. Our workaround is to detect such filters and run those statements
outside DataFusion.
### Willingness to contribute
I cannot contribute a fix for this bug at this time
--
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]
