Poor Man's Parallel Processing (PMPP)

PMPP is an extension for executing multiple statements in parallel. Those statements might be completely independent, or they might need to be further combined and aggregated. In either case, it is up to the user to decide how a larger query should be broken into smaller ones.

Common Strategies for using PMPP

Before delving into to specifics of functions and types provided by PMPP, it is helpful to first see some examples of problems solved by PMPP.

System information

Get a list of all tables in a given schema, and the row count of each of those tables. Show the largest tables first.

sql CREATE TEMPORARY TABLE table_counts(table_name text, row_count bigint); SELECT * FROM pmpp.distribute(null::table_counts, 'dbname=' || current_database(), array ( SELECT format('SELECT %L, count(*) from %I',t.table_name,t.table_name) FROM information_schema.tables t WHERE t.table_schema = 'public' )) ORDER BY row_count DESC; table_name | row_count ------------+----------- foo | 10 a | 0 (2 rows)

This could have been accomplished via a series of SELECT COUNT(*) from foo statements, but then those results would have had to be loaded into a manually created table before being re-queried.

Two-stage aggregation of partitions

Assume that there is a table user_views with partitions for a given number of months. If the following query is run: sql SELECT user_name, sum(view_count) as total_view_count FROM user_views GROUP BY 1 ORDER BY 2

This query will scan all of the partitions in sequence, and then aggregate the entire set.

sql SELECT user_name, sum(view_count) as total_view_count FROM pmpp.distribute(null::user_views, 'dbname=' || current_database(), array[ 'SELECT user_name, sum(view_count) FROM user_views_december GROUP BY 1', 'SELECT user_name, sum(view_count) FROM user_views_november GROUP BY 1', 'SELECT user_name, sum(view_count) FROM user_views_october GROUP BY 1', 'SELECT user_name, sum(view_count) FROM user_views_september GROUP BY 1' ]) GROUP BY 1 ORDER BY 2 DESC

With this query, the individual partial aggregate queries will be run independently, some simultaneously (depending on how many CPUs are available and any limiting parameters specified), and those results will then be returned by the set returning function pmpp.distribute() which can itself be queried to obtain the final aggregate.

Parallel index Building.

Build all indexes for a table at the same time, up to the number of CPUs on the db machine.

CREATE TABLE parallel_index_test( b integer, c integer, d integer); SELECT * FROM pmpp.meta('dbname=' || current_database(), array( SELECT format('create index on %s(%s)',c.table_name,c.column_name) FROM information_schema.columns c WHERE c.table_name = 'parallel_index_test' AND c.table_schema = 'public')) ORDER BY 1; command | result ----------------------------------------+-------- create index on parallel_index_test(b) | OK create index on parallel_index_test(c) | OK create index on parallel_index_test(d) | OK (3 rows)

Hiding passwords

Even if foreign data wrappers are not used, it is helpful to have postgresql_fdw installed. This makes it possible to leverage foreign server definitions and user mappings makes for cleaner function invocations.

```sql SELECT current_database() as mydb, current_user as me \gset CREATE EXTENSION postgres_fdw; CREATE SERVER loopback FOREIGN DATA WRAPPER postgres_fdw OPTIONS (host 'localhost', port '5432', dbname :'mydb'); CREATE USER MAPPING FOR :me SERVER loopback OPTIONS (user :'me', password 'REDACTED');

CREATE TEMPORARY TABLE my_result(x integer); SELECT * FROM pmpp.distribute(null::my_result,'loopback',array['SELECT 1','SELECT 2']);

x

1 2 (2 rows) ```

Clearly, this does not solve the issue of storing passwords, but it does isolate the issue to the pg_user_mappings table.

Functions

distribute() - single database

PMPP relies heavily on polymorphic functions returning SETOF anyelement. This allows the user to define the result set that a set of queries will return, rather than have a separate function for every possible result set. This comes at a cost, however, in that the database must have a existing type that matches the result set shape desired. This type can be an explicit CREATE TYPE, or it can be the type of an existing table. Often, the type is a temporary table created at runtime to match an ad-hoc query.

sql function distribute( row_type anyelement, connection text, sql_list text[], cpu_multiplier float default 1.0, num_workers integer default null, statement_timeout integer default null) returns setof anyelement

This is the simplest form of the distribute() function, geared more towards local parallelism rather than sharding.

