PGQ Tutorial

PGQ is the queueing solution from Skytools. The
Londiste replication solution is a consumer daemon
built on PGQ, and the API is accessible for you to create any asynchronous
processing facility, based on queuing.
Hint: As of 2021 there is a proposal for an additional part of the SQL standard: "SQL/PGQ" (Property Graph Queries) . Don't confuse the information given here with this proposal.
== What problems is PGQ a solution for? ==
PGQ will solve asynchronous batch processing of live transactions.
That means you're doing some INSERT/DELETE/UPDATE of rows from your live
environment, and you want to trigger some actions but not at COMMIT time,
later on. Not in a too distant future either, just asynchronously: without
blocking the live transaction.
Every application of a certain size will need to postpone some processing at
a later date, and PGQ is a generic high-performance solution built for
PostgreSQL that allows implementing just that: the batch processing. PGQ
will care about the asynchronous consuming of events, the error management,
the queuing behaviour, etc, and it comes with a plain SQL API to do this.
The API documentation is online, this document will consider you know about it.
== Installation and setup ==
You'll need to have a ticker instance running on the database where events
get produced, which means you need to provide a ticker.ini configuration
file and run a ticker daemon.
=== The ticker ===
See Londiste_Tutorial#The_ticker_daemon for an introduction on the
topic, a configuration example and how to start the daemon.
The ticker will produce ticks, which will then serve as boundaries to batch
events. Those batches are produced on-demand, each time a consumer is asking
for a new batch, and will get produced such as any batch contains either
ticker_max_lag seconds worth of events or ticker_max_count events, whichever
comes first.
The daemon is responsible for the maintenance of the queues, which is done by
rotating the events INSERT in 3 tables and using TRUNCATE as soon as one
table does not contain unprocessed events anymore.
=== Producing and Consuming Events ===
In order not to lose any event, you'll want to make sure there's at least
one consumer registered before to produce any event, as events nobody's
interested in are lost.
You can have as many consumers as you want to on the same event queue, but
they will all see the same events rather than share the workload. That makes
sense if you consider that the main application is replication: consumers
are replicas, when you have more than one you want them to all process the same
events in order for all the replicas to have the same data set.
There's a work in progress in order to provide pgq cooperative consumers, the progress is currently stalled on this mailing list thread.
=== More Info ===
See the following for more info:
* http://www.slideshare.net/adorepump/skytools-pgq-queues-and-applications
* http://kaiv.wordpress.com/2007/10/19/skytools-database-scripting-framework-pgq/
== Producing Events ==
The documentation is quite clear about this, you produce event either using
a function based API or using a trigger based one. This latter is prefered,
as it provides easy SQL data type validation, and a known interface to
insert event, the classic INSERT INTO.
== Writing a PGQ consumer ==
Your best option is to write the consumer code in python, next best option is PHP. If you want to reuse existing code written in another language, you'll have to write yourself the Operating System daemon support code and the looping around the API entry points of PGQ.
All the PGQ logic is written as PostgreSQL extension and server side code, so all you need your language to provide is a way to connect to your PostgreSQL server and call functions.
=== How to use the PGQ SQL API ===
You will first need your consumer code to be able to register and unregister from an existing queue.
Then the main idea is pretty simple: the consumer loops over the pgq.next_batch() call as fast as possible until it receives NULL, that's when it should take a rest.
Please note that the ticker daemon will produce ticks every ticker_idle_period when your system is not producing events, so it's perfectly normal (and expected) to get empty batches in low activity periods, such as when debugging your first consumer. Empty batches need to get consumed as fast as possible, only have your custom subscriber rest when pgq.next_batch() returns NULL.
When pgq.next_batch() returns an id (bigint), you get to process the batch
events, which you obtain with pgq.get_batch_events(batch_id). What to do
with each event should be specific to the consumer code rather than in a
common PGQ consuming abstraction interface (or library, package, whatever
your programming environment facility is named).
The event processing function should be able to tag events as processed
(OK), failed or to retry later. In the retry case, the event will
get introduced again in a latter batch, which one depends on how much later
you want it retried.
Once all events of a batch are processed, you have to call
pgq.finish_batch(batch_id) then COMMIT your batch processing transaction.
In pseudo language this gives:
 
 do
 batch_id = pgq.next_batch()
 if batch_id is not null
 then
 BEGIN;
 events = pgq.get_batch_events(batch_id);
 
 // this could be a function pointer or a virtual method or a delegate, e.g.
 user_defined_process_event(events);
 
 pgq.finish_batch(batch_id);
 
 COMMIT;
 end if;
 while true;
