DuckDB Index and Storage Growth Gate with Kestra

id: duckdb-index-storage-growth-gate
namespace: company.team
description: |
  Query DuckDB's own duckdb_tables()/duckdb_indexes() catalog functions and
  pragma_database_size() to classify a table HEALTHY, AT_RISK (missing
  index or oversized database file), or UNREACHABLE (table not found),
  alert Slack, and optionally create the missing index with CREATE INDEX.
  auto_remediate and dry_run are two independent gates: no row in the
  table is ever read, inserted, or deleted - only the catalog is touched.

triggers:
  - id: periodic_governance_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check every 6 hours - index coverage and file growth both drift
      slowly compared to a data-freshness check. Shipped disabled; enable it
      once database_path/target_table point at a real DuckDB file and a manual
      run has been verified. Always fires in safe mode (auto_remediate false,
      dry_run true) regardless of this flow's own defaults.
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

inputs:
  - id: database_path
    type: STRING
    displayName: "DuckDB Database File Path"
    defaults: "/tmp/kestra-duckdb/analytics.duckdb"
    description: Local file path for the persistent DuckDB database, rendered into
      jdbc:duckdb:{{ database_path }} for every Query task in this flow.

  - id: target_table
    type: STRING
    displayName: "Target Table"
    defaults: "events"
    description: Table audited in duckdb_tables() and duckdb_indexes().

  - id: required_indexed_column
    type: STRING
    displayName: "Required Indexed Column"
    defaults: "event_id"
    description: Column expected to be covered by an index on target_table. Checked
      by casting duckdb_indexes().expressions to text and searching for this
      column name, since this flow does not assume a specific list type for that
      column.

  - id: max_used_blocks
    type: INT
    displayName: "Max Used Storage Blocks"
    defaults: 100000
    description: Above this, pragma_database_size().used_blocks for the database
      counts as AT_RISK for storage growth. Multiply by
      pragma_database_size().block_size yourself to convert to bytes; block_size
      can vary by DuckDB version and is not assumed fixed here.

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Remediate"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: If "true" and target_table is missing an index on
      required_indexed_column, create one with CREATE INDEX (still gated by
      dry_run below). If "false", the gap is only reported. Has no effect when
      the AT_RISK state is caused only by storage growth, since this flow does
      not compact or vacuum the database file.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: If "true", remediation is only logged, CREATE INDEX never runs.
      Must be explicitly "false", together with auto_remediate "true", for the
      index to actually be created.