Parameters:

  • row_type: a null value casted to the type or table which specifies the shape of the result set that will be returned by this function. The result sets of all queries in sql_list will be coerced to fit this result set, if possible. If not possible, then the function will fail.
  • connection: a text string that represents a valid connection_string to dblink(). It can be any valid connection string that could be passed to psql on the command line, or it can be the name of a foreign server already defined on the local database. It is passed as-is to dblink_connect, so if dblink can use it, it's valid.
  • sql_list: an array of text, each element being an executable statement, almost always a SELECT, though any statement that can be executed over dblink() is possible. Each element of the array must be valid SQL, and it must reference objects found on the connection connection. The result set generated needs to be coercable into row_type, though it should be noted that only column order and (approximate, coercable) type matter, column names do not matter.
  • sql_list: an array of text, each element being an executable statement, almost always a SELECT, though any statement that can be executed over dblink() is possible. Each element of the array must be valid SQL, and it must reference objects found on the connection connection. The result set generated needs to be coercable into row_type, though it should be noted that only column order and (approximate, coercable) type matter, column names do not matter.
  • cpu_multiplier (often omitted, defaults to 1.0): Specifies the number of connections to make to connection relative to the number of CPUs found on that machine, but never less than one or more than the number of array elements in queries. A value of 2.0 means to make two connections per CPU, 0.25 means one connection per four CPUs, etc. This value is trumped by num_workers. If this value is specified (and num_workers is not) then the connection connection must have PMPP installed.
  • num_workers (often omitted/null): If set, specifies the max number of connections to connection to make. PMPP will never make more connections than there are queries to run for that connection. Specifying this value overrides any value in cpu_multiplier. This value is requires if the database at connection does not have PMPP installed.
  • statement_timeout (often omitted/null): If set, each connection to connection has SET statement_timeout=<value> executed after the connection is established.

The number of connections created is never less than one, and never more than the number of queries in sql_list. When a query completes, its results are passed along to the output result set, and the function will give the connection another query, if any remain. When no more queries are available for distribution, the connection will be disconnected. When all connections have disconnected, the function returns the combined result set.

Example

``` CREATE TEMPORARY TABLE x(y integer); CREATE TABLE SELECT * FROM pmpp.distribute(null::x,'dbname=mydb',array['select 1','select 2','select 3 as z']);

y

1 2 3 (3 rows) ```

In this example, a table was defined, but never populated. Rather, it was defined because it matched the shape of the result sets we would expect from all of the queries in the array. It should be noted that the name of the columns do not matter, only the order and datatypes. A common usage is to query a partitioned table foo, specifying null::foo as the table type, and the queries in the array would be queries on the individual partitions of foo. The remote connection does not need to have the object x defined, only the local connection does.

