PostgreSQL Query

Runs a SQL statement against PostgreSQL. Returns the rows as Arrow RecordBatch.

Configuration

- postgres_query:
    name: get_orders
    credentials_path: /etc/postgres/credentials.json
    query: "SELECT * FROM orders WHERE status = $1 AND placed_at > $2"
    parameters:
      - "{{event.data.status}}"
      - "{{event.data.since}}"
    batch_size: 5000

Fields

FieldTypeDefaultDescription
namestringrequiredTask name.
credentials_pathstringrequiredPath to the credentials file.
querystring/resourcerequiredSQL statement. Supports templating and resource files.
parameterslistValues for $1, $2, etc. Sent as text; the server converts them to the parameter types.
batch_sizeint10000Rows per Arrow RecordBatch.
max_connectionsint10Connection pool size.
connection_timeoutduration30sTimeout to open a connection.
query_timeoutduration2mLongest a statement runs on the server (statement_timeout), and timeout to get a connection and the first rows.
depends_onlistUpstream task names.
retryobjectRetry configuration.

Connection errors, serialization failures and deadlocks are retried. SQL errors such as a syntax error or a constraint violation are not, nor is a statement that ran past query_timeout, nor a query that failed after it sent batches.

Output

Rows are streamed and emitted as one event per batch_size rows. A query without rows emits one empty batch with the result columns. A statement without a result set, such as an INSERT without RETURNING, emits one empty batch.

PostgreSQLArrow
booleanBoolean
smallint, integer, bigintInt16, Int32, Int64
oidUInt32
real, double precisionFloat32, Float64
numericUtf8, exact decimal text
text, varchar, char, "char", name, citext, xml, enumsUtf8
json, jsonbUtf8, JSON text
uuidUtf8
dateDate32
timeTime64(Microsecond)
timestampTimestamp(Microsecond)
timestamptzTimestamp(Microsecond, UTC)
byteaBinary

infinity and -infinity dates and timestamps are null.

Other types, such as arrays, interval and inet, fail the event with the column name. Cast them in the query, for example tags::text or to_json(tags).