hhhizzz commented on issue #1826:
URL:
https://github.com/apache/datafusion-ballista/issues/1826#issuecomment-4910371588
I did some profiling on Q72 and I think this issue is very likely caused by
the same underlying shape that @alamb and @Dandandan pointed out: Q72 can
generate a very large intermediate with string columns before the selective
dimension predicates are applied.
My setup is not exactly the same as the original AQE repro, so please treat
this as additional evidence rather than a definitive root-cause proof:
- 3-node Ballista bare-metal cluster
- TPC-DS SF10 Parquet data on S3/Ceph
- scheduler/executors CPU-pinned
- default `datafusion.execution.target_partitions = 16`
Baseline Q72 took about `209s`; `EXPLAIN ANALYZE` took about `175s`.
The expensive part was not the Parquet scan itself. The plan first joined:
```text
catalog_sales: 14.40M rows
inventory: 133.1M rows
```
on `cs_item_sk = inv_item_sk` plus `inv_quantity_on_hand < cs_quantity`,
producing:
```text
1.12B rows
~96GB at the join output
```
Then the plan joined `warehouse` and `item`, adding wide/string payloads,
and only later applied the selective `customer_demographics`,
`household_demographics`, and `date_dim` filters. One shuffle after the `item`
join carried roughly `195GB-203GB` worth of payload and had very heavy spill.
That seems consistent with the `OffsetOverflowError`:
`schema_force_view_types=true` can avoid the i32 offset overflow symptom, but
it does not fix the much larger intermediate/memory pressure.
I tried a few things:
1. Increasing `target_partitions` did not solve it. `target_partitions=24`
and `72` were not useful in my test.
2. Enabling parquet/dynamic filter knobs on the original SQL did not solve
it either.
3. Adding redundant semi-join filters to reduce the fact tables before the
`catalog_sales ⋈ inventory` join helped a lot.
The successful rewrite added equivalent prefilters like:
```sql
cs_bill_cdemo_sk in (
select cd_demo_sk
from customer_demographics
where cd_marital_status = 'M'
)
cs_bill_hdemo_sk in (
select hd_demo_sk
from household_demographics
where hd_buy_potential = '1001-5000'
)
cs_sold_date_sk in (
select d_date_sk
from date_dim
where d_year = 2001
)
inv_date_sk in (
select d2.d_date_sk
from date_dim d2
join (
select distinct d_week_seq
from date_dim
where d_year = 2001
) y
on d2.d_week_seq = y.d_week_seq
)
```
With that rewrite, the profile changed roughly to:
```text
catalog_sales before fact join: 14.40M -> 95.77K
inventory before fact join: 133.1M -> 27.03M
fact join output: 1.12B -> 2.42M
runtime: ~209s -> ~7.3s
```
The rewritten query produced the same result as the baseline in my run.
I would be happy to contribute code here, but I would appreciate guidance on
the preferred direction:
1. Should this be addressed in DataFusion optimizer as a general semi-join
reduction / constraint propagation rule for selective dimension joins?
2. Would a narrower TPC-DS Q72-inspired optimizer rule be acceptable as a
first step, or is that too query-specific?
3. Should we also add a Ballista/AQE regression test for this issue,
checking that Q72 does not build the huge string intermediate / trigger
`OffsetOverflowError`?
4. Is there still a separate Ballista-side fix wanted for the incorrect
`broadcast: true` promotion mentioned above, or should that be tracked
separately?
My current interpretation is: LargeUtf8/View types can mitigate the
immediate overflow, but the main performance and memory fix is to avoid
producing the huge intermediate in the first place.
--
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]