tasks:
  - id: check_table_and_index
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: "duckdb_tables() and duckdb_indexes() are DuckDB's own catalog
      table functions (columns verified from the official DuckDB
      metadata-functions documentation: duckdb_tables() has table_name,
      estimated_size, index_count, column_count among others; duckdb_indexes()
      has table_name, index_name, expressions among others). FETCH_ONE returns
      at most one row for target_table; zero rows means the table does not exist
      in this database."
    url: "jdbc:duckdb:{{ inputs.database_path }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT
        t.table_name,
        t.estimated_size,
        t.index_count,
        t.column_count,
        (
          SELECT count(*)
          FROM duckdb_indexes() i
          WHERE i.table_name = '{{ inputs.target_table }}'
            AND i.expressions::VARCHAR LIKE '%{{ inputs.required_indexed_column }}%'
        ) AS matching_indexes
      FROM duckdb_tables() t
      WHERE t.table_name = '{{ inputs.target_table }}'

  - id: check_storage_size
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: "pragma_database_size() is DuckDB's own built-in table function
      (columns verified from the official DuckDB pragmas documentation:
      database_name, database_size, block_size, total_blocks, used_blocks,
      free_blocks, wal_size, memory_usage, memory_limit). LIMIT 1 assumes the
      common case of a single attached database on this connection; see Pitfalls
      for the multi-database case."
    url: "jdbc:duckdb:{{ inputs.database_path }}"
    fetchType: FETCH_ONE
    sql: "SELECT database_name, database_size, block_size, total_blocks,
      used_blocks, free_blocks FROM pragma_database_size() LIMIT 1"

  - id: evaluate_table_presence
    type: io.kestra.plugin.core.flow.If
    description: UNREACHABLE when duckdb_tables() returns no row for target_table -
      it does not exist in this database file.
    condition: "{{ outputs.check_table_and_index.row.table_name is defined }}"
    then:
      - id: evaluate_governance_risk
        type: io.kestra.plugin.core.flow.If
        description: AT_RISK when target_table has no index covering
          required_indexed_column, or when the database file's used_blocks
          exceeds max_used_blocks.
        condition: "{{ outputs.check_table_and_index.row.matching_indexes == 0 or
          outputs.check_storage_size.row.used_blocks > inputs.max_used_blocks
          }}"
        then:
          - id: route_remediation
            type: io.kestra.plugin.core.flow.Switch
            description: auto_remediate only decides whether remediation is considered at
              all; dry_run, checked separately below, decides whether CREATE
              INDEX actually runs.
            value: "{{ inputs.auto_remediate }}"
            cases:
              "true":
                - id: check_has_missing_index
                  type: io.kestra.plugin.core.flow.If
                  description: Only a missing index can be remediated by this flow - storage
                    growth from used_blocks is not addressed by CREATE INDEX and
                    needs a VACUUM or CHECKPOINT run by hand.
                  condition: "{{ outputs.check_table_and_index.row.matching_indexes == 0 }}"
                  then:
                    - id: check_dry_run
                      type: io.kestra.plugin.core.flow.If
                      condition: "{{ inputs.dry_run == \"false\" }}"
                      then:
                        - id: create_missing_index
                          type: io.kestra.plugin.jdbc.duckdb.Query
                          description: Standard DuckDB DDL. The generated index name is derived from table
                            and column so repeated runs after the index exists
                            are idempotent in intent, though this statement
                            itself has no IF NOT EXISTS guard - see Pitfalls.
                          url: "jdbc:duckdb:{{ inputs.database_path }}"
                          fetchType: NONE
                          sql: "CREATE INDEX idx_{{ inputs.target_table }}_{{
                            inputs.required_indexed_column }} ON {{
                            inputs.target_table }}({{
                            inputs.required_indexed_column }})"
                        - id: log_remediation_applied
                          type: io.kestra.plugin.core.log.Log
                          message: "Created index idx_{{ inputs.target_table }}_{{
                            inputs.required_indexed_column }} on {{
                            inputs.target_table }}({{
                            inputs.required_indexed_column }})."
                      else:
                        - id: log_dry_run_remediation
                          type: io.kestra.plugin.core.log.Log
                          message: "DRY RUN: would create an index on {{ inputs.target_table }}({{
                            inputs.required_indexed_column }}) (dry_run is
                            true). No index was created."
                  else:
                    - id: log_no_missing_index
                      type: io.kestra.plugin.core.log.Log
                      message: "AT_RISK with no missing index to create: the risk is storage growth
                        (used_blocks={{
                        outputs.check_storage_size.row.used_blocks }}, threshold
                        {{ inputs.max_used_blocks }}), which this flow cannot
                        remediate."
              "false":
                - id: log_remediation_disabled
                  type: io.kestra.plugin.core.log.Log
                  message: "auto_remediate is false: {{ inputs.target_table }} is AT_RISK
                    (matching_indexes={{
                    outputs.check_table_and_index.row.matching_indexes }},
                    used_blocks={{ outputs.check_storage_size.row.used_blocks
                    }}/{{ inputs.max_used_blocks }}), but no remediation was
                    attempted."
          - id: alert_governance_risk
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Always alert on AT_RISK, independent of whether remediation ran or
              was even possible.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "DuckDB governance gate: AT_RISK. {{ inputs.target_table }} matching_indexes={{ outputs.check_table_and_index.row.matching_indexes }} on {{ inputs.required_indexed_column }}, used_blocks={{ outputs.check_storage_size.row.used_blocks }} (threshold {{ inputs.max_used_blocks }}), database_size={{ outputs.check_storage_size.row.database_size }}. auto_remediate: {{ inputs.auto_remediate }}. dry_run: {{ inputs.dry_run }}. Remediated: {{ outputs.create_missing_index is defined }}. Execution {{ execution.id }}."
              }
        else:
          - id: log_governance_healthy
            type: io.kestra.plugin.core.log.Log
            message: "HEALTHY: {{ inputs.target_table }} has {{
              outputs.check_table_and_index.row.matching_indexes }} index(es)
              covering {{ inputs.required_indexed_column }}; used_blocks={{
              outputs.check_storage_size.row.used_blocks }} is within the {{
              inputs.max_used_blocks }} threshold."
    else:
      - id: alert_table_unreachable
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: UNREACHABLE means duckdb_tables() returned zero rows for
          target_table - distinct from both HEALTHY and AT_RISK.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "DuckDB governance gate: UNREACHABLE. {{ inputs.target_table }} has no row in duckdb_tables() for database {{ inputs.database_path }} - it may not exist yet, or the path may be wrong. Execution {{ execution.id }}."
          }

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final status line for every run, independent of which branch fired,
      so the execution history always shows the headline numbers.
    message: "Governance audit complete for {{ inputs.target_table }}.
      table_found={{ outputs.check_table_and_index.row.table_name is defined }}.
      auto_remediate={{ inputs.auto_remediate }}. dry_run={{ inputs.dry_run }}."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the flow itself fails outside the handled branches, for
      example a locked or corrupted database file.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "DuckDB governance gate FAILED in execution {{ execution.id }} for {{ inputs.target_table }} on {{ inputs.database_path }}. The index/storage audit did not complete; investigate before trusting the last known HEALTHY/AT_RISK/UNREACHABLE status."
      }