The function connected as the superuser of a local database named mydb. Connecting as superuser is rarely ideal. The distribute() function connected to mydb, determined the number of available processors there (which means that mydb must also have PMPP installed - there are ways around that discussed in num_workers.

distribute() - query_manifest[]

sql function distribute(row_type anyelement, query_manifest query_manifest[]) returns setof anyelement

The type query_manifest contains the following attributes sql type query_manifest as ( connection text, queries text[], num_workers integer, cpu_multiplier float, statement_timeout integer );

Each of these attributes exactly matches the purpose described by the same-named parameters in the single database version of distribute().

For each query_manifest object in the array: 1. Connect to each of the databases with a number of connections equal to num_workers (if specified), or the number of cpus found on the machine at the end of that connection times the mulitplier (defaults to 1.0). 2. Each query specified must return a result set matching the type row_type. 3. The queries are distributed to the connections as they become available, and the results of those queries are in turn pushed to the result set, where they can be further manipulated like any table, view, or set returning function. 4. Returns a result set in the shape of row_type.

Example

CREATE TEMPORARY TABLE xy(x integer, y integer); SELECT * FROM pmpp.distribute( null::xy, array[ row('dbname=mydb', array['SELECT 1,2','SELECT 3,4'], null, 4, null)::pmpp.query_manifest, row('named_foreign_server', array['SELECT 5,6','SELECT 7,8'], 1.5, null, null)::pmpp.query_manifest, row('postgresql://user:secret@localhost', array['SELECT 9,10'], null, null, null)::pmpp.query_manifest ]) ORDER by x; x | y ---+---- 1 | 2 3 | 4 5 | 6 7 | 8 9 | 10 (1 row)

Clearly this isn't the easiest structure to manage, it's likely that if this form is used directly at all, it would be invoked from a parent function.

Notice the variety of connection strings used: * A superuser connection to a local db. * The name of a server defined in pg_foreign_server, which must also ahave a proper user mapping for the invoking user. * A qualified postgres URI.

Also note the following: * The first query_manifest row has num_workers=4 specified. This means that distribute() would not ask the connection how many CPUs it has, but would instead just make 4 connections. However, since there are only 2 queries in the list, only 2 connections will be made. * The second query_manifest row has cpu_multiplier=1.5. This means that distribute() will invoke pmpp.num_cpus() on one of the remote connections to determine how many CPUs that machine has, and the function will fail if PMPP is not installed at that connection. Assuming the remote machine has 4 CPUs, a multiplier of 1.5 would mean that PMPP would try to make 6 connections. Again, however, there are only 2 queries specified, so only 2 connections would be made.

distribute() - jsonb/json versions

sql function distribute( p_row_type anyelement, p_query_manifest jsonb ) returns setof anyelement function distribute( p_row_type anyelement, p_query_manifest json ) returns setof anyelement

These are essentially the same as the query_manifest[] version, but the commands are encoded as single JSON[B] document. The JSON[B] structure is just an array of elements, each element being a collection of attributes with the same names as the attributes in type query_manifest. Omitted elements are left null.

Example

CREATE TEMPORARY TABLE xy(x integer, y integer); SELECT * FROM pmpp.distribute( null::xy, '[{"connection": "foreign_server_without_pmpp", "queries": ["SELECT 1,2","SELECT 3,4"], "num_workers": 4},' ' {"connection": "named_foreign_server", "queries": ["SELECT 5,6", "SELECT 7,8"], "multiplier": 1.5},' ' {"connection": "postgresql://user:secret@localhost", "query": "SELECT 9,10"}]'::jsonb) ORDER by x; x | y ---+---- 1 | 2 3 | 4 5 | 6 7 | 8 9 | 10 (1 row)

meta()

Given a single connection string, execute a series of statements (ones that don't return sets) in parallel.

sql function meta( connection text, sql_list text[], cpu_multiplier float default null, num_workers integer default null, statement_timeout integer default null) returns setof command_with_result

Each statement in sql_list is wrapped in a SELECT pmpp.execute_command(<statement>), which returns a type command_with_result which looks like this:

sql type command_with_result ( command text, result text );

Example:

CREATE TABLE a( b integer, c integer, d integer); CREATE TABLE SELECT * FROM pmpp.meta('dbname=' || current_database(), array( SELECT format('create index on %s(%s)',c.table_name,c.column_name) FROM information_schema.columns c WHERE c.table_name = 'parallel_index_test' AND c.table_schema = 'public')) ORDER BY 1; command | result ----------------------------------------+-------- create index on parallel_index_test(b) | OK create index on parallel_index_test(c) | OK create index on parallel_index_test(d) | OK (3 rows)

This example has the user connecting to the same database as a local superuser (normally not a good idea) and creating an index on every column in the table. The database connected to must also have PMPP installed.

Other functions

execute_command

sql function execute_command(sql text) RETURNS command_with_result

Executes the command sql and returns that string along with a returncode of 'OK' or 'FAIL'.

This function is used inside meta(). It is not particularly useful on its own.

Example:

SELECT * from pmpp.execute_command('analyze'); command | result ---------+-------- analyze | OK (1 row)

Internal functions of no direct utility to the user

Presented mostly to alleviate curiosity.

jsonb_array_to_text_array

sql function jsonb_array_to_text_array(jsonb jsonb) returns text[]

PostgreSQL current has no function to convert direction from a JSONB array to a text[].

num_cpus

sql function num_cpus() RETURNS integer

Returns the number of CPUs detected on this machine at the time the extension was added.

sql function num_cpus(multiplier float) RETURNS integer

Returns multiplier * num_cpus(), but never less than 1.

disconnect

sql function disconnect() returns setof text

Cancels all active queries and disconnects from all connections created by PMPP. This is really only useful if the query dies and PMPP somehow fails to clean up after itself.

query_manifest functions

sql function to_query_manifest(item in jsonb) returns query_manifest

The function behind the jsonb to query_manifest cast.

sql function to_query_manifest_array(manifest in jsonb) returns query_manifest[]

The function behind the jsonb to query_manifest[] cast.

sql function manifest_set( query_manifest jsonb ) returns setof query_manifest

Extracts a set of query_manifest records from a jsonb object.

sql function manifest_array( query_manifest jsonb ) returns query_manifest[]

Extracts an array of query_manifest records from a jsonb object.

Considerations

  • Objects and data created by the parent connection but not yet committed are not visible to the connections created by PMPP.
  • The connection specified may not be the same user as the calling function, and thus may not have the same object permissions as the parent connection.
  • If a sub-query fails, PMPP will try to cancel all other sub-queries and disconnect from those connections.
  • Replica databases are completely read-only, so it is not possible to create temporary tables on them.
  • In situations where the remote machine isn't a "real" PostgreSQL instance, but instead another database that happens to use the libpq protocol (Redshift, Vertica), the num_workers parameter must be specified.
  • When querying a database that has it's own massively-parallel engine (Redshift, Vertica), there is no point in starting more than one worker per server, so set num_workers=1.

Under The Hood.

Parallel querying is accomplished through usage of the async funcions in the dblink extension. Queue management is currently implemented in temp tables with plpgsql, though it may be necessary to recode it in C due to some shortcomings of pl/pgsql.

Wishlist

A few things that could make PMPP better.

  • Polymorphic dblink() functions or TYPEing of SETOF RECORD functions. The function call to dblink_get_result() is dynamic because the typecast is not known until runtime. This is being worked on for PostgreSQL 9.6.

  • Ability to fetch PQcmdStatus(PGresult *res) from PL/PGSQL

    Currently only success or failure is detected.

Support

Submit issues to the GitHub issue tracker.

Author

Corey Huinker, while working at Moat

Copyright and License

Copyright (c) 2015, Moat Inc.

Permission to use, copy, modify, and distribute this software and its documentation for any purpose, without fee, and without a written agreement is hereby granted, provided that the above copyright notice and this paragraph and the following two paragraphs appear in all copies.

IN NO EVENT SHALL Moat, Inc. BE LIABLE TO ANY PARTY FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF Moat, Inc. HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

Moat, Inc. SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND Moat, Inc. HAS NO OBLIGATIONS TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.