If you suspect that you are near your top load for the service or product, you can simply move to a more powerful server. This may not be the best long-time scaling solution if you are still in the middle, or even in the beginning of your growth. You will run out of bigger machines to buy long before you are done. Servers also become disproportionately more expensive as the size increases, and you will be left with at least one different, and thus not easily replaceable, server once you implement a proper scaling solution.

On the other hand, this will work for some time and is often the easiest way to get some headroom while implementing real scaling solutions.

Master-slave replication, either trigger-based or WAL-based, works reasonably well in cases where the large majority of the database accesses are reads. Some things that fall under this case, are website content managers, blogs, and other publishing systems.

As our chat system has more or less a 1:1 ratio of writes and reads, moving reads to a separate server will buy us nothing. The replication itself is more expensive than the possible win from reading from a second server.

Multi master replication is even worse than master-slave(s) when the problem is scaling a write-heavy workload. It has all the problems of master-slave, plus it introduces extra load via cross-partition locking or conflict resolution requirements, which further slows down the whole cluster.

If we split the data, we need a simple and efficient way to determine which server stores each data row. If the data had an integer primary key, you could just go round-robin, store the first row on the first server, the second row on the second, and so on. This would give you a fairly even distribution, even when rows with certain IDs are missing.

The partitioning function for selecting between four servers would be simply:

partition_nr = id & 3

The partitioning mask 3 (binary 11) is for the first two bits. For eight partitions, you would use 7 (binary 111), and for 64 servers it would be 63 (00111111). It is not as easy with things like usernames, where putting all names starting with an A first, B second, and so on, does not produce an even distribution.

Turning the username into a fairly evenly distributed integer via the hash function solves this problem and can be used directly to select the partition.

partition_nr = hashtext(username) & 3

This would distribute the users in the following manner:

hannu=# SELECT username, hashtext(username) & 3 as partition_nr FROM user_info; 
-[ RECORD 1 ]+-------- 
username     | bob 
partition_nr | 1 
-[ RECORD 2 ]+-------- 
username     | jane 
partition_nr | 2 
-[ RECORD 3 ]+-------- 
username     | tom 
partition_nr | 1 
-[ RECORD 4 ]+-------- 
username     | mary 
partition_nr | 3 
-[ RECORD 5 ]+-------- 
username     | jill 
partition_nr | 2 
-[ RECORD 6 ]+-------- 
username     | abigail 
partition_nr | 3 
-[ RECORD 7 ]+-------- 
username     | ted 
partition_nr | 3 
-[ RECORD 8 ]+-------- 
username     | alfonso 
partition_nr | 0 

So, partition 0 gets user alfonso, partition 1 bob and tom, partition 2 jane and jill, and partition 3 gets mary, abigail, and ted. The distribution is not exactly one fourth to each partition; but as the number of partitions increase, it will be pretty close where this actually matters.

If we had no PL/Proxy language, we could write the partitioning functions in the most untrusted PL languages. For example, a simple login proxy function written in PL/Pythonu looks like this:

CREATE OR REPLACE FUNCTION pylogin(
    IN i_username text, IN i_pwdhash text,
    OUT status int, OUT message text ) 
AS $$
    import psycopg2
    partitions = [
        'dbname=chap10p0 port=5433',
        'dbname=chap10p1 port=5433',
        'dbname=chap10p2 port=5433',
        'dbname=chap10p3 port=5433',
    ]
    partition_nr = hash(i_username) & 3
    con = psycopg2.connect(partitions[partition_nr])
    cur = con.cursor()
    cur.execute('select * from login(%s,%s)', ( i_username, i_pwdhash))
    status, message = cur.fetchone()
    return (status, message)
$$ LANGUAGE plpythonu SECURITY DEFINER;

Here, we defined a set of four partition databases, given by their connect strings and stored as a list in variable partitions.

When executing the function, we first evaluate the hash function on the username argument (hash(i_username)) and extract two bits from it (& 3) to get an index into the partitions' list (the partition number) for executing each call.

Then, we open a connection to a partition database using the connect string selected by the partition number (con=psycopg2.connect(partitions[partition_nr])).

Finally, we execute a remote query in the partition database and return the results of this to the caller of this proxy function.

This works reasonably well, if implemented like this, but it also has at least two places where it is suboptimal:

The performance problem can be solved by caching the open connections, and the maintenance problem can be solved by having a single function returning the partition information. However, even when we do these changes and stay with PL/Pythonu for partitioning, we will still be doing a lot of copy and paste programming in each of our proxy functions.

Once we had reached the preceding conclusions, when growing our database systems at Skype, the next logical step was quite obvious. We needed a special partitioning language, which would do just this one thing—calling remote SQL functions, and then make it as fast as possible. And thus, the PL/Proxy database partitioning language was born.

If you are on Debian, Ubuntu, or a Red Hat variant, installing the language is easy.

