Walking with the Elephants: partition-wise join

Showing posts with label partition-wise join. Show all posts

Tuesday, August 8, 2023

Partitioning as a query optimization strategy?

I had discussed about query optimization techniques applicable to queries involving partitioned tables in PGConf.India 2018 (Video recording, slides). (My previous blog discusses these techniques in detail.) The takeaway from that presentation was these query optimization techniques improved query performance if the tables were already partitioned. But partitioning wasn't good enough as a query optimization strategy by itself even when partitionwise join, partitionwise aggregate and parallel query were all used together on small data-sizes. Experiments then hinted that if the data was large enough partitioning would become a query optimization strategy. But we didn't know how large is large enough. Experiments to establish would require beefy machines with larger resources which were costly, took long time to procure or get access to. On top of them it took long time to setup and finish the runs. At one point we stopped experimenting. Fast forward to today and things have changed drastically, thanks to the cloud!

EDB's BigAnimal comes to help

EnterpriseDB offers PostgreSQL-as-a-service in the form of a DBAAS platform called BigAnimal. It allows its users to deploy and run PostgreSQL in cloud on hardware configuration of their choice. It also provides a starter free credit to try out this platform. I experimented with very large datasets by using BigAnimal. I ran the experiments on PostgreSQL 15 hosted on a m5.4xlarge instance (64 GB RAM, 16 vCPUs) with 1500 GB storage. All of this without wasting much time and also money; I destroyed the instance as soon as my experiments were over.

Experiment

I wanted to see the impact of only partitioning as a query optimization strategy. So instead of using whole TPCH setup, I crafted a micro-benchmark with two queries involving two tables li and ord modeled after lineitem and orders tables in TPCH benchmark. When partitioned each of these two tables have matching 1000 partitions each. The tables have following schema

When partitioned they are partitioned by range on key1. Each row in ord has 3 matching rows in li, roughly imitating the data-size ratio between corresponding tables in TPCH benchmark.

Query 1 which extracts relevant parts of TPCH Q3 or Q4 looks like

select count(*)
    from (select o.key1, sum(o.m) revenue, o.d
            from li l, ord o
            where l.key1 = o.key1 and
                o.d > current_date + 300 and
                l.d < current_date + 700
            group by o.key1, o.d
            order by revenue, o.d
    ) as t1

Query 2 which is a pure join between li and ord looks like

select o.key1, l.key1, o.d
            from li l, ord o
            where l.key1 = o.key1 and
                o.d > current_date + 300 and
                l.d < current_date + 700

The time required to execute these two queries is measured using EXPLAIN ANALYZE. We varied the number of rows per partition as well as the number of partitions.

The execution times for queries are given in tables below.

Table 1: 10K rows per partition

	Average execution time Q1 (ms)			Average execution time Q2 (ms)
No. of partitions	unpartitioned table	partitioned table without PWJ	partitioned table with PWJ	unpartitioned table	partitioned table without PWJ	partitioned table with PWJ
5	83.05	93.29	53.68	48.83	60.55	50.85
10	195.87	221.33	90.24	104.40	129.06	105.20
50	1,183.25	1,487.00	432.07	584.31	723.90	527.97
100	2,360.19	3,001.81	888.46	1,342.69	1,595.53	1,053.91
500	11,968.68	15,220.69	4,350.62	6,903.91	8,082.09	5,381.46
1000	33,772.31	31,090.57	8,847.61	16,461.44	17,646.42	10,875.05

Table 2: 100K rows per partition

	Average execution time Q1 (ms)			Average execution time Q2 (ms)
