Vertica Delete Vector Purge Guard

id: vertica-delete-vector-purge-guard
namespace: company.team

description: |
  Find Vertica projections whose delete vectors have grown past a row threshold, alert Slack,
  then optionally reclaim them. Every UPDATE and DELETE in Vertica writes a delete vector that
  stays on disk and is read on every scan of the projection until it is purged, so a table that
  takes a steady stream of deletes keeps getting slower while its visible row count is flat. The
  guard reads v_monitor.delete_vectors, reports the projections over the threshold with the rows
  each one can reclaim. When asked it advances the Ancient History Mark with MAKE_AHM_NOW, runs
  PURGE_TABLE on each flagged table, then reads v_monitor.delete_vectors again and fails if any
  vectors remain.

inputs:
  - id: jdbc_url
    type: STRING
    displayName: Vertica JDBC URL
    defaults: "jdbc:vertica://vertica:5433/VMart"

  - id: username
    type: STRING
    displayName: Vertica user
    defaults: dbadmin

  - id: password
    type: STRING
    displayName: Vertica password
    description: Defaults to the VERTICA_PASSWORD secret. Override per run only for
      a passwordless test database.
    defaults: "{{ secret('VERTICA_PASSWORD') }}"

  - id: reclaimable_rows_threshold
    type: INT
    displayName: Reclaimable rows threshold
    description: A projection is flagged when its delete vectors cover more than
      this many rows.
    defaults: 100000

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for MAKE_AHM_NOW and
      PURGE_TABLE to run.
    defaults: true

variables:
  audit_sql: |
    SELECT dv.schema_name AS projection_schema,
           dv.projection_name AS projection_name,
           t.table_schema AS anchor_schema,
           t.table_name AS anchor_table,
           SUM(dv.deleted_row_count) AS deleted_rows,
           SUM(dv.used_bytes) AS dv_bytes,
           'SELECT PURGE_TABLE(' || QUOTE_LITERAL(QUOTE_IDENT(t.table_schema) || '.' || QUOTE_IDENT(t.table_name)) || ')' AS purge_sql
      FROM v_monitor.delete_vectors dv
      JOIN v_catalog.projections p
        ON p.projection_schema = dv.schema_name AND p.projection_name = dv.projection_name
      JOIN v_catalog.tables t
        ON t.table_id = p.anchor_table_id
     GROUP BY 1, 2, 3, 4
    HAVING SUM(dv.deleted_row_count) > {{ inputs.reclaimable_rows_threshold }}
     ORDER BY deleted_rows DESC

tasks:
  - id: audit
    type: io.kestra.plugin.jdbc.vertica.Query
    description: Read-only. Sums delete-vector rows per projection and keeps the
      ones over the threshold, with the PURGE_TABLE statement for each anchor
      table.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ inputs.username }}"
    password: "{{ inputs.password }}"
    fetchType: FETCH
    sql: "{{ render(vars.audit_sql) }}"

  - id: purges
    type: io.kestra.plugin.core.output.OutputValues
    description: One PURGE_TABLE per flagged anchor table, de-duplicated, since one
      table can carry several projections.
    values:
      sql: "{{ outputs.audit.rows | jq('map(.purge_sql) | unique') | first | toJson
        }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Vertica delete-vector audit on {{ inputs.jdbc_url }}: {{
      outputs.audit.size }} projection(s) over the {{
      inputs.reclaimable_rows_threshold }} row threshold. {{ outputs.audit.rows
      | jq('map("\(.anchor_schema).\(.anchor_table) [\(.projection_name)]
      \(.deleted_rows) rows") | join("; ")') | first }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit.size > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Vertica ' ~ inputs.jdbc_url ~ ': ' ~ outputs.audit.size ~ ' projection(s) carrying delete vectors above ' ~ inputs.reclaimable_rows_threshold ~ ' rows: ' ~ (outputs.audit.rows | jq('map("\(.anchor_schema).\(.anchor_table) (\(.deleted_rows) rows)") | join(", ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Purging now.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: advance_ahm
            type: io.kestra.plugin.jdbc.vertica.Query
            description: Advance the Ancient History Mark to now. PURGE only reclaims delete
              vectors older than the AHM, so without this the purge frees
              nothing.
            url: "{{ inputs.jdbc_url }}"
            username: "{{ inputs.username }}"
            password: "{{ inputs.password }}"
            fetchType: FETCH
            sql: SELECT MAKE_AHM_NOW()

          - id: remediate
            type: io.kestra.plugin.core.flow.Loop
            description: Sequential on purpose. PURGE_TABLE takes locks, so the flagged
              tables are remediated one at a time.
            concurrencyLimit: 1
            values: "{{ outputs.purges.values.sql }}"
            tasks:
              - id: purge
                type: io.kestra.plugin.jdbc.vertica.Query
                url: "{{ inputs.jdbc_url }}"
                username: "{{ inputs.username }}"
                password: "{{ inputs.password }}"
                fetchType: FETCH
                sql: "{{ item.value }}"

          - id: reaudit
            type: io.kestra.plugin.jdbc.vertica.Query
            url: "{{ inputs.jdbc_url }}"
            username: "{{ inputs.username }}"
            password: "{{ inputs.password }}"
            fetchType: FETCH
            sql: "{{ render(vars.audit_sql) }}"

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.reaudit.size > 0 }}"
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Delete vectors still over the threshold after purge: {{
                  outputs.reaudit.rows |
                  jq('map(\"\\(.anchor_schema).\\(.anchor_table)
                  \\(.deleted_rows) rows\") | join(\", \")') | first }}. The
                  connecting user needs to own the flagged tables or be a
                  superuser."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: "Purged and verified. {{ fromJson(outputs.purges.values.sql) | length
                  }} table(s) back under the threshold."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No projection carries delete vectors over the threshold."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Vertica delete-vector purge guard failed on ' ~ inputs.jdbc_url ~ ' (execution ' ~ execution.id ~ '). Check the logs.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode.
    cron: "0 4 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true