=== Remote consuming ===
This is the situation where the processing of the events happens on another
database than the one where the events got produced. Londiste is a good
example of remote consuming.
In this case, you will need to implement a way to avoid processing again a
batch when the processing COMMIT succeeded but the pgq.finish_batch() (which
has to be done on the producer database rather than the consumer one)
fails. How the 3 other scenarios are not a problem is left to the reader.
Skytools provides pgq_ext (extension) as a way to ensure that you won't
process the same batch more than once, and this is discussed here:
[].
The pgq_ext idea is to record on the consumer database the last batch_id
processed, and have the UPDATE be done in the processing transaction. This
can be refined to the last event id processed, for cases where you don't
have an easy way to rollback all the batch processing in case of failure at
the remote site.
As a side note, you might want to know that londiste is using pgq_ext but
moves the SQL entries (functions and tables) into the londiste schema.
=== Non transactional processing ===
Your consumer code could be sending mail rather than changing a database
state. In this case the processing isn't transactional (you can't rollback
the batch processing) and you'll have to solve the reliability issues
yourself, PGQ won't provide the magic.
=== Using the python API ===
Skytools are written mainly in python and provide everything you need to
write your own consumers.
==== Use-case: row counter for count(*) speedup ====
Here's a python code snippet of a PGQ consumer:
 import pgq 
 class RowCounter(pgq.Consumer): 
 def process_batch(self, db, batch_id, ev_list): 
 tbl = self.cf.get('table_name'); delta = 0 
 for ev in ev_list: 
 if ev.type == 'I' and ev.extra1 == tbl: delta += 1 
 elif ev.type == 'D' and ev.extra1 == tbl: delta -= 1 
 ev.tag_done() 
 q = 'select update_stats(%s, %s)' 
 db.cursor().execute(q, [tbl, delta]) 
 RowCounter('row_counter', 'db', sys.argv[1:]).start() 
=== Using the PHP API ===
A PHP API has been contributed too, allowing to write PGQ consumer daemon in
PHP easily. See README
for more detail.
Here's an example
 #!/usr/bin/php5
 <?php
 require( "pgq/PGQRemoteConsumer.php" );
 require("conf/duration.php");
 
 define("CONFIGURATION", "conf/duration.php");
 
 $con_src = "dbname=foo_db port=5432 host=localhost";
 $con_dst = "dbname=bar_db port=5432 host=localhost"
 
 class PGQDaemonExample extends PGQRemoteConsumer
 {
 	public function config( )
 	{
 		unset($Config);
 		if( $this->log !== null )
 			$this->log->notice("Reloading configuration (HUP) from '%s'", CONFIGURATION);
 
 		global $Config;
 		require(CONFIGURATION);
 		
 		$this->loglevel = $Config["LOGLEVEL"];
 		$this->logfile = $Config["LOGFILE"];
 		$this->delay = $Config["DELAY"];
 	}
 	
 	public function process_event( &$event ) 
 	{
 		$this->log->notice("Starting process event");
 		
 		$id = $event->data["id"];
 		$code = $event->data["code"];
 		$data = $event->data["data"];
 		
 		$this->log->notice("Processing event : %d ", $id);
 		
 		$query = sprintf( "UPDATE table SET ... WHERE ...", $code );
 		$this->log->debug( $query );
 		$result = pg_query( $this->pg_dst_con, $query );
 		
 		if( $result === False ) 
 		{
 			$this->log->error( "Unable to update : %s ", $query );
 			$event->retry_delay = 2 * $this->delay;
 			return PGQ_EVENT_RETRY;
 		}
 				
 		return PGQ_EVENT_OK;
 	}
 }
 
 $daemon = new PGQDaemonExample( "mydaemon", "daemonq", "table", $argc, $argv, $con_src, $con_dst );
 ?>
=== Java API ===
A Java consumer API is also available. Documentation is available in the README.md file.
== Trouble Shooting ==
See Londiste's sections about it.
Category:Replication
Category: Languages
Category:Howto