No. of partitions	unpartitioned table	partitioned table without PWJ	partitioned table with PWJ	unpartitioned table	partitioned table without PWJ	partitioned table with PWJ
5	1,157.23	1,489.53	514.68	609.81	773.31	582.07
10	2,326.40	2,990.32	1,041.11	1,375.69	1,597.55	1,152.33
50	11,899.34	15,181.49	4,792.88	7,196.35	8,446.64	5,828.54
100	24,139.10	30,660.87	9,594.33	14,277.53	16,753.36	11,512.35
500	1,53,922.35	1,65,550.06	50,308.85	74,387.34	85,175.79	58,282.17
1000	3,13,534.59	3,38,718.63	1,31,482.31	2,03,569.14	1,32,688.60	1,23,643.18

Same numbers in the form of graphs are better to understand. Next we see graphs depicting the average execution time of each of these queries varying with the number of partitions. In each graph Y-axis shows the execution times in logarithmic scale, X-axis shows the number of partitions. Blue line shows the query execution times when tables are not partitioned. Red line shows query execution times when tables are partitioned but partitionwise join and aggregation are not used (turning both enable_partitionwise_join and enable_partitionwise_aggregate OFF). Yellow line shows query execution times when tables are partitioned and partitionwise join and partitionwise aggregate is used.

Note that the Y-axis denoting the execution time is drawn with logarithmic scale. Thus the linear difference on that axis shows improvement in integer multiples instead of fractions. For example, Q1's execution time improves almost by 4 times when tables are partitioned and partitionwise join and aggregate are enabled.

Graph 1

Graph 2

Graph 3

Graph 4

Key takeaways

The graphs above make it clear that when datasizes are very large partitioning can also be used as a query optimization technique along with its other advantages. I will share some key points here

When the total data size reaches the house of millions, partitioning can be considered as a query optimization strategy. The exact number of partitions and average rows per partition do not make much difference. We see similar performance whether 5M rows are divided into 500 partitions or 50 partitions.
The exact thresholds depend upon properties of data and queries. E.g. size of each rows, columns used in query, operations performed by the query etc.
Since these optimization techniques are very much dependent upon the partition key, choosing the right partition key is very important.
When tables are partitioned, queries perform better when partitionwise operations are used irrespective of the datasize.

Each workload is different. Above charts provide some guidance. But experimenting with the size and number of partitions as well as the partition key is important to know whether partitioning will help you optimize queries in your application or not. Experimentation shouldn't be an issue anymore. EDB's BigAnimal platform allows its users to experiment quickly without requiring a large upfront investment.

Wednesday, January 26, 2022

Advanced partition matching for partition-wise join

Earlier I had written a blog about partition-wise join in PostgreSQL. In that blog I had talked about an advanced partition matching technique which will allow partition-wise join to be used in more cases. In this blog we will discuss this technique in detail. I will suggest to read my blog on basic partition-wise join again to get familiar with the technique.

Basic partition matching technique allows a join between two partitioned tables to be performed using partition-wise join technique if the two partitioned tables had exactly same partitions (more precisely exactly matching partition bounds). For example consider two partitioned tables prt1 and prt2

\d+ prt1
... [output clipped]
Partition key: RANGE (a)
Partitions: prt1_p1 FOR VALUES FROM (0) TO (5000),
prt1_p2 FOR VALUES FROM (5000) TO (15000),
prt1_p3 FOR VALUES FROM (15000) TO (30000)

and

\d+ prt2

... [ output clipped ]

Partition key: RANGE (b)

Partitions: prt2_p1 FOR VALUES FROM (0) TO (5000),

prt2_p2 FOR VALUES FROM (5000) TO (15000),

prt2_p3 FOR VALUES FROM (15000) TO (30000)

A join between prt1 and prt2 is executed as join between matching partitions i.e. prt1_p1 joins prt2_p1, prt1_p2 joins prt2_p2 and prt1_p3 joins prt2_p3. This has many advantages as discussed in my previous blog.

