fjetter
commented
1 month ago Thanks for opening this issue and apologies for the lack of engagement on https://github.com/coiled/benchmarks/issues/1498 so far.
Our intention is not to publish any unfair results and we are interested in comparing engines on equal footing to the best of our ability.
You are mentioning a couple of different topics and some of them require further discussion. From what I can tell, you are raising these problems
- Premature filtering and projections
- leftsemi joins
- Definition of groupby aggregations
I will grant you that 1.) is indeed not correct in some situations. This wasn't done intentionally and we still have to address this. I think https://github.com/coiled/benchmarks/issues/1498 describes this problem already such that I will not open a separate issue.
I think the two other issues require a bit further debate.
Leftsemi
Dask is using a leftsemi statement in three queries, namely query 4, 18 and 20
Let's talk about Q4 in detail since it's the simplest example. The functional definition of the query according to the TPC-H specification is
select
o_orderpriority,
count(*) as order_count
from
orders
where
o_orderdate >= date '[DATE]'
and o_orderdate < date '[DATE]' + interval '3' month
and exists (
select
*
from
lineitem
where
l_orderkey = o_orderkey
and l_commitdate < l_receiptdate
)
group by
o_orderpriority
order by
o_orderpriority;The leftsemi join method is used to implement the exist statement on a correlated subquery (see the where clause l_orderkey = o_orderkey) in the above SQL expression. Translating this to a DataFrame is a little difficult since SQL is a declarative language and DataFrames are typically imperative.
My attempt to put into words what the above query is describing...
"Take every row of left (i.e. order) for which there exists a matching (l_orderkey = o_orderkey) row in right (i.e. lineitem) which also satisfies the condition l_commitdate < l_receiptdate".
This is pretty much the definition of a leftsemi join with a filter applied, isn't it?
Comparing this to how polars is implementing this query, namely joining the table with a simple left join and subsequently dropping duplicates on ["o_orderpriority", "l_orderkey"] is semantically a different operation. If the left table already included duplicates, this operation would yield a different result. The left table does not include duplicates and it is therefore yielding the same result. However, that approach includes knowledge about the data while the leftsemi join does not.
The cases for query 18 and 20 are more complex and instead of exists they describe subqueries with in and > operators but the argument is pretty much the same.
I would actually encourage polars to use a semi join as well. Thoughts?
Groupby aggregations
As an example, you are pointing here to Q1 and especially how the following line is being translated.
select
...
sum(l_extendedprice*(1-l_discount)*(1+l_tax)) as sum_charge,
...
from
lineitem
...In dask and pandas we currently don't have a way to express more complex columnar expressions like this. Instead, we do...
lineitem_filtered["sum_charge"] = (
lineitem_filtered.l_extendedprice
* (1 - lineitem_filtered.l_discount)
* (1 + lineitem_filtered.l_tax)
)
...
gb = lineitem_filtered.groupby(["l_returnflag", "l_linestatus"])
total = gb.agg(
{
...
"sum_charge": "sum",
...
}
)I fully agree that the way polars can describe this operation using expressions is more elegant. We currently don't support this but have it on our roadmap (see https://github.com/dask/dask-expr/issues/386). I see how optimizers could produce suboptimal results or how time spent on optimization could skew the results but I would argue that this is negligible. On the other hand, this does unlock a whole range of optimizations that we currently don't have access to so I would likely argue that engines like polars or spark have an advantage over dask and pandas because of this. I think we have to acknowledge this bias but I wouldn't want to stop comparing engines because of this. There is no DataFrame standard similar to the SQL ISO definition so we have to make the best out of what we have.
ritchie46
mrocklin
Similar #1498. I think that as the queries are currently written it isn't a fair comparison between DataFrame API's.
For SQL it is fair as the TPCH benchmark states that all engines should parse the same SQL. However for DataFrames with different API's this is ofcourse harder.
TPCH is not only a operation benchmark, it is also an optimization benchmark. If we compare against SQL engines, the DataFrame API's shouldn't enjoy more optimizations than the SQL queries.
For this reason we removed hand written optimizations on the polars tpch benchmarks in Polars, pandas, Dask and modin:
To make the benchmarks apples to apples I think this should be done in this repo as well. The Dask queries for instance show hand written "leftsemi"
pre-filtering:
< actually almost all queries in some form >
And pre-aggregation computation where this defined in the aggregation context.
In many group-by's:
For this one, we are lenient towards pandas API's upstream as there isn't a good way to describe this. But since
dask-expris developed I believe Dask should just inline this and leave this to the optimizer.If we compare DataFrame API's and SQL I truly believe it is important that we give all implementations a fair starting position. Optimization engines could come to the same conclusions, but have to spend compute to do so, and often will not come to the most optimal conclusions.