The lifecycle of a sharded Postgres query

Planning

By now we've done a significant amount of work and all we’ve accomplished is turning a SQL query into an abstract syntax tree. We’ve yet to consider sharding, yet to touch Postgres, and yet to answer the question "how do we execute this query fast?"

The thing that makes a sharded database system good is how well engineered the query planner is. Strap in, you’re about to find out just how much work went into making Neki’s do as little work as possible. It’s downright lazy.

Let’s reconsider the original query we’ve been following the journey of:

SELECT customers.name, orders.total, orders.created_at
FROM customers
JOIN orders ON orders.customer_id = customers.id
WHERE orders.created_at >= $1;

If we were on a single node of Postgres, what would it take to execute this? No shards, proxies, or query routers (for now). Postgres has its own query planner which, after getting the AST from the parser, must decide things like:

• Using internal statistics, estimate how many orders match created_at >= $1.
• Decide how to read each table: sequential, index, bitmap, or index-only scan.
• Choose the join direction and whether to use a nested loop, hash join, or merge join.
• Check whether an index on orders.created_at would help filter the orders.

These are just some of the decisions needed for our simple query. More complex queries give the planner more to consider, including additional joins, aggregations, subqueries, and sorting.

Recall that since this is a sharded database, all the rows from each of our tables are spread across shards. In this case we have 4 shards, but the principles we’re going through in this section apply whether there are 4 shards storing a terabyte or 400 shards storing a petabyte.

To decide which shard gets each row, we designate a column as the shard key. Earlier, we chose customers.id for customers and orders.id for orders. Let's revisit the visual from earlier.

That decision will soon become a problem.

This layout allows for the row of a customer to reside on one shard, while the corresponding orders for that single customer are spread across many. How are we supposed to join them now?

A sharded query planner must be built to take any SQL query and construct the most optimal plan for executing across many Postgres nodes.

Building the plan

The result of the query planner is yet another tree. The planner’s job is to convert an AST describing the query's request into a query plan describing the full sequence of operations the cluster must complete to compute the results.

To construct this, the planner within the router must be aware of both (a) the full Postgres schema and (b) the rules for data distribution across shards.

It gets (a) from the database’s authoritative shard, the shard designated as the source of schema metadata for that database. The router gets the table names, columns, and types from here. It does not request this info from the authoritative shard for every query, but maintains a cache and communicates with it regularly for updates.

It gets (b) from the data topology, stored centrally in etcd and cached on every router node.

Using this information, the router resolves the columns in our query and their types, a step called semantic analysis. It also looks for potential shortcuts that could avoid distributed planning, but our query is not so lucky.

How to join?

Next up: Take the AST and turn it into a query plan tree. The first step is to transform this into a rudimentary plan, which it will refine in several phases. The base plan produced from the original query looks like:

This is pretty simple. JoinCluster groups our two tables and their inner join condition. Inner joins can be reordered, giving the planner different ways to approach the same query. OutputClauses describes what to do with the rows we find. Here, we just need three columns back. Other queries might need sorting, grouping, aggregation, LIMIT, or even DISTINCT.

The planner still has to decide how to join those rows, which is the harder part.

Joining across shards

The planner uses our data topology to determine which parts of the query can run on the shards. We didn't show it earlier, but we specify this routing map in a data topology JSON file. For our two tables, each sharded by a hash of its own id, it looks like this (using postgres as the database name):

{
  "authoritative_shard_group": "metadata",
  "shard_indexes": {
    "xxhash_id": {
      "type": "xxhash",
      "columns": ["id"]
    }
  },
  "shard_groups": [
    {
      "uid": "metadata",
      "key_ranges": [{ "shard_uid": "shard1" }]
    },
    {
      "uid": "main_shards",
      "default_shard_index": "xxhash_id",
      "key_ranges": [
        { "shard_uid": "shard1", "end": "40" },
        { "shard_uid": "shard2", "start": "40", "end": "80" },
        { "shard_uid": "shard3", "start": "80", "end": "c0" },
        { "shard_uid": "shard4", "start": "c0" }
      ]
    }
  ],
  "databases": {
    "postgres": {
      "schemas": {
        "public": {
          "tables": {
            "customers": { "shard_group": "main_shards" },
            "orders": { "shard_group": "main_shards" }
          }
        }
      }
    }
  }
}

This configuration tells the routers that the customers and orders tables are each to be spread out across the four shards in the main_shards group, and each is sharded by a hash of its id column. When storing a new row, the router:

