id: csv-postgres-parallel-snapshot
namespace: company.data
description: Parse a CSV export, load bounded ION chunks in parallel, and
publish a complete PostgreSQL snapshot in one transaction.
concurrency:
limit: 1
behavior: QUEUE
inputs:
- id: source_url
type: STRING
defaults: https://huggingface.co/datasets/kestra/datasets/raw/main/csv/orders.csv
description: URL of an immutable CSV export with the seven documented order columns.
- id: rows_per_chunk
type: INT
defaults: 5000
min: 1
max: 50000
- id: allow_empty
type: BOOL
defaults: false
description: Allow a header-only export to clear the destination. False protects
the last good snapshot.
tasks:
- id: download_export
type: io.kestra.plugin.core.http.Download
uri: "{{ inputs.source_url }}"
- id: parse_csv
type: io.kestra.plugin.serdes.csv.CsvToIon
description: Parse CSV records before splitting, so quoted commas and newlines
remain in one record.
from: "{{ outputs.download_export.uri }}"
header: true
onBadLines: ERROR
skipEmptyRows: true
- id: create_tables
type: io.kestra.plugin.jdbc.postgresql.Queries
url: "{{ secret('SNAPSHOT_JDBC_URL') }}"
username: "{{ secret('SNAPSHOT_USER') }}"
password: "{{ secret('SNAPSHOT_PASSWORD') }}"
sql: |
CREATE SCHEMA IF NOT EXISTS csv_snapshot;
CREATE TABLE IF NOT EXISTS csv_snapshot.orders (
order_id bigint PRIMARY KEY,
customer_name text NOT NULL,
customer_email text NOT NULL,
product_id bigint NOT NULL,
price numeric NOT NULL CHECK (price >= 0),
quantity integer NOT NULL CHECK (quantity > 0),
total numeric NOT NULL CHECK (total >= 0)
);
CREATE TABLE IF NOT EXISTS csv_snapshot.staging (
execution_id text NOT NULL,
order_id bigint NOT NULL,
customer_name text NOT NULL,
customer_email text NOT NULL,
product_id bigint NOT NULL,
price numeric NOT NULL CHECK (price >= 0),
quantity integer NOT NULL CHECK (quantity > 0),
total numeric NOT NULL CHECK (total >= 0),
PRIMARY KEY (execution_id, order_id)
);
CREATE TABLE IF NOT EXISTS csv_snapshot.publications (
execution_id text PRIMARY KEY,
expected_rows bigint NOT NULL,
staged_rows bigint NOT NULL,
allow_empty boolean NOT NULL,
published_at timestamptz NOT NULL DEFAULT now(),
CHECK (staged_rows = expected_rows),
CHECK (allow_empty OR staged_rows > 0)
);
- id: split_records
type: io.kestra.plugin.core.storage.Split
description: Split binary ION by record count, rather than by CSV physical lines.
from: "{{ outputs.parse_csv.uri }}"
rows: "{{ inputs.rows_per_chunk }}"
- id: stage_chunks
type: io.kestra.plugin.core.flow.Loop
description: Load at most three chunks at once. A repeated chunk replaces the
same staging keys.
values: "{{ outputs.split_records.uris }}"
concurrencyLimit: 3
tasks:
- id: load_chunk
type: io.kestra.plugin.jdbc.postgresql.Batch
url: "{{ secret('SNAPSHOT_JDBC_URL') }}"
username: "{{ secret('SNAPSHOT_USER') }}"
password: "{{ secret('SNAPSHOT_PASSWORD') }}"
from: "{{ item.value }}"
sql: |
INSERT INTO csv_snapshot.staging
(execution_id, order_id, customer_name, customer_email, product_id, price, quantity, total)
VALUES ('{{ execution.id }}', CAST(CAST(? AS text) AS bigint), ?, ?,
CAST(CAST(? AS text) AS bigint), CAST(CAST(? AS text) AS numeric),
CAST(CAST(? AS text) AS integer), CAST(CAST(? AS text) AS numeric))
ON CONFLICT (execution_id, order_id) DO UPDATE SET
customer_name = EXCLUDED.customer_name, customer_email = EXCLUDED.customer_email,
product_id = EXCLUDED.product_id, price = EXCLUDED.price,
quantity = EXCLUDED.quantity, total = EXCLUDED.total
columns:
- order_id
- customer_name
- customer_email
- product_id
- price
- quantity
- total
- id: inspect_staging
type: io.kestra.plugin.jdbc.postgresql.Query
description: A duplicate order ID reduces the staging count; it must not reach
the live table.
url: "{{ secret('SNAPSHOT_JDBC_URL') }}"
username: "{{ secret('SNAPSHOT_USER') }}"
password: "{{ secret('SNAPSHOT_PASSWORD') }}"
fetchType: FETCH_ONE
sql: |
SELECT count(*) AS staged_rows
FROM csv_snapshot.staging WHERE execution_id = '{{ execution.id }}';
- id: validate_snapshot
type: io.kestra.plugin.core.execution.Assert
conditions:
- "{{ outputs.inspect_staging.row.staged_rows == outputs.parse_csv.size }}"
- "{{ inputs.allow_empty or outputs.parse_csv.size > 0 }}"
errorMessage: The snapshot is empty or contains duplicate order IDs. The last
published snapshot is unchanged.
- id: publish_snapshot
type: io.kestra.plugin.jdbc.postgresql.Queries
description: Validate again under a database lock, then replace all live rows in
the same transaction.
url: "{{ secret('SNAPSHOT_JDBC_URL') }}"
username: "{{ secret('SNAPSHOT_USER') }}"
password: "{{ secret('SNAPSHOT_PASSWORD') }}"
sql: |
BEGIN;
SET LOCAL lock_timeout = '30s';
SET LOCAL statement_timeout = '120s';
LOCK TABLE csv_snapshot.orders IN EXCLUSIVE MODE;
INSERT INTO csv_snapshot.publications
(execution_id, expected_rows, staged_rows, allow_empty)
SELECT '{{ execution.id }}', {{ outputs.parse_csv.size }}, count(*), {{ inputs.allow_empty }}
FROM csv_snapshot.staging WHERE execution_id = '{{ execution.id }}'
ON CONFLICT (execution_id) DO UPDATE SET
expected_rows = EXCLUDED.expected_rows, staged_rows = EXCLUDED.staged_rows,
allow_empty = EXCLUDED.allow_empty, published_at = now();
DELETE FROM csv_snapshot.orders;
INSERT INTO csv_snapshot.orders
(order_id, customer_name, customer_email, product_id, price, quantity, total)
SELECT order_id, customer_name, customer_email, product_id, price, quantity, total
FROM csv_snapshot.staging WHERE execution_id = '{{ execution.id }}';
COMMIT;
- id: verify_publication
type: io.kestra.plugin.jdbc.postgresql.Query
url: "{{ secret('SNAPSHOT_JDBC_URL') }}"
username: "{{ secret('SNAPSHOT_USER') }}"
password: "{{ secret('SNAPSHOT_PASSWORD') }}"
fetchType: FETCH_ONE
sql: |
SELECT p.staged_rows AS published_rows,
(SELECT count(*) FROM csv_snapshot.orders) AS live_rows
FROM csv_snapshot.publications p WHERE execution_id = '{{ execution.id }}';
- id: require_complete_publication
type: io.kestra.plugin.core.execution.Assert
conditions:
- "{{ outputs.verify_publication.row.published_rows ==
outputs.parse_csv.size }}"
- "{{ outputs.verify_publication.row.live_rows == outputs.parse_csv.size
}}"
triggers:
- id: daily_snapshot
type: io.kestra.plugin.core.trigger.Schedule
cron: "0 3 * * *"
disabled: true
outputs:
- id: published_rows
type: INT
value: "{{ outputs.verify_publication.row.published_rows }}"
- id: chunks
type: INT
value: "{{ outputs.split_records.uris | length }}"