A table queried by a column with no backing index degrades the same way in DuckDB as in any SQL engine: scans get slower as the table grows, without ever throwing an error that points at the real cause. A DuckDB file that keeps growing without anyone checking pragma_database_size() is the other half of the same blind spot - both are easy to see with one query and easy to never check otherwise. This blueprint reads DuckDB's own catalog functions - duckdb_tables() and duckdb_indexes() for index coverage, pragma_database_size() for file size - classifies the result into a three-state risk level, alerts on anything but a clean bill of health, and can create a missing index with a plain CREATE INDEX statement, behind two independent gates.

How it works

periodic_governance_check (io.kestra.plugin.core.trigger.Schedule) runs every 6 hours, always forcing auto_remediate: "false" and dry_run: "true" through its own inputs: override regardless of this flow's defaults. Shipped disabled so you can validate a manual run first.
check_table_and_index (io.kestra.plugin.jdbc.duckdb.Query, fetchType: FETCH_ONE) joins duckdb_tables() against a correlated count from duckdb_indexes(), populating row.table_name, row.estimated_size, row.index_count, row.matching_indexes - or no row at all if target_table does not exist.
check_storage_size (io.kestra.plugin.jdbc.duckdb.Query, fetchType: FETCH_ONE) reads pragma_database_size(), populating row.database_size, row.block_size, row.used_blocks, row.total_blocks.
evaluate_table_presence (io.kestra.plugin.core.flow.If) branches on whether check_table_and_index returned a row.
- else (UNREACHABLE): alert_table_unreachable reports that target_table does not exist in this database.
- then (table found): nested evaluate_governance_risk (io.kestra.plugin.core.flow.If) branches on matching_indexes == 0 or used_blocks > max_used_blocks.
  - else (HEALTHY): log_governance_healthy records a clean check.
  - then (AT_RISK): route_remediation (io.kestra.plugin.core.flow.Switch on auto_remediate) either considers remediation ("true", further gated by check_has_missing_index - only a missing index can be created, not storage reclaimed - and then by dry_run through check_dry_run) or only logs that remediation is disabled ("false"). When every gate allows it, create_missing_index runs CREATE INDEX ... ON target_table(required_indexed_column). alert_governance_risk always fires to Slack regardless of which remediation path ran.
log_audit_complete always runs last, printing the headline numbers regardless of which branch fired.
The errors block alerts Slack separately if the flow itself fails outright - a locked or corrupted database file must not read as "no risk".

What you get