But basic partition matching can not join two partition tables with different partitions (more precisely different partition bounds). For example, in the above case if prt1 an extra partition prt1_p4 FOR VALUES FROM (30000) TO (50000), a join between prt1 and prt2 would not use partition-wise join. Many applications use partitions to segregate actively used and stale data, a technique I discussed in my another blog. The stale data is eventually removed by dropping partitions. New partitions are created to accommodate fresh data. Let's say the partition scheme of two such tables is such that they usually have matching partitions. But when an active partition gets added to one of these tables or a stale one gets deleted, they will have mismatched partitions for a small duration. We don't want a join hitting the database during this small duration to perform bad since it can not use partition-wise join. Advanced partition matching algorithm helps here.

Advanced partition matching algorithm

Advanced partition matching is very much similar to the merge join algorithm. It takes the sorted partition bounds and finds matching partitions by comparing the bounds from both the tables in their sorted order. Any two partitions, one from either partitioned table, whose bounds match exactly or overlap are considered to be joining partners since they may contain rows that join. Continuing with the above example:

\d+ prt1

... [output clipped]

Partition key: RANGE (a)

Partitions: prt1_p1 FOR VALUES FROM (0) TO (5000),

prt1_p2 FOR VALUES FROM (5000) TO (15000),

prt1_p3 FOR VALUES FROM (15000) TO (30000)

and

\d+ prt2

... [ output clipped ]

Partition key: RANGE (b)

Partitions: prt2_p1 FOR VALUES FROM (0) TO (5000),

prt2_p2 FOR VALUES FROM (5000) TO (15000),

prt2_p3 FOR VALUES FROM (15000) TO (30000),

prt1_p4 FOR VALUES FROM (30000) TO (50000)

Similar to the basic partition matching algorithm this will join prt1_p1 and prt2_p1, prt1_p2 and prt2_p2, and prt1_p3 and prt2_p3. But unlike basic partition matching it will also know that prt1_p4 does not have any join partner in prt1. Thus if the join between prt1 and prt2 is INNER join or when prt2 is INNER relation of join, the join will contain only three joins leaving prt2_p4 aside. In PostgreSQL, a join where prt2 is OUTER relation, we won't be able to use partition-wise join even if We will come back to this again when we will discuss outer joins further.

This is simple right, but consider another example of listed partitioned tables

\d+ plt1_adv

Partition key: LIST (c)

Partitions: plt1_adv_p1 FOR VALUES IN ('0001', '0003'),

plt1_adv_p2 FOR VALUES IN ('0004', '0006'),

plt1_adv_p3 FOR VALUES IN ('0008', '0009')

and

\d+ plt2_adv

Partition key: LIST (c)

Partitions: plt2_adv_p1 FOR VALUES IN ('0002', '0003'),

plt2_adv_p2 FOR VALUES IN ('0004', '0006'),

plt2_adv_p3 FOR VALUES IN ('0007', '0009')

Observe that there are exactly three partitions in both the relations but lists corresponding plt1_adv_p2 match exactly that of plt2_adv_p2 but other two partitions do not have exactly matching lists. Advanced partition matching algorithm helps to determine that plt1_adv_p1 and plt2_adv_p1 have overlapping lists and their lists do not overlap with any other partition from the other relation. Similarly for plt1_adv_p3 and plt2_adv_p3. Thus it allows join between plt1_adv and plt2_adv to be executed as partition wise join by joining their matching partitions. The algorithm can find matching partitions in even more complex partition bound sets.

The problem with outer joins

Outer joins pose a particular problem in PostgreSQL world. Consider again the example of join between prt2 LEFT JOIN prt1. prt2_p4 does not have a joining partner in prt1 and yet the rows in that partition will be part of the join since it is an outer relation, albeit with the columns from prt1 all "null"ed. Usually in PostgreSQL when the INNER side is empty, it's represented by a "dummy" relation which emits no rows but still knows the schema of that relation. Without partition-wise join a "concrete" relation which has some presence in the original query turns dummy and thus planner has "something" to join the outer relation with. So PostgreSQL's planner doesn't have to do anything extra when such outer joins occur. But when there is no matching inner partition for an outer partition e.g. prt2_p4, there is "no entity" which can represent the "dummy" inner side of that outer join. PostgreSQL does not have a way right now to induce such "dummy" relations during planning right now. But that's not required. Ideally such a join with empty inner only requires schema of the inner relation and not an entire relation itself. Once we build support to execute such a join with a solid outer relation and schema of inner relation, we will be able to tackle partition-wise join where there are no matching partitions on inner side. Hopefully we will solve that problem some time soon.

