Amazon Redshift Table Unsorted Bloat and Stale Statistics Sentinel

id: redshift-table-bloat-and-stale-statistics-sentinel
namespace: company.dataops
description: |
  Audits tables in the target Amazon Redshift database for unsorted row bloat and stale query optimizer statistics.
  Queries SVV_TABLE_INFO to detect tables exceeding unsorted row or stale statistics thresholds,
  dispatching prioritized VACUUM SORT ONLY and ANALYZE operational remediation alerts to Slack.

inputs:
  - id: unsorted_pct_threshold
    type: INT
    displayName: Unsorted Row Threshold (%)
    description: Percentage of unsorted rows in a table triggering VACUUM alert
      recommendation (1-100%, applies to tables with defined sort keys).
    defaults: 20
    min: 1
    max: 100

  - id: stats_off_threshold
    type: INT
    displayName: Stale Statistics Threshold Score
    description: Query optimizer statistics staleness score (stats_off) triggering
      ANALYZE alert recommendation (10-100, matching Redshift default
      analyze_threshold_percent cutoff).
    defaults: 10
    min: 10
    max: 100

  - id: min_table_size_mb
    type: INT
    displayName: Minimum Table Size (MB)
    description: Minimum table size in megabytes to include in the operational
      hygiene audit.
    defaults: 100
    min: 1

  - id: limit_rows
    type: INT
    displayName: Maximum Tables Reported
    description: Maximum number of anomalous tables to highlight in the operational
      incident alert card (1-47, capped by Slack block limits).
    defaults: 10
    min: 1
    max: 47

triggers:
  - id: weekly_redshift_maintenance_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled weekly maintenance audit checking Amazon Redshift table
      unsorted bloat and stale statistics.
    cron: "0 2 * * 0"
    timezone: "UTC"
    disabled: true

