Short answer - The docs’ data-retention script deletes flow runs but does not rely on a database cascade to remove logs. - In Prefect 3.x, log rows are not FK-cascaded from

flow_run

or

task_run

. Instead, when you delete a flow run via the API, the server queues a background task to delete the associated logs by

flow_run_id

. - If you’re deleting runs directly in SQL (or the server isn’t running long enough for the background task), logs will remain. References - Docs: https://docs.prefect.io/v3/advanced/database-maintenance#data-retention - Flow run delete enqueues log cleanup: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/api/flow_runs.py - Log model (no FK cascade; columns include

flow_run_id

and

task_run_id

): https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/database/orm_models.py What this means for your script - If your script uses the Prefect client to call

delete_flow_run(...)

against a running Prefect server, the server should enqueue deletion of logs for that

flow_run_id

. If you still see logs, likely causes are: - Runs were deleted via direct SQL (bypassing the API), or - The background cleanup task did not run/complete, or - Some log rows only reference

task_run_id

(rare) and not

flow_run_id

. If you want to explicitly delete logs with your retention job (PostgreSQL) Below are safe SQL snippets you can add to your retention routine so logs are removed even when you delete runs directly in the database. These also cover logs that are only linked by

task_run_id

. 1) Preview counts first

-- Logs for flow runs older than 30 days (completed/failed/cancelled)
SELECT COUNT(*) FROM log
WHERE flow_run_id IN (
  SELECT id FROM flow_run
  WHERE created < NOW() - INTERVAL '30 days'
    AND state_type IN ('COMPLETED','FAILED','CANCELLED')
);

-- Logs tied to task runs from those flow runs
SELECT COUNT(*) FROM log
WHERE task_run_id IN (
  SELECT id FROM task_run
  WHERE flow_run_id IN (
    SELECT id FROM flow_run
    WHERE created < NOW() - INTERVAL '30 days'
      AND state_type IN ('COMPLETED','FAILED','CANCELLED')
  )
);

2) Delete logs first, then delete runs (CTE form)

WITH old_flow_runs AS (
  SELECT id FROM flow_run
  WHERE created < NOW() - INTERVAL '30 days'
    AND state_type IN ('COMPLETED','FAILED','CANCELLED')
),
old_task_runs AS (
  SELECT id FROM task_run
  WHERE flow_run_id IN (SELECT id FROM old_flow_runs)
)
-- Delete logs for those flow runs and their task runs
DELETE FROM log
WHERE flow_run_id IN (SELECT id FROM old_flow_runs)
   OR task_run_id IN (SELECT id FROM old_task_runs);

-- Optionally delete artifacts tied to those runs (also not FK-cascaded)
DELETE FROM artifact
WHERE flow_run_id IN (SELECT id FROM old_flow_runs)
   OR task_run_id IN (SELECT id FROM old_task_runs);

-- Finally delete the flow runs (FK cascades will take care of states, etc.)
DELETE FROM flow_run
WHERE id IN (SELECT id FROM old_flow_runs);

3) For very large datasets, batch the deletion

-- Repeat in a loop until 0 rows are affected
WITH doomed AS (
  SELECT l.id
  FROM log l
  WHERE l.flow_run_id IN (
    SELECT id FROM flow_run
    WHERE created < NOW() - INTERVAL '30 days'
      AND state_type IN ('COMPLETED','FAILED','CANCELLED')
  )
  OR l.task_run_id IN (
    SELECT tr.id
    FROM task_run tr
    JOIN flow_run fr ON tr.flow_run_id = fr.id
    WHERE fr.created < NOW() - INTERVAL '30 days'
      AND fr.state_type IN ('COMPLETED','FAILED','CANCELLED')
  )
  LIMIT 100000
)
DELETE FROM log WHERE id IN (SELECT id FROM doomed);

SQLite notes - Use

date('now','-30 days')

instead of

NOW() - INTERVAL '30 days'

. - Batch sizes should be smaller; WAL mode can help:

PRAGMA journal_mode = WAL;

. A few checks to consider - Make sure your retention job uses the Prefect API to delete runs if you want the server to do the log cleanup automatically.

- If you prefer direct SQL (for speed/operational simplicity), include the explicit

DELETE FROM log ...

as shown above. - If you also want to reclaim space from artifacts, include the optional artifacts deletion. Questions to help me tailor this for you - Which database are you using (Postgres vs SQLite)? - Are you deleting runs through the Prefect API/client, or via direct SQL? - Is your Prefect server running at the time you run the retention job and for some time after (so background cleanup tasks can complete)?

Great, thanks for the details. Direct answer - Prefect 3.x: When you delete a flow run via the API, the server enqueues a background task that deletes logs for that

flow_run_id

. There is no DB-level FK cascade for logs. - Prefect 2.x: There is no background task or cascade that deletes logs when a flow run is deleted. Deleting a flow run via the API removes the run and cascaded states/task runs, but logs remain. Key references - 3.x flow run delete enqueues log cleanup: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/api/flow_runs.py - 3.x log model (no FK cascade): https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/database/orm_models.py - 2.x flow run delete (no log cleanup): https://github.com/PrefectHQ/prefect/blob/2.14.24/src/prefect/server/models/flow_runs.py - 2.x logs model (no FK, no delete helper): https://github.com/PrefectHQ/prefect/blob/2.14.24/src/prefect/server/database/orm_models.py - 2.x logs API (no DELETE): https://github.com/PrefectHQ/prefect/blob/2.14.24/src/prefect/server/api/logs.py What to do on Prefect 2.x (Postgres) If you need to prune logs alongside flow run deletion in 2.x, add explicit deletes before/after deleting runs. Example pattern: - Preview counts