When there is no matching partition on the outer side of the join, the inner partition does not contribute to the result of join and can be just ignored. So partition-wise joins where there are no matching partitions on the inner side are not a problem at all.

Multiple matching partitions

When the partitioned tables are such that multiple partitions from one side match one partition or more partitions on the other side, partition-wise join simply bails out since there is no way to induce an "Append" relation during planning time which represents two or more partitions together. Hopefully we will remove that limitation as well sometime.

Curious case of hash partitioned tables

It doesn't make much sense to use it for a hash partitioned table since usually partitions of two hash partitioned table using same modulo always match. When the modulo is different, the data from one a given partition of one table can find its join partners in all the partitions of the other, thus rendering partition-wise join ineffective.

Even with all these limitation, what we have today is a very useful solution which serves most of the practical cases. Needless to say that this feature works seemlessly with FDW join push down to adding to sharding capabilities that PostgreSQL already has!

Tuesday, December 19, 2017

Partition-wise joins: "divide and conquer" for joins between partitioned table

Unlike inheritance-based partitioning, declarative partitioning introduced in PostgreSQL 10 leaves nothing to infer about how the data is divided into partitions. PostgreSQL 11's query optimizer is gearing up to take advantage of this "no-inference" representation. The first one that got committed was basic partition-wise join.

What is partition-wise join?

A join between two similarly partitioned tables can be broken down into joins between their matching partitions if there exists an equi-join condition between the partition keys of the joining tables. The equi-join between partition keys implies that all the join partners for a given row in a given partition of one partitioned table must be in the corresponding partition of the other partitioned table. Because of this, the join between partitioned tables can be broken down into joins between the matching partitions. This technique of breaking down a join between partition tables into joins between their partitions is called partition-wise join.

Partition-wise join in PostgreSQL

Let's start with an example. Consider two tables partitioned as follows:

CREATE TABLE prt1 (a int, b int, c varchar) PARTITION BY RANGE(a);

CREATE TABLE prt1_p1 PARTITION OF prt1 FOR VALUES FROM (0) TO (5000);

CREATE TABLE prt1_p2 PARTITION OF prt1 FOR VALUES FROM (5000) TO (15000);

CREATE TABLE prt1_p3 PARTITION OF prt1 FOR VALUES FROM (15000) TO (30000);

CREATE TABLE prt2 (a int, b int, c varchar) PARTITION BY RANGE(b);

CREATE TABLE prt2_p1 PARTITION OF prt2 FOR VALUES FROM (0) TO (5000);

CREATE TABLE prt2_p2 PARTITION OF prt2 FOR VALUES FROM (5000) TO (15000);

CREATE TABLE prt2_p3 PARTITION OF prt2 FOR VALUES FROM (15000) TO (30000);

All join partners for a row in prt1_p1 come from prt2_p1. All join partners for a row in prt1_p2 come from prt2_p2. And all join partners for a row in prt1_p3 come from prt2_p3. Those three form the matching pairs of partitions.

Without partition-wise join, the plan for a join between these two tables looks like:

explain (costs off)

select * from prt1 t1, prt2 t2 where t1.a = t2.b and t1.b = 0 and t2.b between 0 and 10000;

QUERY PLAN

--------------------------------------------------------

Hash Join

Hash Cond: (t2.b = t1.a)

-> Append

-> Seq Scan on prt2_p1 t2

Filter: ((b >= 0) AND (b <= 10000))

-> Index Scan using prt2_p2_b on prt2_p2 t2_1