A direct answer, from DuckDB's own catalog functions, to "does this table actually have an index on this column" and "how big is this database file" - not inferred from query latency or file size on disk checked by hand.
A distinct UNREACHABLE state so a table that does not exist yet is never conflated with either a healthy or an at-risk table.
Remediation scoped to exactly one CREATE INDEX statement on the one table and column you named - it never touches row data or any other table.
Two independent gates (auto_remediate, dry_run) between "gap detected" and "index actually created".
A final status log every run, gap or not, so the execution history doubles as an index/storage-compliance trend for that table.

Who it's for

Data engineering teams running DuckDB as an embedded analytics or metrics-history store (the same kind of file this repo's own dbt-model-runtime-regression-alert.yaml writes to) who need an early signal when a hot query column was never indexed.
Teams running long-lived DuckDB files that only grow, never shrink, who want a threshold-based nudge before the file becomes unwieldy.
Anyone evaluating the DuckDB plugin who wants a worked example of Query driving catalog introspection (duckdb_tables(), duckdb_indexes(), pragma_database_size()) and a DDL statement (CREATE INDEX), not just file-based SQL analytics.

Why orchestrate this with Kestra

duckdb_tables() and pragma_database_size() run by hand answer one moment in time; they do not decide a cadence, classify the result into a three-state risk level, gate a schema change behind two independent confirmations, or notify anyone. Kestra supplies the schedule, the HEALTHY/AT_RISK/UNREACHABLE classification as a first-class branch, two independent authorization gates before any index is created, and an execution history that shows exactly when a table went uncovered or a file outgrew its budget.

Prerequisites

A persistent DuckDB database file (not an in-memory, execution-scoped database) that the Kestra Worker's filesystem can read and write, with target_table already created in it.
No credentials are required for DuckDB itself - it is an embedded, in-process engine with no server or authentication layer, unlike the other database plugins in this repo's governance-gate series.
A Slack incoming webhook for alerts.

Local testing: This flow has no Docker image of its own to run, since DuckDB is embedded directly inside the Kestra Worker process via its JDBC driver, not a separate server. docker network connect YOUR_KESTRA_NETWORK <container> and a host-port mapping away from 8080 therefore do not apply here - there is no container to connect. Instead, create the database file and table once, from any environment with a DuckDB client, for example: docker run --rm -v /tmp/kestra-duckdb:/data python:3.12-slim bash -c "pip install duckdb -q && python3 -c \"import duckdb; con = duckdb.connect('/data/analytics.duckdb'); con.execute('CREATE TABLE IF NOT EXISTS events (event_id VARCHAR, payload VARCHAR)')\"". Ensure database_path in this flow resolves to the same file from the Kestra Worker's own filesystem (mount the same host path into the Worker container if Kestra itself runs in Docker).

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook used by alert_governance_risk, alert_table_unreachable, and the errors block.
No database credentials are needed: DuckDB has no built-in authentication for local file access, so there is nothing to put behind {{ secret(...) }} for check_table_and_index, check_storage_size, or create_missing_index.
In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }} - for example SECRET_SLACK_WEBHOOK_URL=$(echo -n 'https://hooks.slack.com/...' | base64). This keeps the webhook out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Inputs

database_path (STRING, default /tmp/kestra-duckdb/analytics.duckdb): file path rendered into every task's url.
target_table (STRING, default events): table audited and, if remediated, altered with a new index.
required_indexed_column (STRING, default event_id): column expected to be index-covered.
max_used_blocks (INT, default 100000): storage-growth ceiling on pragma_database_size().used_blocks.
auto_remediate (SELECT: "false", "true"; default "false"): must be "true" for remediation to even be considered.
dry_run (SELECT: "true", "false"; default "true"): must be explicitly "false", together with auto_remediate: "true", for CREATE INDEX to actually run.

Outputs

outputs.check_table_and_index.row.table_name / .estimated_size / .index_count / .column_count / .matching_indexes: catalog facts for target_table, every run.
outputs.check_storage_size.row.database_size / .block_size / .total_blocks / .used_blocks / .free_blocks: file-size facts for the database, every run.

Quick start