SELECT COUNT(*) FROM log
WHERE flow_run_id IN (
  SELECT id FROM flow_run
  WHERE created < NOW() - INTERVAL '30 days'
    AND state_type IN ('COMPLETED','FAILED','CANCELLED')
)
OR task_run_id IN (
  SELECT tr.id FROM task_run tr
  JOIN flow_run fr ON tr.flow_run_id = fr.id
  WHERE fr.created < NOW() - INTERVAL '30 days'
    AND fr.state_type IN ('COMPLETED','FAILED','CANCELLED')
);

- Delete logs, then delete runs (or do logs after if you rely on the API to delete runs)

WITH old_flow_runs AS (
  SELECT id FROM flow_run
  WHERE created < NOW() - INTERVAL '30 days'
    AND state_type IN ('COMPLETED','FAILED','CANCELLED')
),
old_task_runs AS (
  SELECT id FROM task_run
  WHERE flow_run_id IN (SELECT id FROM old_flow_runs)
)
DELETE FROM log
WHERE flow_run_id IN (SELECT id FROM old_flow_runs)
   OR task_run_id IN (SELECT id FROM old_task_runs);

DELETE FROM flow_run
WHERE id IN (SELECT id FROM old_flow_runs);

- For large tables, batch the log deletion with a LIMIT loop to avoid long locks:

WITH doomed AS (
  SELECT l.id
  FROM log l
  WHERE l.flow_run_id IN (SELECT id FROM old_flow_runs)
     OR l.task_run_id IN (SELECT id FROM old_task_runs)
  LIMIT 100000
)
DELETE FROM log WHERE id IN (SELECT id FROM doomed);
-- Repeat until 0 rows affected

If you’re staying on 2.x but want to keep using the API for run deletion, you can still run the SQL log-prune job on a schedule to clean up orphaned logs. If/when you move to 3.x Your API-triggered deletions will also queue log cleanup. If you still want belt-and-suspenders, you can keep an explicit SQL cleanup job, but it shouldn’t be necessary if the server remains healthy and running.

Short answer: No. Prefect does not expose an API endpoint to execute arbitrary SQL or to bulk-delete logs. The only public log endpoints are for creating and reading logs; deletion is internal and only triggered when you delete runs via the API in Prefect 3.x. There’s no admin/raw-SQL endpoint in either 2.x or 3.x. Your options - Prefect 3.x: Keep deleting flow runs via the API; the server will queue log cleanup for those runs. If you still see logs piling up, we can investigate why the background task isn’t completing. - Prefect 2.x: There’s no API-based log cleanup. You’ll need to connect to Postgres and run SQL explicitly. If you want to orchestrate SQL cleanup with Prefect (without a separate cron), you can run a small maintenance flow that connects to Postgres and executes the deletes on a schedule. Example (psycopg 3):

import os
import psycopg
from prefect import flow, get_run_logger

RETENTION_DAYS = int(os.getenv("RETENTION_DAYS", "30"))
BATCH_SIZE = int(os.getenv("BATCH_SIZE", "100000"))
STATE_TYPES = ("COMPLETED","FAILED","CANCELLED")  # adjust if needed

SQL_BATCH_DELETE = f"""
WITH old_flow_runs AS (
  SELECT id FROM flow_run
  WHERE created < NOW() - %s::interval
    AND state_type = ANY(%s)
),
old_task_runs AS (
  SELECT id FROM task_run
  WHERE flow_run_id IN (SELECT id FROM old_flow_runs)
),
doomed AS (
  SELECT id FROM log
  WHERE flow_run_id IN (SELECT id FROM old_flow_runs)
     OR task_run_id IN (SELECT id FROM old_task_runs)
  LIMIT %s
)
DELETE FROM log WHERE id IN (SELECT id FROM doomed)
RETURNING id;
"""

@flow
def prune_prefect_logs():
    logger = get_run_logger()
    dsn = os.environ["PREFECT_DB_DSN"]  # e.g. <postgresql://user:pass@host:5432/prefect>
    interval = f"{RETENTION_DAYS} days"

    total = 0
    with psycopg.connect(dsn) as conn:
        with conn.cursor() as cur:
            # optional safety
            cur.execute("SET lock_timeout = '5s';")
            while True:
                cur.execute(SQL_BATCH_DELETE, (interval, list(STATE_TYPES), BATCH_SIZE))
                deleted = cur.rowcount
                conn.commit()
                total += deleted
                <http://logger.info|logger.info>(f"Deleted {deleted} log rows in this batch...")
                if deleted == 0:
                    break

    <http://logger.info|logger.info>(f"Total deleted log rows: {total}")

Notes - Store

PREFECT_DB_DSN

in a secret (e.g., environment variable or a Prefect Block) and deploy this as a scheduled maintenance flow. - You can adapt the same pattern to delete artifacts as well if desired. - For very large datasets, keep batch sizes modest to reduce lock time. References - 3.x log endpoints (no delete): https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/api/logs.py - 3.x flow-run delete enqueues log cleanup: https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/api/flow_runs.py - 3.x log model (no FK cascade): https://github.com/PrefectHQ/prefect/blob/main/src/prefect/server/database/orm_models.py