Index Cond: ((b >= 0) AND (b <= 10000))

-> Hash

-> Append

-> Seq Scan on prt1_p1 t1

Filter: (b = 0)

-> Seq Scan on prt1_p2 t1_1

Filter: (b = 0)

-> Seq Scan on prt1_p3 t1_2

Filter: (b = 0)

(15 rows)

With partition-wise join the plan for the same query looks like:

explain (costs off)

select * from prt1 t1, prt2 t2 where t1.a = t2.b and t1.b = 0 and t2.b between 0 and 10000;

QUERY PLAN

------------------------------------------------------------------------

Append

-> Hash Join

Hash Cond: (t2.b = t1.a)

-> Seq Scan on prt2_p1 t2

Filter: ((b >= 0) AND (b <= 10000))

-> Hash

-> Seq Scan on prt1_p1 t1

Filter: (b = 0)

-> Nested Loop

-> Seq Scan on prt1_p2 t1_1

Filter: (b = 0)

-> Index Scan using prt2_p2_b on prt2_p2 t2_1

Index Cond: ((b = t1_1.a) AND (b >= 0) AND (b <= 10000))

(13 rows)

There are a few things to be noted here:

There exists an equi-join condition t1.a = t2.b, which includes partition keys from both the tables.
Without partition-wise join, the join will be performed after "appending" all the rows from each partition of either partitioned table. With partition-wise join, the join between matching partitions is performed and the results are appended. This is advantageous when the size of join result is significantly smaller than the result of cross product. More advantageous, if the partitions themselves are foreign tables, i.e. data in the partitions resides on the foreign server.
Without partition-wise join, it used hash join, but with partition-wise join it used different strategies for each join between partitions, choosing optimal strategy for each join. For example, the join between prt1_p2 and prt2_p2 uses nested loop join with index scan on prt2_p2_b as parameterized inner side, whereas the other join uses hash join.
The condition t2.b between 0 and 10000 eliminated partition prt2_p3 so it does not get scanned by the plan without partition-wise join. But it didn't notice that no row in prt1_p3 had a join partner at all and still scanned that partition. With partition-wise join, it sensed the lack of a matching partition and eliminated a scan on prt1_p3 as well. Eliminating an entire partition is a significant improvement, since sequential scans are expensive.

Partition-wise join wins over unpartitioned join because it can take advantage of properties of the partitions and use smaller hash tables that may completely fit in memory, faster in-memory sorts, join push-down in case of foreign partitions and so on. We will talk about performance more in a follow-on post.

Beyond basic partition-wise join

In the basic version that was committed, the technique is applied when the joining tables have exactly the same partition key data types and have exactly matching partition bounds. But there are few enhancements possible:

Even if the partition bounds do not match exactly, the technique can be used when every partition in one partitioned table has at most one matching partition from the other partitioned table. A patch for that is being worked on.
A join between an unpartitioned table and a partitioned table can be executed using this technique by joining the unpartitioned table with each partition separately and combining the results of these joins. This may help the cases when a few tables in query are unpartitioned but other tables are similarly partitioned and an optimal plan interleaves partitioned and unpartitioned tables.
The technique uses more memory and CPU even when partition-wise join is not an optimal strategy. Reduce the memory and CPU footprint of this technique.
When two differently partitioned tables are joined, repartition one of them to match the partition scheme of the other and join using partition-wise join; a technique usually helpful to join differently sharded tables by redistributing the data.

Contributions

I authored the patch to implement basic partition-wise join, but a lot of people helped with testing and reviewing. Rajkumar Raghuvanshi, my colleague from EDB tested it. Another EDBean Rafia Sabih ran benchmarks and published results. Robert Haas committed it after several rounds of reviews. Thomas Munro, Jeevan Chalke from EDB, Amit Langote from NTT, Antonin Houska and few others also reviewed some portions of it.

Also read about this feature's role in analytics here.