Create a DuckDB file and an events table with no index on event_id, using the command in Prerequisites (or any DuckDB client pointed at the same file).
Add the SLACK_WEBHOOK_URL secret.
Run the flow manually with the defaults and confirm alert_governance_risk fires (AT_RISK: the table exists, no index covers event_id yet).
Run once with auto_remediate: "true" and dry_run: "true" and confirm log_dry_run_remediation describes the index without creating it.
Run once with auto_remediate: "true" and dry_run: "false" and confirm create_missing_index runs, then verify with SELECT * FROM duckdb_indexes() that the index now exists.
Point target_table at a table name that does not exist and re-run to confirm alert_table_unreachable fires (UNREACHABLE) instead of either HEALTHY or AT_RISK.
Enable periodic_governance_check once you trust the check; it always runs in the safe auto_remediate: "false" / dry_run: "true" mode regardless of what you leave the flow's own defaults set to.

How to extend

Loop over several target_table/required_indexed_column pairs with io.kestra.plugin.core.flow.Loop to run the same gate across every governed table in one database file.
Add VACUUM or CHECKPOINT as a separate, more cautiously gated remediation path for the storage-growth half of AT_RISK, since CREATE INDEX intentionally does not address it.
Filter check_storage_size's query by database_name instead of LIMIT 1 if this connection ever attaches more than one database.
Add has_primary_key from duckdb_tables() to the HEALTHY/AT_RISK decision for tables where a primary key, not just a secondary index, is the real governance requirement.

Pitfalls

This blueprint runs in-process, not against a server. DuckDB has no standalone server mode here - every Query task opens the same local file directly through the JDBC driver running inside the Kestra Worker. If the Worker runs in a container, database_path must resolve to a file visible inside that container, not just on the host.
pragma_database_size() LIMIT 1 assumes a single attached database. If more than one database is ever attached to this connection (via ATTACH), the row returned may not be the one you expect - filter by database_name explicitly in that case, per How to extend.
CREATE INDEX here has no IF NOT EXISTS guard. Re-running create_missing_index after the index already exists will error on the duplicate name; check_has_missing_index prevents this in the normal gated path by only firing when matching_indexes == 0, but a manually re-run task outside that guard is not protected.
The expressions::VARCHAR LIKE check is a substring match, not an exact column match. A column named event_id will also match an index on a differently named column that happens to contain that substring (for example source_event_id) - acceptable for a compliance gate erring toward false negatives on "no index", but worth tightening with an exact expression match if your schema has overlapping column names.
There is no structured "list indexes" or "database size" task in this plugin. Both catalog introspection and the CREATE INDEX DDL go through the same generic io.kestra.plugin.jdbc.duckdb.Query task used elsewhere in this repo's own DuckDB flows.
auto_remediate and dry_run are independent gates, not a single boolean. Both must be "true"/"false" respectively for CREATE INDEX to run; setting only one leaves the other still blocking.
Switch case keys "true"/"false" are quoted strings. auto_remediate renders as the literal string "true" or "false"; unquoted true:/false: YAML map keys would parse as booleans instead and would not match.
This blueprint is UNTESTED against a live DuckDB file. The duckdb_tables(), duckdb_indexes(), and pragma_database_size() column names are taken verbatim from DuckDB's own official documentation, and the Query task's properties (url, sql, fetchType) are taken verbatim from this repo's own existing DuckDB flows - but no Kestra engine was run to execute this flow end to end.

DuckDB Index and Storage Growth Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

DuckDB Index and Storage Growth Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Query data from Druid and load it to Redshift

Fetch data from Apache Druid and transform it in Python with Pandas

DocumentDB analytics report with Slack digest

Druid Segment Availability Gate

Query data from Dremio and process it in Polars with a Python script

Report the four DORA metrics per service every week and alert when delivery degrades

More Infrastructure Blueprints

Digest-Pinned Container Promotion with Approval

Container image size regression guard

DockerDash

Docs Drift Guard with AI-Proposed Fixes and Approval for Every Merged Pull Request

Docker Host Image Retention and Disk Reclaim

Manufacturing Downtime Root-Cause Triage with AI and MQTT

New to Kestra?