tasks:
  - id: fetch_database_context
    type: io.kestra.plugin.jdbc.redshift.Query
    description: Identifies the target Amazon Redshift database name using
      current_database().
    url: "{{ secret('REDSHIFT_JDBC_URL') }}"
    username: "{{ secret('REDSHIFT_USER') }}"
    password: "{{ secret('REDSHIFT_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: SELECT current_database() AS "database_name";

  - id: audit_table_hygiene
    type: io.kestra.plugin.jdbc.redshift.Query
    description: Queries Amazon Redshift SVV_TABLE_INFO for the connected database.
      Requires superuser privileges or sys:operator role on REDSHIFT_USER.
    url: "{{ secret('REDSHIFT_JDBC_URL') }}"
    username: "{{ secret('REDSHIFT_USER') }}"
    password: "{{ secret('REDSHIFT_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        "database" AS "database_name",
        "schema" AS "schema_name",
        "table" AS "table_name",
        "size" AS "size_mb",
        "pct_used" AS "pct_cluster_used",
        ROUND("unsorted", 2) AS "unsorted_pct",
        ROUND("stats_off", 2) AS "stats_off_score",
        "tbl_rows" AS "row_count",
        "sortkey_num",
        CASE
          WHEN "sortkey_num" > 0 AND "unsorted" >= {{ inputs.unsorted_pct_threshold }} AND "stats_off" >= {{ inputs.stats_off_threshold }} THEN 'VACUUM AND ANALYZE REQUIRED'
          WHEN "sortkey_num" > 0 AND "unsorted" >= {{ inputs.unsorted_pct_threshold }} THEN 'VACUUM SORT REQUIRED'
          WHEN "stats_off" >= {{ inputs.stats_off_threshold }} THEN 'ANALYZE REQUIRED'
          ELSE 'HEALTHY'
        END AS "recommended_action",
        COUNT(*) OVER() AS "total_anomalies"
      FROM SVV_TABLE_INFO
      WHERE (("sortkey_num" > 0 AND "unsorted" >= {{ inputs.unsorted_pct_threshold }}) OR "stats_off" >= {{ inputs.stats_off_threshold }})
        AND "size" >= {{ inputs.min_table_size_mb }}
        AND "schema" NOT LIKE 'pg!_temp!_%' ESCAPE '!'
        AND "schema" NOT LIKE 'pg!_toast%' ESCAPE '!'
        AND "schema" NOT IN ('pg_catalog', 'information_schema', 'pg_internal')
      ORDER BY (CASE WHEN "sortkey_num" > 0 THEN "unsorted" ELSE 0 END) DESC, "stats_off" DESC
      LIMIT {{ inputs.limit_rows }};

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Returns structured audit telemetry summarizing inspected Redshift
      table health metrics prior to notification.
    format: |
      {{
        {
          "flowId": flow.id,
          "executionId": execution.id,
          "targetDatabase": outputs.fetch_database_context.row.database_name ?? outputs.fetch_database_context.row.DATABASE_NAME,
          "status": (outputs.audit_table_hygiene.rows is not empty and (outputs.audit_table_hygiene.rows | length) > 0) ? "ANOMALIES_DETECTED" : "HEALTHY",
          "unsortedThresholdPct": inputs.unsorted_pct_threshold,
          "statsOffThreshold": inputs.stats_off_threshold,
          "minTableSizeMb": inputs.min_table_size_mb,
          "totalAnomaliesDetected": (outputs.audit_table_hygiene.rows is not empty and (outputs.audit_table_hygiene.rows | length) > 0) ? (((outputs.audit_table_hygiene.rows[0].total_anomalies ?? outputs.audit_table_hygiene.rows[0].TOTAL_ANOMALIES) ?? (outputs.audit_table_hygiene.rows | length))) : 0,
          "flaggedTables": (outputs.audit_table_hygiene.rows is not empty ? (outputs.audit_table_hygiene.rows | jq('map({
            database: (.database_name // .DATABASE_NAME),
            schema: (.schema_name // .SCHEMA_NAME),
            table: (.table_name // .TABLE_NAME),
            sizeMb: (.size_mb // .SIZE_MB),
            pctClusterUsed: (.pct_cluster_used // .PCT_CLUSTER_USED),
            unsortedPct: (.unsorted_pct // .UNSORTED_PCT),
            statsOffScore: (.stats_off_score // .STATS_OFF_SCORE),
            rowCount: (.row_count // .ROW_COUNT),
            sortkeyNum: (.sortkey_num // .SORTKEY_NUM),
            action: (.recommended_action // .RECOMMENDED_ACTION)
          })') | first) : [])
        } | toJson
      }}

  - id: evaluate_table_health
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether any monitored tables in the connected database
      violated unsorted or statistics thresholds.
    condition: "{{ outputs.audit_table_hygiene.rows is not empty and
      (outputs.audit_table_hygiene.rows | length) > 0 }}"
    then:
      - id: notify_slack_incident
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Delivers an operational triage card to Slack with bloated and stale
          tables and VACUUM/ANALYZE commands.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{
            ({
              "db": (outputs.fetch_database_context.row.database_name ?? outputs.fetch_database_context.row.DATABASE_NAME),
              "total": (((outputs.audit_table_hygiene.rows[0].total_anomalies ?? outputs.audit_table_hygiene.rows[0].TOTAL_ANOMALIES) ?? (outputs.audit_table_hygiene.rows | length))),
              "unsorted_pct_threshold": inputs.unsorted_pct_threshold,
              "stats_off_threshold": inputs.stats_off_threshold,
              "min_table_size_mb": inputs.min_table_size_mb,
              "limit_rows": inputs.limit_rows,
              "execution_id": execution.id,
              "rows": outputs.audit_table_hygiene.rows
            } | jq('
              (.db as $db_fallback |
               .unsorted_pct_threshold as $u_thresh |
               .stats_off_threshold as $s_thresh |
               {
                 "channel": "#dataops-alerts",
                 "blocks": ([
                   {
                     "type": "header",
                     "text": {
                       "type": "plain_text",
                       "text": "🚨 Amazon Redshift Table Hygiene Alert"
                     }
                   },
                   {
                     "type": "section",
                     "text": {
                       "type": "plain_text",
                       "text": ("Found " + (.total | tostring) + " tables requiring VACUUM / ANALYZE maintenance in database: " + (.db | tostring) + "\n\nThresholds Evaluated:\n• Unsorted rows: >= " + (.unsorted_pct_threshold | tostring) + "% (requires sort keys)\n• Stale stats_off: >= " + (.stats_off_threshold | tostring) + "\n• Minimum table size: >= " + (.min_table_size_mb | tostring) + " MB\n\nTop Bloated & Stale Tables (Displaying up to " + (.limit_rows | tostring) + "):")
                     }
                   }
                 ] + (.rows[0:47] | map({
                   "type": "section",
                   "text": {
                     "type": "plain_text",
                     "text": (
                       (.database_name // .DATABASE_NAME // $db_fallback // "") as $raw_d |
                       (.schema_name // .SCHEMA_NAME // "") as $raw_s |
                       (.table_name // .TABLE_NAME // "") as $raw_t |
                       ("\"" + ($raw_s | gsub("\""; "\"\"")) + "\".\"" + ($raw_t | gsub("\""; "\"\"")) + "\"") as $ident |
                       ((.sortkey_num // .SORTKEY_NUM // 0) | tonumber) as $sortkey_num |
                       ((.unsorted_pct // .UNSORTED_PCT // 0) | tonumber) as $unsorted |
                       ((.stats_off_score // .STATS_OFF_SCORE // 0) | tonumber) as $statsoff |
                       ((.size_mb // .SIZE_MB // 0) | tostring) as $size |
                       ((.row_count // .ROW_COUNT // 0) | tostring) as $rows |
                       ((.pct_cluster_used // .PCT_CLUSTER_USED // 0) | tostring) as $pctused |
                       ((.recommended_action // .RECOMMENDED_ACTION // "") | tostring) as $action |
                       (if $sortkey_num > 0 and $unsorted >= $u_thresh then "VACUUM SORT ONLY " + $ident + " TO 100 PERCENT; " else "" end) as $vac |
                       (if $statsoff >= $s_thresh then "ANALYZE " + $ident + " THRESHOLD 0 PERCENT;" else "" end) as $anz |
                       "• Database: " + $raw_d + " | Table: " + $ident + " (" + $size + " MB, " + $rows + " rows)\n  - Unsorted: " + ($unsorted | tostring) + "% | Stats Off: " + ($statsoff | tostring) + " | Cluster Space: " + $pctused + "%\n  - Recommended Action: " + $action + "\n  - Remediation: " + $vac + $anz
                     )
                   }
                 })) + [
                   {
                     "type": "context",
                     "elements": [
                       {
                         "type": "plain_text",
                         "text": ("Database: " + (.db | tostring) + " | Execution ID: " + (.execution_id | tostring))
                       }
                     ]
                   }
                 ])
               }
              )
            ') | first) | toJson
          }}
    else:
      - id: log_healthy_database
        type: io.kestra.plugin.core.log.Log
        description: Logs confirmation that all monitored tables in the connected
          database satisfy sort and query statistics health standards.
        message: "Amazon Redshift table hygiene audit clean: all monitored tables >= {{
          inputs.min_table_size_mb }} MB in database '{{
          outputs.fetch_database_context.row.database_name ??
          outputs.fetch_database_context.row.DATABASE_NAME }}' are below
          unsorted threshold ({{ inputs.unsorted_pct_threshold }}%) and
          stats_off threshold ({{ inputs.stats_off_threshold }})."

errors:
  - id: notify_workflow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts operational channels if the sentinel workflow encounters an
      unhandled execution error.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{
        {
          "channel": "#dataops-alerts",
          "text": "❌ *Amazon Redshift Sentinel Failure*: Table hygiene sentinel execution `" ~ execution.id ~ "` encountered an error. Please inspect execution task logs for failure details."
        } | toJson
      }}