PGO

PG...Oh god not nother Postgres client in Erlang...

Origin

This is a fork of pgo by Tristan Sloughter and the pgo contributors, licensed under the Apache License, Version 2.0. The design and nearly all of the code are theirs, and the credit for them is too.

The fork exists to carry a small set of changes that projects here depend on, and to release them on a cadence those projects control. Relative to upstream it adds:

• prepared statements: pgo:prepare/2,3 and pgo:query_prepared/3,4, including auto-preparation across the connections in a pool
• a tcp_closed message is handled the same way as ssl_closed in pgo_connection
• a COMMIT that the server downgrades to ROLLBACK is surfaced instead of being reported as a successful commit

Everything else is upstream's. Full attribution for this project and for the projects it in turn derives from is in NOTICE; licence terms are in LICENSE.

Why

• No message passing. Clients checkout the socket and use it directly.
• Binary protocol with input oids cached.
• Simple and direct. Tries to limit runtime options as much as possible.
• Instrumented with OpenTelemetry
• Mix apps currently too hard to use in a Rebar3 project.

Requirements

Erlang/OTP 26 and above.

Use

Pools defined in the pgo application's environment will be started on boot. You can also add pools dynamically with pgo:start_pool/3.

To try pgo simply modify config/example.config by replacing the host, database, user and password values for the database you wish to connect to:

[
  {pgo, [{pools, [{default, #{pool_size => 10,
                              host => "127.0.0.1",
                              database => "test",
                              user => "test"}}]}]}
].

default is the name of the pool, size is the number of connections to create for the pool. Or you can start the pool through pgo:start_pool/2 which creates it as a child of pgo's simple one for one:

> application:ensure_all_started(pgo).
{ok,[backoff,opentelemetry_api,pg_types,pgo]}
> pgo:start_pool(default, #{pool_size => 5, host => "127.0.0.1", database => "test", user => "test"}). 

Or start a pool as a child of your application's supervisor:

ChildSpec = #{id => pgo_pool,
              start => {pgo_pool, start_link, [Name, PoolConfig]},
              shutdown => 1000},

Then start a shell with rebar3 shell, it will boot the applications which will start the pool automatically if it is configured through sys.config.

> pgo:query("select 1").
#{command => select, num_rows => 1, rows => [{1}]}
> pgo:transaction(fun() ->
>     pgo:query("INSERT INTO my_table(name) VALUES('Name 1')"),
>     pgo:query("INSERT INTO my_table(name) VALUES('Name 2')")
> end).
#{command => insert,num_rows => 1,rows => []}

Options

Pool configuration includes the Postgres connection information, pool configuration like size and defaults for options used at query time.

#{host => string(),
  port => integer(),
  user => string(),
  password => string(),
  database => string(),

  %% pool specific settings
  pool_size => integer(),
  queue_target => integer(),
  queue_interval => integer(),
  idle_interval => integer(),

  %% gen_tcp socket options
  socket_options => [gen_tcp:socket_option()],

  %% defaults for options used at query time
  queue => boolean(),
  trace => boolean(),
  decode_opts => [decode_option()]}

The query time options can also be set through options passed to pgo:query/3:

decode_fun() :: fun((row(), fields()) -> row()) | undefined.

decode_option() :: return_rows_as_maps | {return_rows_as_maps, boolean()} |
                   column_name_as_atom | {column_name_as_atom, boolean()} |
                   {decode_fun, decode_fun()}.
                         
#{pool => atom(),
  trace => boolean(),
  queue => boolean(),
  decode_opts => [decode_option()]}

Query Options

• pool (default: default): Name of the pool to use for checking out a connection to the database.
• return_rows_as_maps (default: false): When true each row is returned as a map of column name to value instead of a list of values.
• column_name_as_atom (default: false): If true converts each column name in the result to an atom.
• decode_fun (default: undefined): Optional function for performing transformations on each row in a result. It must be a 2-arity function returning a list or map for the row and takes the row (as a list or map) and a list of #row_description_field{} records.
• queue (default: true): Whether to wait for a connection from the pool if none are available.
• trace (default: false): pgo is instrumented with OpenTelemetry and when this option is true a span will be created (if sampled).

Database Settings

• host (default: 127.0.0.1): Database server hostname.
• port (default: 5432): Port the server is listening on.
• user: Username to connect to database as.
• password: Password for the user.
• database: Name of database to use.
• ssl (default: false): Whether to use SSL or not.
• ssl_options: List of SSL options to use if ssl is true. See the Erlang SSL connect options.
• connection_parameters (default: []): List of 2-tuples, where key and value must be binary strings. You can include any Postgres connection parameter here, such as {<<"application_name">>, <<"myappname">>} and {<<"timezone">>, <<"GMT">>}.

Pool Settings

• pool_size (default: 1): Number of connections to keep open with the database
• queue_target (default: 50) and queue_interval (default: 1000): Checking out connections is handled through a queue. If it takes longer than queue_target to get out of the queue for longer than queue_interval then the queue_target will be doubled and checkouts will start to be dropped if that target is surpassed.
• idle_interval (default: 1000): The database is pinged every idle_interval when the connection is idle.

Erlang TCP Socket Settings

• socket_options (default []): Addition options to pass to gen_tcp:connect such as inet6 for IPv6 support.

Telemetry and Tracing

OpenTelemetry spans can be enabled for queries and transactions by either setting the trace to true for the pool:

> pgo:start_pool(default, #{host => "127.0.0.1", 
                            database => "test", 
                            user => "test",
                            pool_size => 5,
                            trace => true}]). 

Or by passing #{trace => true} in the options for a query or transaction:

> pgo:query("select 1", [], #{trace => true}).
#{command => select, num_rows => 1, rows => [{1}]}
> pgo:transaction(fun() ->
>     pgo:query("INSERT INTO my_table(name) VALUES('Name 1')"),
>     pgo:query("INSERT INTO my_table(name) VALUES('Name 2')")
> end, #{trace => true}).
#{command => insert,num_rows => 1,rows => []}

Note that since this is optional the opentelemetry application is not included as a dependency of pgo -- only opentelemetry_api is included by default. So it must be included as a rebar3 dependency and runtime dependency (listed in your application's .app.src applications or the list of applications for relx to include in a release).

Listen/Notify

Use pgo_notifications:start_link to create a process holding a connection to the database that you can call pgo_notifications:listen/2 with to listen on a channel for notifications. Those notifications will arrive at the calling process in the form:

{notification, Pid, Ref, Channel, Payload}

Running Tests

Pool functionality is tested with common test suites:

$ rebar3 ct

Postgres query functionality is tested with eunit, create user test and database test:

$ rebar3 eunit

Acknowledgements

Much is owed to https://github.com/semiocast/pgsql (especially for protocol step logic) and https://github.com/epgsql/epgsql/ (especially for some decoding logic).

The pool implementation is owed to James Fish's found in db_connection PR 108. While db_connection and postgrex as a whole were both used as inspiration as well.

Licence

Apache License, Version 2.0. See LICENSE for the terms, and NOTICE for the origin and licence of the code this project is derived from, including code under the BSD 2-Clause Licence. Both files must be kept with the source when it is redistributed.