First, you have to install the required packages on your operating system:

sudo apt-get install postgresql-9.4-plproxy 

Or:

sudo yum install plproxy94 

If you need to install PL/Proxy from the source, you can download it from http://pgfoundry.org/projects/plprox, extract the sources in the contrib folder of your PostgreSQL source tree and run make and make install.

To install PL/Proxy you should run the plproxy.sql file, which is part of the source code or the package you installed.

The PL/Proxy language itself is very simple. The purpose of a PL/Proxy function is to hand off the processing to another server, so that the language only needs six statements:

The first group of statements handles the remote connectivity to the partitions. The help determines which database to run the query on. You specify the exact partition to run the query using CONNECT:

CONNECT 'connect string' ;

Here, connect string determines the database to run. connect string is the standard PostgreSQL connect string you would use to connect to the database from a client application, for example: dbname=p0 port=5433 username=test host=localhost.

Or, you can specify a name using CLUSTER:

CLUSTER 'usercluster';

Finally, you can specify a partition number using RUN ON:

RUN ON part_func(arg[, ...]) ; 

part_func()can be any existing or user-defined PostgreSQL function returning an integer. PL/Proxy calls that function with the given arguments and then uses N lower bits from the result to select a connection to a cluster partition.

There are two more versions of the RUN ON statement:

RUN ON ANY;

This means that the function can be executed on any partition in a cluster. This can be used when all the required data for a function is present on all partitions.

The other version is:

RUN ON ALL;

This runs the statement on all partitions in parallel and then returns a concatenation of results from the partitions. This has at least three main uses:

The default behavior of a PL/Proxy function, if no SELECT or TARGET is present, is to call the function with the exact same signature as itself in the remote partition.

Suppose we have the function:

CREATE OR REPLACE FUNCTION login(
    IN i_username text, IN i_pwdhash text,
    OUT status int, OUT message text ) 
AS $$
    CONNECT 'dbname=chap10 host=10.10.10.1';
$$ LANGUAGE plproxy SECURITY DEFINER;

If it is defined in schema public, the following call select * from login('bob', 'secret') connects to the database chap10 on host 10.10.10.1 and runs the following SQL statement there:

SELECT * FROM public.login('bob', 'secret')

This retrieves the result and returns it to its caller.

If you don't want to define a function inside the remote database, you can substitute the default select * from <thisfunction>(<arg1>, ...) call with your own by writing it in the function body of the PL/Proxy function:

CREATE OR REPLACE FUNCTION get_user_email(i_username text) 
RETURNS SETOF text AS $$
    CONNECT 'dbname=chap10 host=10.10.10.1';
    SELECT email FROM user_info where username = i_username;
$$ LANGUAGE plproxy SECURITY DEFINER;

Only a single SELECT is supported; for any other, or more complex SQL statements, you have to write a remote function and call it.

The third option, is to still call a function similar to itself, but named differently. For example, if you have a proxy function defined not in a separate proxy database, but in a partition, you may want it to target the local database for some data:

CREATE OR REPLACE FUNCTION public.get_user_email(i_username text) RETURNS SETOF text AS $$
    CLUSTER 'messaging';
    
    RUN ON hashtext(i_username);
    TARGET local.get_user_email;
$$ LANGUAGE plproxy SECURITY DEFINER;

In this setup, the local version of get_user_email()is in schema local on all partitions. Therefore, if one of the partitions connects back to the same database that it is defined in, it avoids circular calling.

The last PL/Proxy statement is for cases where you want some bigger chunk of work to be done in appropriate partitions.

For example, if you have a function to create several users in one call and you still want to be able to use it after partitioning, the SPLIT statement is a way to tell PL/Proxy to split the arrays between the partitions based on the partitioning function:

CREATE or REPLACE FUNCTION create_new_users(
    IN i_username text[], IN i_pwdhash text[], IN i_email text[],
    OUT status int, OUT message text )  RETURNS SETOF RECORD
AS $$
BEGIN
  FOR i IN 1..array_length(i_username,1) LOOP
      SELECT *
        INTO status, message
        FROM new_user(i_username[i], i_pwdhash[i], i_email[i]);
      RETURN NEXT;
  END LOOP;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

The following PL/Proxy function definition, created on the proxy database, can be used to split the calls across the partitions:

CREATE or REPLACE FUNCTION create_new_users(
    IN i_username text[], IN i_pwdhash text[], IN i_email text[],
    OUT status int, OUT message text )  RETURNS SETOF RECORD
AS $$
  CLUSTER 'messaging';
  RUN ON hashtext(i_username);
  SPLIT  i_username,  i_pwdhash,  i_email;
$$ LANGUAGE plproxy SECURITY DEFINER;

It would be called by sending in three arrays to the function:

SELECT * FROM create_new_users(
    ARRAY['bob', 'jane', 'tom'],
    ARRAY[md5('bobs_pwd'), md5('janes_pwd'), md5('toms_pwd')],
    ARRAY['[email protected]', '[email protected]', '[email protected]']
);

It will result in two parallel calls to partitions 1 and 2 (as using hashtext(i_username) bob and tom map to partition 1 and mary to partition 2 of the total for the partitions, as explained earlier), with the following arguments for partition 1:

SELECT * FROM create_new_users(
    ARRAY['bob', 'tom'],
    ARRAY['6c6e5b564fb0b192f66b2a0a60c751bb','edcc36c33f7529f430a1bc6eb7191dfe'],
    ARRAY['[email protected]','[email protected]']
);

And this for partition 2:

SELECT * FROM create_new_users(
    ARRAY['jane'],
    ARRAY['cbbf391d3ef4c60afd851d851bda2dc8'],
    ARRAY['[email protected]']
);

Then, it returns a concatenation of the results:

status | message 
--------+--------- 
    200 | OK 
    200 | OK 
    200 | OK 
(3 rows) 

First, what is a cluster in PL/Proxy? Well, the cluster is a set of partitions that make up the whole database. Each cluster consists of a number of partitions, as determined by the cluster configuration. Each partition is uniquely specified by its connect string. The list of connection strings is what makes up a cluster. The position of the partition in this list is what determines the partition number, so the first element in the list is partition 0, the second partition is 1, and so on.

The partition is selected by the output of the RUN ON function, and then masked by the right number of bits to map it on the partitions. So, if hashtext(i_username) returns 14 and there are four partitions (2 bits, mask binary 11 or 3 in decimal), the partition number will be 14 and 3 = 2, and the function will be called on partition 2 (starting from zero), which is the third element in the partition list.

The actual configuration of the cluster, the definition of partitions, can be done in two ways, either by using a set of functions in schema plproxy, or you can take advantage of the SQL/MED connection management (SQL/MED is available starting at PostgreSQL 8.4 and above). You can read more about SQL/MED here https://wiki.postgresql.org/wiki/SQL/MED

This is the original way to configure PL/Proxy, which works on all versions of PostgreSQL. When a query needs to be forwarded to a remote database, the function plproxy.get_cluster_partitions(cluster) is invoked by PL/Proxy to get the connection string to use for each partition.

The following function is an example, which returns information for a cluster with four partitions, p0 to p3:

CREATE OR REPLACE FUNCTION plproxy.get_cluster_partitions(cluster_name text) 
RETURNS SETOF text AS $$ 
BEGIN 
    IF cluster_name = 'messaging' THEN 
        RETURN NEXT 'dbname=p0'; 
        RETURN NEXT 'dbname=p1'; 
        RETURN NEXT 'dbname=p2'; 
        RETURN NEXT 'dbname=p3'; 
    ELSE
        RAISE EXCEPTION 'Unknown cluster'; 
    END IF; 
END; 
$$ LANGUAGE plpgsql; 

A production application might query some configuration tables, or even read some configuration files to return the connection strings. Once again, the number of partitions returned must be a power of two. If you are absolutely sure that some partitions are never used, you can return empty strings for these.

We also need to define a plproxy.get_cluster_version(cluster_name) function. This is called on each request and if the cluster version has not changed, the output from a cached result from plproxy.get_cluster_partitions can be reused. So, it is best to make sure that this function is as fast as possible:

CREATE OR REPLACE FUNCTION plproxy.get_cluster_version(cluster_name text) 
RETURNS int4 AS $$ 
BEGIN 
    IF cluster_name = 'messaging' THEN 
        RETURN 1; 
    ELSE
        RAISE EXCEPTION 'Unknown cluster'; 
    END IF; 
END; 
$$ LANGUAGE plpgsql; 

The last function needed is plproxy.get_cluster_config, which enables you to configure the behavior of PL/Proxy. This sample will set the connection lifetime to 10 minutes:

CREATE OR REPLACE FUNCTION plproxy.get_cluster_config( 
    in cluster_name text, 
    out key text, 
    out val text) 
RETURNS SETOF record AS $$ 
BEGIN 
    -- lets use same config for all clusters 
    key := 'connection_lifetime'; 
    val := 10*60; 
    RETURN NEXT; 
    RETURN; 
END; 
$$ LANGUAGE plpgsql; 

Since version 8.4, PostgreSQL has support for an SQL standard for management of external data, usually referred to as SQL/MED. SQL/MED is simply a standard way to access a database. Using functions to configure partitions is arguably insecure, as any caller of plproxy.get_cluster_partitions() can learn connection strings for partitions that may contain sensitive info like passwords. PL/Proxy also provides a way to do the cluster configuration using SQL/MED, which follows the standard SQL security practices.

The same configuration, as discussed earlier, when done using SQL/MED, is as follows: