Database Doctor
Writing on databases, performance, and engineering.

TPC-H Query 13 - Left / right handed joins and pre-aggregation

Today, we are going to learn about left and right handed joins. These joins allow the execution engine of a database to still pick the smaller side as the hash build, even when the smaller side is an outer, row preserving side.

It's easier to illustrate with an example - which brings us right to Query 13.

Query 13

Here is Query 13 - pay special attention to the LEFT OUTER JOIN here:

SELECT c_count,
       COUNT(*) AS custdist
FROM (SELECT c_custkey,
             COUNT(o_orderkey) AS c_count
      FROM tpch.customer
      LEFT OUTER JOIN tpch.orders
          ON c_custkey = o_custkey
         AND o_comment NOT LIKE ' % special % requests % '
      GROUP BY c_custkey) AS c_orders (c_custkey, c_count)
GROUP BY c_count
ORDER BY custdist DESC,
         c_count DESC

Estimation and Join Order

The only filter in the query is: o_comment NOT LIKE ' % special % requests % '. For most databases, this one is nearly impossible to estimate.

But here is the selectivity:

Filter Selectivity Cardinality
o_comment NOT LIKE ' % special % requests % ' 99.9% 1.5M

Basically, the filter does almost nothing and it is nearly impossible to estimate.

Even if the database uses its typical "I have no idea" estimate of 1/3 - the correct join order is still to build a hash table on customer and probe into that with orders.

Thus, we expect this join order: orderscustomer.

Right and Left handed joins

Notice that we have a LEFT JOIN here:

FROM tpch.customer
LEFT OUTER JOIN tpch.orders
  ON c_custkey = o_custkey

We know we want to build a hash table on customer (because it is 10x smaller than orders).

But how do we execute a LEFT JOIN where we are probing with orders, but want to keep the customer entries that are not matched?

Let's first consider how we would run the query if the join was the other way around. If this was what the user asked for:

FROM tpch.orders
LEFT OUTER JOIN tpch.customer 
  ON o_custkey = c_custkey 

Then we could easily run a loop like this:

# Build hash table on c
c_hash = dict()
for c_row in c:
    c_hash[c_row.c_custkey] = c_row

# Probe the hash
for o_row in o:
    if o_row.o_custkey in c_hash:
      # INNER part, could be more than one match
      probe_matches = c_hash[o_row.o_custkey] 
      for c_row in probe_matches:
        output.append(Row(o_row, c_row))
    else:
        # no match, emit the OUTER LEFT
        output.append(Row(o_row, None))

This way of running an outer join is what I call a "left-handed" join. SQL Arena renders this as LEFT OUTER JOIN.

But... The above isn't the query we are being asked to run! We are being asked to build a hash table on customer (to save memory) and emit rows from customer that did not have a match in orders (as well as the rows that match both sides).

This requires a slightly modified join algorithm. Something like this:

# Build hash table on c
c_hash = dict()
for c_row in c:
    c_row.matched = False
    c_hash[c_row.c_custkey] = c_row

# Probe the hash for matches
for o_row in o:
  if o_row.o_custkey in c_hash:
    # INNER
    probe_matches = c_hash[o_row.o_custkey] 
    for c_row in probe_matches:
      output.append(Row(o_row, c_row))
      # Remember that this is now matched
      c_row.matched = True

# Loop over the hash to emit non-matched rows
for c_row in c_hash.values():
    if c_row.matched:
        # already match in INNER
        continue
    # This was an outer row
    output.append(Row(None, c_row))
    

This join is what I call the "right-handed" outer join. SQL Arena renders this as a RIGHT OUTER JOIN.

Can all databases do right-handed joins?

Why run right handed joins at all? Because they allow us to build much smaller hash tables - which means we need less memory to run the query.

Here, we see DuckDB run this right handed join:

Estimate    Actual  Operator
       -        42  SORT count_star(), c_orders.c_count
   83106        42  AGGREGATE count_star() GROUP BY HASH #0
  124908    150000  PROJECT c_count
  124908    150000  PROJECT COUNT(o_orderkey)
  124908    150000  AGGREGATE COUNT(#1) GROUP BY HASH #0
  300000   1549963  PROJECT c_custkey, o_orderkey
  300000   1549963  RIGHT OUTER JOIN HASH ON o_custkey = c_custkey <--- Right handed
  150000    150000  │└TABLE SCAN customer
  300000   1499959  TABLE SCAN orders WHERE o_comment NOT LIKE ' % special % requests % '

PostgreSQL does the same thing DuckDB does, even Clickhouse manages a right join.

But if we look at DataBricks there is a large build side of orders. In a distributed system, this not only uses more memory, it also causes the engine to distribute more rows than necessary.

Estimate    Actual  Operator
  147000        42  SORT custdist DESC NULLS LAST, c_orders.c_count DESC NULLS LAST
  147000        42  AGGREGATE COUNT(1) GROUP BY HASH c_orders.c_count
  147000       460  DISTRIBUTE HASH ON c_orders.c_count
  147000       460  AGGREGATE COUNT(1) GROUP BY HASH c_orders.c_count
  147000    150000  AGGREGATE COUNT(o_orderkey) GROUP BY HASH c_custkey
      0B   1549963  LEFT OUTER JOIN HASH ON c_custkey = o_custkey
      0B   1499959  │└DISTRIBUTE HASH ON o_custkey
 1500000   1499959  TABLE SCAN orders WHERE  NOT o_comment LIKE ' % special % requests % '
      0B    150000  DISTRIBUTE HASH ON c_custkey
  150000    150000  TABLE SCAN customer

From my parsing of the plans, it appears that DataFusion and Trino has the same problem.

Pre-aggregation

SQL Server makes a very different query plan than the other databases. It looks like this:

Estimate    Actual  Operator
      24        42  SORT Expr1007, Expr1006
      24        42  PROJECT CONVERT_IMPLICIT(int,Expr1015,0) AS Expr1007
      24        42  AGGREGATE COUNT(*) AS Expr1015 GROUP BY HASH Expr1006
  150000    150000  PROJECT CASE WHEN Expr1006 IS NULL THEN 0 ELSE Expr1006 END AS Expr1006
  150000    150000  LEFT OUTER JOIN HASH ON c_custkey = o_custkey
   96548     99996  │└PROJECT CONVERT_IMPLICIT(int,Expr1014,0) AS Expr1006
   96548     99996  AGGREGATE COUNT(*) AS Expr1014 GROUP BY HASH o_custkey
 1500000   1499959  TABLE SCAN orders WHERE  NOT o_comment LIKE ' % special % requests % '
  150000    150000  TABLE SCAN customer

SQL Server realises that the number of distinct customers in orders is in fact smaller than the total number of customers in customer (ratio: 2:3). That means that if you first aggregate all o_custkey together, you actually create a smaller output of orders than you get from customer. You can then build a hash table over the aggregate value. Aggregation before join!

When you do this, the entire RIGHT JOIN strategy is not needed and we can just use the slightly cheaper LEFT JOIN, which consumes less memory too.

This form of advanced reasoning is still rare in query optimisers.

Summary

In today's short analysis of Q13 from TPC-H we learned:

Most query optimisers, despite these tricks having been known for over 40 years, still don't apply the optimisations.