• Extracts its id.
• Runs it through a deterministic xxhash function.
• Uses the resulting hash to choose a shard, based on the provided ranges.

Each of our four shards is responsible for a different range of hash values.

We said earlier that our choice of sharding keys would make things more complex. Here’s the first consequence. A customer’s orders can be spread across all four shards, while the customer row lives on just one. To join them, we have to bring the matching rows together in the router.

Our planner is smart, though, and begins updating the JoinCluster node by turning it into a plan for retrieving and joining the rows. For this query, it needs to separate the customers and orders work, identify filters each shard can apply, and choose how the router will match the results.

It starts by creating a Route plan node for each table. The work inside each route becomes a SQL query for Postgres to execute. The route also specifies how to choose the destination shards, so a single route can send its query to one shard or several.

We now know we will need to send scatter-gather queries to fetch the necessary rows both for the customers and orders tables. However, there are several approaches to take for this join. Should we fetch customers first or orders first? Should we filter orders at the shard layer, or the router? Should the join be completed in a nested loop or with a hash table?

That's the next step of the planning process.

Joining across many Postgres instances

To fulfill JOIN orders ON orders.customer_id = customers.id, the router must match qualifying orders with their customers. There are several ways to approach this.

A nested-loop join is a concept that may be familiar to database internals enthusiasts. For everyone else, the idea is simple: take rows from one input and, for each one, search the other input for matches.

The router would first send a query to every shard to gather all customers rows. As rows start coming in, the router loops through each customer, one-by-one. For each, it makes a scatter-gather query to all shards, collecting all orders for that customer. Even a customer with no qualifying orders costs us four shard queries to find that out.

This is inefficient. There are better ways.

Hash join is another join technique which will suit our needs much better. Here, the idea is to build a lookup table from one input, then use it to find matches as rows arrive from the other. Similarly, the router will send out a request to fetch all customers rows upfront, but this time build a hash map in the router’s memory, keyed by customer ID. We can then do a single, large scatter-gather, asking for all orders rows created on or after September 9th.

Keeping the customer hash table in memory uses resources, but lets us match orders as they arrive without sending another query for each customer. The join rule compares the estimated costs of these approaches. Both perform the matching in the router, using rows supplied by Postgres on the shards.

The planner also considers reversing the inputs. customers first isn’t a requirement of either type of join.

These aren’t the planner’s only options. It also supports merge joins, batched nested-loop joins, and more. Its job is to find the most efficient plan for the query using the information available.

Choosing for this query

Which table the planner chooses to fetch first, and which join methodology to use, depends significantly on the table sizes.

Suppose we have 100,000 customers and 1,000,000 orders across the four shards. Using its default estimate for the date inequality, it expects roughly one third of the orders to qualify, about 300,000.

The Picasso diagram below visualizes under what scenarios the planner would choose various join algorithms and orderings. Change the row counts to see how the planner’s choice changes.

Let’s walk through why it chooses a hash join for these row counts.

With the nested loop approach we just described, that would mean 100,000 customer lookups (all of them) and then for each one, a scatter-gather across orders. That means fetching approximately 400,000 rows (100,000 customers and 300,000 orders), but spread across thousands of individual queries between the router and shards.

The hash join would fetch customers with four shard queries and qualifying orders with another four. Those eight queries would still transfer an estimated 400,000 rows into the router, including customers we might never match. This technique requires the additional step of building the hash table in memory.

The planner also tracks the number of distinct values in indexed columns, helping it estimate how many rows an equality filter or join will match. Using those estimates, it weighs the number of shard requests, rows transferred, rows processed, and rows held in memory. For these row counts, fetching the customer input and keeping it available for local lookups has a much lower estimated cost than making the repeated requests.

There’s another choice for the planner to make here. It could build the hash table from either input. The planner estimates a smaller memory requirement for the 100,000 customer rows than for 300,000 qualifying orders, so it chooses customers.

With those choices made, we can turn our initial plan into the final query plan.

The customers route builds the hash table, and the orders route supplies the rows to probe it. Both routes are scatter routes, meaning each sends its query to all four shards. Neither has a condition that narrows its request to particular shard-key values.

But what happens if that hash table doesn’t fit in memory?

Memory is really good and fast until you run out of it. If the hash table exceeds its memory budget, the router spills to disk. It partitions customers and orders by their join keys and joins corresponding partitions one at a time.

Whether the hash table fits in memory or spills to disk, the join still happens in the router. We now have a plan. Let’s execute it.