PostgreSQL

Runs a query you write against a PostgreSQL database every poll, and reads each row of its result as one record.

Use it for what an application keeps in its own tables, such as an audit table or a sign-in history. The source reads only what your query returns, and its mode decides which of those rows are new.

Example

Reads the table auth.events in the database auth on db.internal, as the user smeltr_reader, whose password is in the worker-host variable PG_PASSWORD. It begins with the rows added after its first poll, and each poll reads the rows whose event_id is past the last one sent.

{
  "id": "auth-events",
  "type": "postgres",
  "host": "db.internal",
  "database": "auth",
  "user": "smeltr_reader",
  "password": "PG_PASSWORD",
  "query": "SELECT event_id, occurred_at, user_name FROM auth.events",
  "mode": {
    "type": "incrementing",
    "position": [
      "event_id"
    ]
  },
  "fields": [
    {
      "name": "event_id",
      "type": "i64"
    },
    {
      "name": "occurred_at",
      "type": "datetime"
    },
    {
      "name": "user_name",
      "type": "str"
    }
  ]
}

Open complete screenshot ↗

Open complete screenshot ↗

Before you start

• Create a user that can connect to the database and read the tables the query reads, and nothing more. See PostgreSQL's CREATE ROLE and GRANT.
• Let the user connect from the workers' addresses in pg_hba.conf, and allow workers to reach the server's port.
• With tls require, the default, the server needs SSL on, with a certificate that names host. For a certificate from a private authority, give the source a connection identity that trusts it.
• Put the password in a credential.
• For Incrementing or Updated at, index the first field of position, so each poll reads only the new rows. See CREATE INDEX.

Settings

The form on the Sources page asks for the connection first, then the query. Its Run button runs the query on a worker of the group and shows the rows it read, so you choose the mode and the fields from the real result. The source also takes the settings every source takes.

Advanced

Fields it gives

Each record is one row. It holds a field for each entry of fields, read from the result column of the same name and held as the type the entry declares. A column no entry names is not read.

How it finds new rows

mode says which rows a poll sends. The form lists the modes cheapest first, and suggests one from the result of Run.

Incrementing

Sends each row once. position names a field whose value only grows as rows are added, such as an id or the time a row was inserted. Each poll reads the rows past the last row sent, in the order of position. A row added later with a smaller value than one already sent is never read.

Updated at

Sends each row when it is added and again each time it changes. position starts with a datetime that every insert and every update sets. A deleted row sends nothing.

Comparing

Sends each row whose key is new, or whose fields changed since the last poll. Each poll reads the whole result, up to 1,000,000 rows, and the leader keeps a hash of each row. Use it for a table with no column that grows, such as a list of accounts. A deleted row sends nothing; a row whose key comes back is sent again.

Every poll

Sends the whole result on every poll, and remembers nothing. Use it for a small table read as a whole, such as current settings.

In position, the fields after the first tell apart rows that share a value in it, such as an id after a time. When many rows share one time, add such a field. Without one, a poll that fills batch leaves its last rows sharing one value to the next poll, which reads up to twice batch. Past that, the rest of the rows sharing the value are skipped, and the worker log says so.

A position that starts with a datetime reads only rows at least settleSecs behind the database's clock. A transaction that commits after a later one is then still read, as long as it commits within settleSecs. Rows arrive that much later.

:last.<field> in the query is the value that field held in the last row sent, and NULL before the first. Use it to filter where the statement's own test cannot reach, such as inside a GROUP BY. Keep rows with the last value (>=, not >), because rows sharing it may not all have been sent.

Column types

Run suggests each field's type from its column's type. A column of a type not listed here is read as its text.

A numeric is read as its exact decimal text. An interval counts a month as 30 days. Run suggests an array of one dimension; declare an array<…> for each further one, or variant.

Where it starts

A source in Incrementing or Updated at with no progress begins where start says:

• now, the default: at the largest position the query returns on its first poll. It sends the rows that hold that value, then the rows added after them.
• beginning: every row the query returns.
• from: the rows whose position is at or past startFrom.

Comparing sends every row on its first poll. After that, the leader keeps how far every destination has taken, and a restarted worker, or another worker that takes the source over, resumes there. It may send a row twice and never skips one. Changing the connection's host, port or database, the query, the mode, or the fields of position or key starts the source afresh from start. The other settings keep its progress. The source keeps reading until you remove it.

How it works

One worker of the group holds the source. Each poll runs your query inside a statement of Smeltr's own, which adds the position test, the order and the batch limit. After the example's first row, each of its polls runs:

SELECT * FROM (
SELECT event_id, occurred_at, user_name FROM auth.events
) AS q
WHERE q."event_id" > $1
ORDER BY q."event_id"
LIMIT 5000

The form shows the exact statement for your settings under what smeltr runs instead. The query is one SELECT or WITH, with no ORDER BY, LIMIT or placeholders of its own. The source runs the statement in a read-only transaction, with the session's time zone set to UTC, and reads a time without a zone as UTC. The settle test compares with now(), the database's own clock.

A poll that fills batch reads the next at once. Otherwise the next poll starts pollIntervalSecs after this one. When Run checks a poll's plan and finds that each poll would read a whole table of more than about 100,000 rows, the form warns. Index the first field of position, or filter inside the query with :last.<field>.

A poll's result may hold up to 256 MiB. The source has the database stop a statement after 290 seconds, or sooner if the database's own limit is shorter, and drops a connection that has not answered in 300. After a failure, the source asks again after one second, then waits twice as long each time, up to five minutes. It opens a new connection after one breaks.

Verify

1. In the form, choose Run. It shows the first rows the query returns, the record the pipeline sees for the first, and, once you choose a position, whether a poll reads through an index.
2. Once the source is deployed, add a row the query returns. It reaches the destination within pollIntervalSecs, plus settleSecs for a datetime position. On the Sources page, the worker column names the worker that holds the source.

Troubleshooting

The form shows what the PostgreSQL server answered to Run, and the worker log shows it for a poll.

The PostgreSQL password is in …, which this worker does not have

The credential is not on the worker that holds the source. See Credentials.

Run fails with a refusal to read a table

Grant the user SELECT on the table, and USAGE on its schema.

The result has no column x, which the record declares

An entry of fields names a column the query does not return. Rename the entry, or give the column that name with AS in the query.

The result has two columns named x

Give one of them another name with AS in the query.

Each poll scans all of a table

The database reads the whole table on each poll to find the new rows. Index the first field of position, or filter inside the query with :last.<field>.

More rows share one position value than twice the batch

Add a field to position that tells those rows apart, such as an id, or raise batch.

Two rows have the key …

In Comparing, key must be unique in the result. Add fields to key until it is.

Every poll reads at most … rows

In Every poll, the result holds more rows than batch. Raise batch, up to 100,000, or choose another mode.

This poll's result holds more than 256 MiB

Lower batch, or read fewer or narrower fields.

The position's first column is of a type smeltr reads only as text

Cast the column in the query to a type in the table under Column types.

See also

• Sources
• Credentials
• MySQL and MariaDB source
• PostgreSQL: Database roles