Automate Postgres-to-Blob with durable workflow (pg_durable + pg_duckdb) in Azure Database for PostgreSQL
In the archive post I moved cold rows out of a Postgres table into Parquet on Azure Blob and kept querying them. The steps were manual: export to Blob, verify the file, delete the rows, record what happened. If the process dies after the export but before the delete, you're stuck — re-run and you re-export (wasteful), skip ahead and you might delete rows that were never safely written (dangerous). That partial-failure gap is exactly what a durable workflow closes.
Azure Database for PostgreSQL ships another open-source Microsoft extension that fits here: pg_durable. It runs checkpointed, crash-safe workflows inside Postgres — define a graph of SQL steps, and the engine persists progress between them. If the database restarts or a step fails, execution resumes from the last completed step instead of starting over. No cron-plus-status-table-plus-worker scaffolding. This post turns the archive into one durable, schedulable workflow, with pg_duckdb doing the write to Blob.
Enabling pg_durable
Like pg_duckdb, pg_durable loads through shared_preload_libraries, so it needs allow-listing and a restart:
1. Add pg_durable to azure.extensions (the allow-list).
2. Add pg_durable to shared_preload_libraries (append to the current value, next to pg_duckdb) and restart the server.
Then create both extensions in the database:
CREATE EXTENSION IF NOT EXISTS pg_duckdb;
CREATE EXTENSION IF NOT EXISTS pg_durable;
SELECT duckdb.create_azure_secret('<blob connection string>');
pg_durable runs its steps in a background worker. Right after a restart that worker takes a moment to come up. The first df.start() can return pg_durable background worker not yet initialized — try again in a moment. Wait a few seconds and retry. A normal (non-superuser) login also needs a one-time grant:
SELECT df.grant_usage(current_user);
A quick smoke test confirms the engine runs:
postgres=> SELECT df.start('SELECT 1 AS ok')
;
start
----------
50a27f7d
(1 row)
postgres=> SELECT df.status('50a27f7d')
;
status
-----------
completed
(1 row)
postgres=> SELECT df.result('50a27f7d')
;
result
--------------------------------
{"rows": [{"ok": 1}], "row_count": 1}
(1 row)
The workflow status, errors, and results are persisted for each step and you can query them with pg_durable functions and tables.
The data to archive
A small orders table spanning three years — the 2023 rows are the cold slice we'll move out:
DROP TABLE IF EXISTS orders;
CREATE TABLE orders AS
SELECT g AS order_id,
(DATE '2023-01-01' + ((g * 7) % 1095))::date AS order_date,
((g % 500) + 1)::numeric(10,2) AS amount
FROM generate_series(1, 1000) g;
CREATE TABLE IF NOT EXISTS archive_log (
id serial,
archived_year int,
n_rows bigint,
blob text,
at timestamptz DEFAULT now()
);
Here is the current data in this table:
postgres=> -- total number of rows
postgres=> SELECT count(*) FROM orders;
count
-------
1000
postgres=> -- the cold slice
postgres=> SELECT count(*) FROM orders
WHERE order_date < DATE '2024-01-01'
;
count
-------
364
I have a reproducible test case where I want to archive orders from before 2024. I'll create a one-time workflow for that and let it run.
The durable workflow
The whole archive is one df.start():
- steps are chained with ~>
- step's result is captured into a variable with |=> 'name'
- next step can reuse the result as $name
- if-then-else is declared with (), ?>, !>
SELECT df.start(
-- 1. count the cold slice, capture it as $n
$Step1$
SELECT count(*) AS n FROM orders WHERE order_date < DATE '2024-01-01'
$Step1$ |=> 'n'
-- 2. write those rows to Blob as Parquet, through the DuckDB engine
~> $Step2$
SELECT duckdb.raw_query(
$DuckDB$
COPY (
SELECT order_id, order_date, amount
FROM pgduckdb.public.orders
WHERE order_date < DATE '2024-01-01'
) TO 'az://lake/archive/orders_2023.parquet' (FORMAT parquet)
$DuckDB$
)
$Step2$
-- 3. verify the file reads back, capture the row count as $v
~> $Step3$
SELECT count(*) AS v
FROM read_parquet('az://lake/archive/orders_2023.parquet')
$Step3$ |=> 'v'
-- 4. delete only if the archive count matches
~> ( $Check$ SELECT $v = $n AS archive_complete $Check$
?> $Del$ DELETE FROM orders WHERE order_date < DATE '2024-01-01' $Del$
!> $Fail$ DO $do$ BEGIN
RAISE EXCEPTION 'archive incomplete: expected % rows, file has %', $n, $v;
END $do$;
$Fail$
)
-- 5. record what happened
~> $Step5$
INSERT INTO archive_log(archived_year, n_rows, blob)
VALUES (2023, $n, 'az://lake/archive/orders_2023.parquet')
$Step5$,
'durable-archive' -- a label to find the run by
);
Two things about the write step (step 2), both carried over from the archive post:
- The blob write goes through duckdb.raw_query(), not plain COPY TO 'az://…' so that DuckDB authorizes the write according to its own secret and allow-list — the object-storage controls that PostgreSQL's server-file gate knows nothing about and would otherwise reject.
- Inside raw_query you're in DuckDB's catalog, so the table is pgduckdb.public.orders, not a bare orders.
The condition is a safety check to run the delete only when the count matches, or fail otherwise.
And a word on the quoting: the steps are SQL statements declared as text arguments for df.start() and themselves include nested text. This would require multiple levels of double quotes in standard SQL. It is solved easily with PostgreSQL dollar quoting ($$ with an optional label inside).
Watching it run
df.start() returns immediately with an instance id. The worker executes the steps. You can check progress directly with the id or getting the last id with your label:
postgres=> SELECT instance_id, label, status, execution_count
FROM df.list_instances()
WHERE instance_id = (SELECT id FROM df.instances
WHERE label = 'durable-archive'
ORDER BY created_at DESC LIMIT 1)
;
instance_id | label | status | execution_count
-------------+-----------------+-----------+-----------------
b9b24177 | durable-archive | completed | 1
(1 row)
The steps are visible with df.explain():
postgres=> SELECT df.explain('b9b24177')
;
explain
-----------------------------------------------------------------------
Instance: b9b24177 (durable-archive) +
Status: ✓ Completed +
Output: {"rows":[],"row_count":0} +
+
SQL: +
SELECT count(*) AS n FROM orders WHER... |=> 'n' ✓ +
→ SQL: +
SELECT duckdb.raw_query( +
$Du... ✓ +
→ SQL: +
SELECT count(*) AS v +
FROM re... ✓ +
→ IF ✓ +
? condition: +
└─SQL: SELECT $v = $n AS archive_complete ✓ +
✓ then: +
└─SQL: DELETE FROM orders WHERE order_date < DATE '2... ✓ +
✗ else: +
└─SQL: DO $do$ BEGIN +
RAISE EXCEPT... ⊘ +
→ SQL: +
INSERT INTO archive_log(archived_year... ✓
(1 row)
Each step is a node you can inspect — this is the checkpoint record:
postgres=> SELECT node_type, status, query
FROM df.nodes
WHERE instance_id = (SELECT id FROM df.instances
WHERE label = 'durable-archive'
ORDER BY created_at DESC LIMIT 1)
ORDER BY updated_at
;
node_type | status | query
-----------+-----------+------------------------------------------------------------------------------------------------
SQL | pending | DO $do$ BEGIN +
| | RAISE EXCEPTION 'archive incomplete: expected % rows, file has %', $n, $v; +
| | END $do$; +
| |
SQL | completed | +
| | SELECT count(*) AS n FROM orders WHERE order_date < DATE '2024-01-01' +
| |
SQL | completed | +
| | SELECT duckdb.raw_query( +
| | $DuckDB$ +
| | COPY ( +
| | SELECT order_id, order_date, amount +
| | FROM pgduckdb.public.orders +
| | WHERE order_date < DATE '2024-01-01' +
| | ) TO 'az://lake/archive/orders_2023.parquet' (FORMAT parquet) +
| | $DuckDB$ +
| | ) +
| |
THEN | completed |
SQL | completed | +
| | SELECT count(*) AS v +
| | FROM read_parquet('az://lake/archive/orders_2023.parquet') +
| |
THEN | completed |
SQL | completed | SELECT $v = $n AS archive_complete
SQL | completed | DELETE FROM orders WHERE order_date < DATE '2024-01-01'
IF | completed | {"condition_node":"3149a512"}
THEN | completed |
SQL | completed | +
| | INSERT INTO archive_log(archived_year, n_rows, blob) +
| | VALUES (2023, $n, 'az://lake/archive/orders_2023.parquet') +
| |
THEN | completed |
(12 rows)
The gate adds nodes: the SELECT $v = $n AS archive_complete check, its IF node, and both branches — the DELETE ran and is completed, while the DO ... RAISE EXCEPTION branch stayed pending because it was never taken.
It worked, end to end
As expected by the completed workflow, the archived file is in Blob that I can query though pg_duckdb functions:
postgres=> SELECT count(*) FROM read_parquet('az://lake/archive/orders_2023.parquet')
;
count
-------
364
(1 row)
The cold rows are gone from Postgres:
postgres=> SELECT count(*), min(order_date) FROM orders
;
count | min
-------+------------
636 | 2024-01-01
(1 row)
The audit row is written in my custom table:
postgres=> SELECT archived_year, n_rows, blob FROM archive_log ORDER BY at
;
archived_year | n_rows | blob
---------------+--------+---------------------------------------
2023 | 364 | az://lake/archive/orders_2023.parquet
(1 row)
As we have seen in the previous post, history is still one query away, live rows unioned with the archived file, because pg_duckdb can run queries on heterogenous sources:
postgres=> SELECT yr, count(*) AS n
FROM (
SELECT extract(year FROM order_date)::int AS yr FROM orders
UNION ALL
SELECT extract(year FROM (r['order_date'])::date)::int
FROM read_parquet('az://lake/archive/orders_2023.parquet') r
) u
GROUP BY yr ORDER BY yr
;
yr | n
------+-----
2023 | 364
2024 | 324
2025 | 312
(3 rows)
2023 comes from the archived Blob file and 2024 and 2025 from the live Postgres table.
Why durable, not just a cron job
Steps 2→3→4 are export, verify, then delete — and because pg_durable checkpoints after each step, a crash between them is safe:
- Die after the export but before the delete, and the run resumes at the verify and delete steps on restart. The export is not repeated. The delete is not skipped.
- A failed step stops the workflow with its state intact in df.nodes, so you can see exactly where it stopped and why — no guessing, no half-done cleanup.
With a plain pg_cron + PL/pgSQL procedure you'd be rebuilding this yourself: a status table, retry counters, and a way to tell "exported but not yet deleted" apart from "nothing done yet." pg_durable keeps that bookkeeping inside Postgres.
Make it nightly
Wrap the same workflow in an eternal loop (@>) that waits on a cron schedule, and the archive runs on its own, forever, surviving restarts. Point the date window at a rolling boundary instead of a fixed year:
SELECT df.start(
@> (
$Step0$
SELECT (date_trunc('year', now()) - interval '1 year')::date AS cutoff
$Step0$ |=> 'cutoff'
~> $Step1$
SELECT count(*) AS n FROM orders WHERE order_date < $cutoff
$Step1$ |=> 'n'
-- copy / verify
~> $View$
CREATE OR REPLACE VIEW orders_all AS
SELECT order_id, order_date, amount FROM orders
WHERE order_date >= $cutoff -- exclude rows about to be deleted
UNION ALL
SELECT order_id, order_date, amount
FROM read_parquet('az://lake/archive/orders_*.parquet')
$View$
-- the same ( check ?> delete !> fail ) gate, then log — all referencing $cutoff
~> df.wait_for_schedule('0 2 * * *')
),
'nightly-archive'
);
I added a step to (re)build a view that reads from the archives and adds the operational data. I'll explain how to query a partitioned data lake in a future post.
Takeaways
- pg_durable is another open-source extension on Azure Database for PostgreSQL, enabled like pg_duckdb (azure.extensions + shared_preload_libraries + restart), usable in your database after CREATE EXTENSION pg_durable, and by your user after df.grant_usage(current_user).
- The archive becomes one df.start() graph: count → write to Blob (duckdb.raw_query) → verify → delete → log, checkpointed between steps.
- Durability closes the export-then-delete gap: a crash resumes at the next uncompleted step instead of re-exporting or deleting unsafely.
- @> + df.wait_for_schedule('0 2 * * *') turns it into a nightly job that survives restarts — no external scheduler, no status table.
- Everything is plain SQL with additional operators to define the workflow.
- Each step is a transaction, so intermediate results are durable — the workflow continues automatically after a failover.
You can visualize the workflow from VS Code with the PostgreSQL extension by Microsoft
Mentions
Automate Postgres-to-Blob with durable workflow (pg_durable + pg_duckdb) in Azure Database for PostgreSQL In the archive post I moved cold rows out of a Postgres table into Parquet on Azure Blob and kept querying them. The steps were manual: export to Blob, verify the file, delete the rows, record what happened. If the process dies after the export but before the delete, you're stuck — re-run and you re-export (wasteful), skip ahead and you might delete rows that were never safely written (dangerous). That partial-failure gap is exactly what a durable workflow closes. Azure Database for PostgreSQL ships another open-source Microsoft extension that fits here: pg_durable. It runs checkpointed, crash-safe workflows inside Postgres — define a graph of SQL steps, and the engine persists progress between them. If the database restarts or a step fails, execution resumes from the last completed step instead of starting over. No cron-plus-status-table-plus-worker scaffolding. This post turns the archive into one durable, schedulable workflow, with pg_duckdb doing the write to Blob. Enabling pg_durable Like pg_duckdb, pg_durable loads through shared_preload_libraries, so it needs allow-listing and a restart: 1. Add pg_durable to azure.extensions (the allow-list). 2. Add pg_durable to shared_preload_libraries (append to the current value, next to pg_duckdb) and restart the server. Then create both extensions in the database: CREATE EXTENSION IF NOT EXISTS pg_duckdb; CREATE EXTENSION IF NOT EXISTS pg_durable; SELECT duckdb.create_azure_secret('<blob connection string>'); pg_durable runs its steps in a background worker. Right after a restart that worker takes a moment to come up. The first df.start() can return pg_durable background worker not yet initialized — try again in a moment. Wait a few seconds and retry. A normal (non-superuser) login also needs a one-time grant: SELECT df.grant_usage(current_user); A quick smoke test confirms the engine runs: postgres=> SELECT df.start('SELECT 1 AS ok') ; start ---------- 50a27f7d (1 row) postgres=> SELECT df.status('50a27f7d') ; status ----------- completed (1 row) postgres=> SELECT df.result('50a27f7d') ; result -------------------------------- {"rows": [{"ok": 1}], "row_count": 1} (1 row) The workflow status, errors, and results are persisted for each step and you can query them with pg_durable functions and tables. The data to archive A small orders table spanning three years — the 2023 rows are the cold slice we'll move out: DROP TABLE IF EXISTS orders; CREATE TABLE orders AS SELECT g AS order_id, (DATE '2023-01-01' + ((g * 7) % 1095))::date AS order_date, ((g % 500) + 1)::numeric(10,2) AS amount FROM generate_series(1, 1000) g; CREATE TABLE IF NOT EXISTS archive_log ( id serial, archived_year int, n_rows bigint, blob text, at timestamptz DEFAULT now() ); Here is the current data in this table: postgres=> -- total number of rows postgres=> SELECT count(*) FROM orders; count ------- 1000 postgres=> -- the cold slice postgres=> SELECT count(*) FROM orders WHERE order_date < DATE '2024-01-01' ; count ------- 364 I have a reproducible test case where I want to archive orders from before 2024. I'll create a one-time workflow for that and let it run. The durable workflow The whole archive is one df.start(): - steps are chained with ~> - step's result is captured into a variable with |=> 'name' - next step can reuse the result as $name - if-then-else is declared with (), ?>, !> SELECT df.start( -- 1. count the cold slice, capture it as $n $Step1$ SELECT count(*) AS n FROM orders WHERE order_date < DATE '2024-01-01' $Step1$ |=> 'n' -- 2. write those rows to Blob as Parquet, through the DuckDB engine ~> $Step2$ SELECT duckdb.raw_query( $DuckDB$ COPY ( SELECT order_id, order_date, amount FROM pgduckdb.public.orders WHERE order_date < DATE '2024-01-01' ) TO 'az://lake/archive/orders_2023.parquet' (FORMAT parquet) $DuckDB$ ) $Step2$ -- 3. verify the file reads back, capture the row count as $v ~> $Step3$ SELECT count(*) AS v FROM read_parquet('az://lake/archive/orders_2023.parquet') $Step3$ |=> 'v' -- 4. delete only if the archive count matches ~> ( $Check$ SELECT $v = $n AS archive_complete $Check$ ?> $Del$ DELETE FROM orders WHERE order_date < DATE '2024-01-01' $Del$ !> $Fail$ DO $do$ BEGIN RAISE EXCEPTION 'archive incomplete: expected % rows, file has %', $n, $v; END $do$; $Fail$ ) -- 5. record what happened ~> $Step5$ INSERT INTO archive_log(archived_year, n_rows, blob) VALUES (2023, $n, 'az://lake/archive/orders_2023.parquet') $Step5$, 'durable-archive' -- a label to find the run by ); Two things about the write step (step 2), both carried over from the archive post: - The blob write goes through duckdb.raw_query(), not plain COPY TO 'az://…' so that DuckDB authorizes the write according to its own secret and allow-list — the object-storage controls that PostgreSQL's server-file gate knows nothing about and would otherwise reject. - Inside raw_query you're in DuckDB's catalog, so the table is pgduckdb.public.orders, not a bare orders. The condition is a safety check to run the delete only when the count matches, or fail otherwise. And a word on the quoting: the steps are SQL statements declared as text arguments for df.start() and themselves include nested text. This would require multiple levels of double quotes in standard SQL. It is solved easily with PostgreSQL dollar quoting ($$ with an optional label inside). Watching it run df.start() returns immediately with an instance id. The worker executes the steps. You can check progress directly with the id or getting the last id with your label: postgres=> SELECT instance_id, label, status, execution_count FROM df.list_instances() WHERE instance_id = (SELECT id FROM df.instances WHERE label = 'durable-archive' ORDER BY created_at DESC LIMIT 1) ; instance_id | label | status | execution_count -------------+-----------------+-----------+----------------- b9b24177 | durable-archive | completed | 1 (1 row) The steps are visible with df.explain(): postgres=> SELECT df.explain('b9b24177') ; explain ----------------------------------------------------------------------- Instance: b9b24177 (durable-archive) + Status: ✓ Completed + Output: {"rows":[],"row_count":0} + + SQL: + SELECT count(*) AS n FROM orders WHER... |=> 'n' ✓ + → SQL: + SELECT duckdb.raw_query( + $Du... ✓ + → SQL: + SELECT count(*) AS v + FROM re... ✓ + → IF ✓ + ? condition: + └─SQL: SELECT $v = $n AS archive_complete ✓ + ✓ then: + └─SQL: DELETE FROM orders WHERE order_date < DATE '2... ✓ + ✗ else: + └─SQL: DO $do$ BEGIN + RAISE EXCEPT... ⊘ + → SQL: + INSERT INTO archive_log(archived_year... ✓ (1 row) Each step is a node you can inspect — this is the checkpoint record: postgres=> SELECT node_type, status, query FROM df.nodes WHERE instance_id = (SELECT id FROM df.instances WHERE label = 'durable-archive' ORDER BY created_at DESC LIMIT 1) ORDER BY updated_at ; node_type | status | query -----------+-----------+------------------------------------------------------------------------------------------------ SQL | pending | DO $do$ BEGIN + | | RAISE EXCEPTION 'archive incomplete: expected % rows, file has %', $n, $v; + | | END $do$; + | | SQL | completed | + | | SELECT count(*) AS n FROM orders WHERE order_date < DATE '2024-01-01' + | | SQL | completed | + | | SELECT duckdb.raw_query( + | | $DuckDB$ + | | COPY ( + | | SELECT order_id, order_date, amount + | | FROM pgduckdb.public.orders + | | WHERE order_date < DATE '2024-01-01' + | | ) TO 'az://lake/archive/orders_2023.parquet' (FORMAT parquet) + | | $DuckDB$ + | | ) + | | THEN | completed | SQL | completed | + | | SELECT count(*) AS v + | | FROM read_parquet('az://lake/archive/orders_2023.parquet') + | | THEN | completed | SQL | completed | SELECT $v = $n AS archive_complete SQL | completed | DELETE FROM orders WHERE order_date < DATE '2024-01-01' IF | completed | {"condition_node":"3149a512"} THEN | completed | SQL | completed | + | | INSERT INTO archive_log(archived_year, n_rows, blob) + | | VALUES (2023, $n, 'az://lake/archive/orders_2023.parquet') + | | THEN | completed | (12 rows) The gate adds nodes: the SELECT $v = $n AS archive_complete check, its IF node, and both branches — the DELETE ran and is completed, while the DO ... RAISE EXCEPTION branch stayed pending because it was never taken. It worked, end to end As expected by the completed workflow, the archived file is in Blob that I can query though pg_duckdb functions: postgres=> SELECT count(*) FROM read_parquet('az://lake/archive/orders_2023.parquet') ; count ------- 364 (1 row) The cold rows are gone from Postgres: postgres=> SELECT count(*), min(order_date) FROM orders ; count | min -------+------------ 636 | 2024-01-01 (1 row) The audit row is written in my custom table: postgres=> SELECT archived_year, n_rows, blob FROM archive_log ORDER BY at ; archived_year | n_rows | blob ---------------+--------+--------------------------------------- 2023 | 364 | az://lake/archive/orders_2023.parquet (1 row) As we have seen in the previous post, history is still one query away, live rows unioned with the archived file, because pg_duckdb can run queries on heterogenous sources: postgres=> SELECT yr, count(*) AS n FROM ( SELECT extract(year FROM order_date)::int AS yr FROM orders UNION ALL SELECT extract(year FROM (r['order_date'])::date)::int FROM read_parquet('az://lake/archive/orders_2023.parquet') r ) u GROUP BY yr ORDER BY yr ; yr | n ------+----- 2023 | 364 2024 | 324 2025 | 312 (3 rows) 2023 comes from the archived Blob file and 2024 and 2025 from the live Postgres table. Why durable, not just a cron job Steps 2→3→4 are export, verify, then delete — and because pg_durable checkpoints after each step, a crash between them is safe: - Die after the export but before the delete, and the run resumes at the verify and delete steps on restart. The export is not repeated. The delete is not skipped. - A failed step stops the workflow with its state intact in df.nodes, so you can see exactly where it stopped and why — no guessing, no half-done cleanup. With a plain pg_cron + PL/pgSQL procedure you'd be rebuilding this yourself: a status table, retry counters, and a way to tell "exported but not yet deleted" apart from "nothing done yet." pg_durable keeps that bookkeeping inside Postgres. Make it nightly Wrap the same workflow in an eternal loop (@>) that waits on a cron schedule, and the archive runs on its own, forever, surviving restarts. Point the date window at a rolling boundary instead of a fixed year: SELECT df.start( @> ( $Step0$ SELECT (date_trunc('year', now()) - interval '1 year')::date AS cutoff $Step0$ |=> 'cutoff' ~> $Step1$ SELECT count(*) AS n FROM orders WHERE order_date < $cutoff $Step1$ |=> 'n' -- copy / verify ~> $View$ CREATE OR REPLACE VIEW orders_all AS SELECT order_id, order_date, amount FROM orders WHERE order_date >= $cutoff -- exclude rows about to be deleted UNION ALL SELECT order_id, order_date, amount FROM read_parquet('az://lake/archive/orders_*.parquet') $View$ -- the same ( check ?> delete !> fail ) gate, then log — all referencing $cutoff ~> df.wait_for_schedule('0 2 * * *') ), 'nightly-archive' ); I added a step to (re)build a view that reads from the archives and adds the operational data. I'll explain how to query a partitioned data lake in a future post. Takeaways - pg_durable is another open-source extension on Azure Database for PostgreSQL, enabled like pg_duckdb (azure.extensions + shared_preload_libraries + restart), usable in your database after CREATE EXTENSION pg_durable, and by your user after df.grant_usage(current_user). - The archive becomes one df.start() graph: count → write to Blob (duckdb.raw_query) → verify → delete → log, checkpointed between steps. - Durability closes the export-then-delete gap: a crash resumes at the next uncompleted step instead of re-exporting or deleting unsafely. - @> + df.wait_for_schedule('0 2 * * *') turns it into a nightly job that survives restarts — no external scheduler, no status table. - Everything is plain SQL with additional operators to define the workflow. - Each step is a transaction, so intermediate results are durable — the workflow continues automatically after a failover. You can visualize the workflow from VS Code with the PostgreSQL extension by Microsoft