PostgreSQL Batch

PostgreSQL Batch

Certified

Bulk insert rows into PostgreSQL using prepared statements

Reads ION-formatted data from Kestra internal storage and performs high-performance batch inserts using JDBC batch operations. Data is processed in chunks (default 1,000 rows) to optimize memory and performance. Supports auto-commit for databases without transaction support.

yaml
type: io.kestra.plugin.jdbc.postgresql.Batch

Fetch rows from a table, and bulk insert them to another one.

yaml
id: postgres_bulk_insert
namespace: company.team

tasks:
  - id: query
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://dev:5432/
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      SELECT *
      FROM xref
      LIMIT 1500;
    fetchType: STORE

  - id: update
    type: io.kestra.plugin.jdbc.postgresql.Batch
    from: "{{ outputs.query.uri }}"
    url: jdbc:postgresql://prod:5433/
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      insert into xref values( ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ? )

Fetch rows from a table, and bulk insert them to another one, without using sql query.

yaml
id: postgres_bulk_insert
namespace: company.team

tasks:
  - id: query
    type: io.kestra.plugin.jdbc.postgresql.Query
    url: jdbc:postgresql://dev:5432/
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    sql: |
      SELECT *
      FROM xref
      LIMIT 1500;
    fetchType: STORE

  - id: update
    type: io.kestra.plugin.jdbc.postgresql.Batch
    from: "{{ outputs.query.uri }}"
    url: jdbc:postgresql://prod:5433/
    username: "{{ secret('POSTGRES_USERNAME') }}"
    password: "{{ secret('POSTGRES_PASSWORD') }}"
    table: xref

Use Postgres Batch to bulk insert rows

yaml
  id: postgres_batch
  namespace: company.team

  tasks:
    - id: download_products_csv_file
      type: io.kestra.plugin.core.http.Download
      uri: https://huggingface.co/datasets/kestra/datasets/raw/main/csv/products.csv

    - id: products_csv_to_ion
      type: io.kestra.plugin.serdes.csv.CsvToIon
      from: "{{ outputs.download_products_csv_file.uri }}"

    - id: postgres_create_table
      type: io.kestra.plugin.jdbc.postgresql.Query
      url: "jdbc:postgresql://{{ secret('POSTGRES_HOST') }}:5432/postgres"
      username: "{{ secret('POSTGRES_USERNAME') }}"
      password: "{{ secret('POSTGRES_PASSWORD') }}"
      sql: |
        CREATE TABLE IF NOT EXISTS products(
          product_id varchar(5),
          product_name varchar(100),
          product_category varchar(50),
          brand varchar(50)
        )

    - id: postgres_batch_insert
      type: io.kestra.plugin.jdbc.postgresql.Batch
      url: "jdbc:postgresql://{{ secret('POSTGRES_HOST') }}:5432/postgres"
      username: "{{ secret('POSTGRES_USERNAME') }}"
      password: "{{ secret('POSTGRES_PASSWORD') }}"
      from: "{{ outputs.products_csv_to_ion.uri }}"
      sql: |
        insert into products values (?, ?, ?, ?)
Properties

Input file from internal storage

URI of the source file (kestra://) containing rows to insert

The JDBC URL to connect to the database

Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.

Definitions
assetFailureBehaviorstring
Possible Values
IGNOREFAILWARN

Asset failure behavior

Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.

enableAutobooleanstring

Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.

inputsarray

The assets consumed as inputs.

id*string
Min length1
typestring
outputs

The assets produced as outputs.

id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*object
descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
id*string
Min length1
Max length150
type*string
Min length1

Custom asset type

descriptionstring
displayNamestring
metadataobject
Default{}
namespacestring
Min length1
Max length150
Default1000

Batch size per executeBatch call

Number of rows sent per JDBC batch before commit; default 1,000

SubTypestring

Columns bound to placeholders

Ordered column names matching ? placeholders; if omitted, placeholder count must match all columns in the input row

DefaultAUTO
Possible Values
AUTOSTREAMLOCAL

Input handling strategy

Controls how input is read during processing and retries. AUTO buffers small files locally (<= localBufferMaxBytes) and streams large files. STREAM always streams from internal storage. LOCAL always buffers input to a local temporary file before processing.

Default104857600

Maximum number of bytes buffered locally

Used by AUTO and LOCAL input handling. In AUTO, files larger than this threshold are streamed. In LOCAL, files larger than this threshold fail fast.

Default3

Maximum number of retries for transient failures

Retries are attempted only for transient failures such as temporary I/O and recoverable SQL errors.

Defaultfalse

Expose the executed SQL statement in the outputs

When true, the INSERT statement is exposed in the statement output, truncated to 64 KiB. Disabled by default so that SQL text never leaks into outputs unless requested

The database user's password

Defaulttrue

Resume from the last successfully committed chunk on retry

DefaultPT1S

Delay between retry attempts

Uses ISO-8601 duration format, for example PT1S.

DefaultINPUT
Possible Values
NONEINPUTALL

Controls which failures are retried

INPUT retries input handling failures, ALL retries all retryable failures.

Parameterized INSERT statement to execute

Prepared INSERT with ? placeholders for each bound column. Example: INSERT INTO <table_name> VALUES (?, ?, ?) for three columns; use column list if inserting a subset

Defaultfalse

Is the connection SSL?

The SSL cert

Must be a PEM encoded certificate

The SSL key

Must be a PEM encoded key

The SSL key password

Possible Values
DISABLEALLOWPREFERREQUIREVERIFY_CAVERIFY_FULL

The SSL mode

The SSL root cert

Must be a PEM encoded certificate

Table used to auto-discover columns

Retrieves column names from the given table when columns is empty. If sql is also omitted, an INSERT statement is generated automatically using the discovered columns

The time zone id to use for date/time manipulation. Default value is the worker's default time zone id

The database user

Batch execution duration in milliseconds

Includes input handling and retries

Total rows read

Number of rows affected by the batch

Sum of the update counts reported by the JDBC driver for every executed batch

Executed SQL statement

Only populated when outputStatement is true; truncated to 64 KiB

Tables written by the statement

Detected from the SQL text, so this is a best effort. Limited to 50 entries, see tablesTruncated

Definitions
accessstring
Possible Values
READWRITECREATEDROP

How the statement uses the table

One of READ, WRITE, CREATE or DROP.

confidencestring
Possible Values
MEDIUMLOW

How reliable the detection is

MEDIUM when detected from the SQL text, LOW for temporary tables and ambiguous names.

databasestring

Database name

Null when the statement does not qualify the table with a database.

kindstring
Possible Values
TABLEFILEURL

Kind of source

Only set on DuckDB: TABLE, FILE or URL (files read through read_parquet, read_csv, ATTACH and quoted paths).

namestring

Table name

For DuckDB file reads, the file path or URL.

schemastring

Schema name

Null when the statement does not qualify the table with a schema.

Whether the tables list was cut at 50 entries

Rows inserted or updated

Kept for compatibility, prefer rowsAffected

Number of SQL warnings raised by the batch

Count of the SQLWarning chain reported by the driver across all executed batches

The duration of the batch execution, including input handling and retries.

Unitqueries

The number of batch queries executed.

Unitrecords

The number of records processed.

Unitrows

The number of rows affected, as reported by the JDBC driver.

Unitrecords

The number of records updated.

More Plugins in this Category

Firecrawl
Firecrawl

Tasks that scrape and crawl websites into clean, LLM-ready content with Firecrawl.

Data

2 tasks

Liquibase
Liquibase

Tasks that run Liquibase change management commands.

Data

2 tasks

8 blueprints