introducing pgasync and pgflow - Mailing list pgsql-general

I'd like to introduce two libraries I've been working on: pgasync and pgflow.   Both libraries are implemented 100% in SQL and therefore can be deployed in just about any environment, only needing the dblink extension to be available (although pg_cron is super nice to have as well).  Both libraries are in beta; they are battle tested but not yet stable enough for production guarantees, especially through version upgrades.

pgasync is a background query processing engine that will feel very similar to pg_background.  It has what I believe is a unique and interesting architecture in that it starts up a stored procedure daemon process and runs forever.  pgasync also supports a number of useful features including:

pgflow extends the pgasync backend into running airflow style processing DAGs for various kinds of batch processing utility.  Your queries are organized into various nodes and steps and processed deterministically.  If you want an Airflow-style way to process queries directly from the database, this is the library you've been looking for.   pgflow is feature-packed, which is too much to cover in this email! :-)

While it may on the surface seem limiting to only be able to orchestrate database queries, it's possible do things like call into pg_http to tap a web service and have it report back.

merlin

example:

server session:

ingestdb_superuser@postgres=# \i ~/src/ingestdb/lib/async/dist/scripts/async_server_full.sql
DO
DO
DO
DO
DO
ingestdb_superuser@postgres=# call async.main(true);
NOTICE:  Initializing asynchronous query processor
NOTICE:  Performing heavy maintenance
NOTICE:  Performed heavy maintenance in 0.00 seconds
NOTICE:  Performing light maintenance
NOTICE:  Performed light maintenance in 0.02 seconds
NOTICE:  Cleaning up unfinished tasks (if any)
NOTICE:  Initializing workers
NOTICE:  Force disconnecting existing workers
NOTICE:  Initializing concurrency pools
NOTICE:  Worker initialization complete
NOTICE:  Initialization of query processor complete

client session:

SELECT async.configure($j${
  "targets": [
    {
      "target": "SELF",
      "max_concurrency": 20,
      "connection_string": "host=localhost dbname=postgres user=ingestdb_superuser"
    }
  ]
}$j$);

CREATE TABLE test
(
  test_id SERIAL,
  pid INT DEFAULT pg_backend_pid(),
  audit TIMESTAMPTZ DEFAULT now()
);

ingestdb_superuser@postgres=# SELECT COUNT(*) FROM (SELECT async.push_tasks(array_agg(t))
FROM (select async.task('INSERT INTO test DEFAULT VALUES', 'SELF') t FROM generate_series(1,1000)));
-[ RECORD 1 ]
count | 1000

ingestdb_superuser@postgres=# select count(*), count(DISTINCT pid), max(audit) - min(audit) from test;
-[ RECORD 1 ]-------------
count    | 1000
count    | 20
?column? | 00:00:00.239228

I'd love to hear your thoughts and suggestions.  Thank you!

merlin