MariaDB System-Versioned History Purge Guard

id: mariadb-system-versioned-history-purge-guard
namespace: company.team

description: |
  Find MariaDB system-versioned (temporal) tables whose expired history has grown past a row
  threshold, alert Slack, then optionally purge only the expired history and re-check. A
  system-versioned table keeps a copy of every row that an UPDATE or DELETE replaces. Those
  history rows stay on disk and are scanned by every query that reads the table with
  FOR SYSTEM_TIME, so a table under steady write traffic keeps growing while its current row
  count is flat. MariaDB never trims this on its own. The guard reads information_schema to find
  the versioned tables, counts the history rows whose system-time period ended before a retention
  cutoff and reports the tables over the threshold. When asked it runs DELETE HISTORY ... BEFORE
  SYSTEM_TIME one table at a time, then counts the expired history again and fails the run if it
  did not fall. Read-only by default.

inputs:
  - id: jdbc_url
    type: STRING
    displayName: MariaDB JDBC URL
    description: Connects to the database the user has rights on. The schema to scan
      is set by target_schema.
    defaults: "jdbc:mariadb://mariadb:3306/mydb"

  - id: username
    type: STRING
    displayName: MariaDB user
    defaults: root

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

  - id: target_schema
    type: STRING
    displayName: Schema to scan
    description: Database (schema) whose system-versioned tables are checked. One
      schema per run.
    defaults: mydb

  - id: retention_days
    type: INT
    displayName: History retention days
    description: History older than this many days is treated as expired. The cutoff
      is now minus this value.
    defaults: 30

  - id: history_rows_threshold
    type: INT
    displayName: Expired history rows threshold
    description: A table is flagged when its expired history rows exceed this number.
    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 DELETE HISTORY to run.
    defaults: true

tasks:
  - id: compute_cutoff
    type: io.kestra.plugin.jdbc.mariadb.Query
    description: Resolve the retention cutoff once so the count, the purge and the
      re-check all use the same instant.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ inputs.username }}"
    password: "{{ inputs.password }}"
    fetchType: FETCH
    sql: "SELECT DATE_FORMAT(DATE_SUB(NOW(6), INTERVAL {{ inputs.retention_days }}
      DAY), '%Y-%m-%d %H:%i:%s.%f') AS cutoff"

  - id: list_versioned
    type: io.kestra.plugin.jdbc.mariadb.Query
    description: Read-only. Lists the system-versioned tables in the schema from
      information_schema and builds a per-table count of history rows that ended
      before the cutoff. A plain table is never listed, so a FOR SYSTEM_TIME
      count never runs against one.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ inputs.username }}"
    password: "{{ inputs.password }}"
    fetchType: FETCH
    parameters:
      cutoff: "{{ outputs.compute_cutoff.rows[0].cutoff }}"
      schema: "{{ inputs.target_schema }}"
    sql: |
      SELECT
        t.TABLE_NAME AS table_name,
        CONCAT(
          "SELECT '", t.TABLE_NAME, "' AS table_name, COUNT(*) AS expired_rows FROM `",
          t.TABLE_SCHEMA, "`.`", t.TABLE_NAME,
          "` FOR SYSTEM_TIME ALL WHERE `", COALESCE(p.END_COLUMN_NAME, 'row_end'),
          "` < '", :cutoff, "'"
        ) AS count_frag
      FROM information_schema.TABLES t
      LEFT JOIN information_schema.PERIODS p
        ON p.TABLE_SCHEMA = t.TABLE_SCHEMA
       AND p.TABLE_NAME = t.TABLE_NAME
       AND p.PERIOD = 'SYSTEM_TIME'
      WHERE t.TABLE_SCHEMA = :schema
        AND t.TABLE_TYPE = 'SYSTEM VERSIONED'
      ORDER BY t.TABLE_NAME

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      MariaDB system-versioned scan on {{ inputs.jdbc_url }} schema {{
      inputs.target_schema }}. Versioned tables found {{
      outputs.list_versioned.size }}. Retention cutoff {{
      outputs.compute_cutoff.rows[0].cutoff }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.list_versioned.size > 0 }}"
    then:
      - id: build_count
        type: io.kestra.plugin.core.output.OutputValues
        description: Join the per-table fragments into one UNION ALL that returns the
          expired history count per versioned table.
        values:
          count_sql: >-
            {{ outputs.list_versioned.rows | jq('map(.count_frag) | join(" UNION
            ALL ")') | first }}

      - id: count_expired
        type: io.kestra.plugin.jdbc.mariadb.Query
        description: Read-only. Counts expired history per table and keeps only the
          tables over the threshold.
        url: "{{ inputs.jdbc_url }}"
        username: "{{ inputs.username }}"
        password: "{{ inputs.password }}"
        fetchType: FETCH
        sql: >-
          SELECT q.table_name, q.expired_rows FROM ( {{
          outputs.build_count.values.count_sql }} ) q WHERE q.expired_rows > {{
          inputs.history_rows_threshold }} ORDER BY q.expired_rows DESC

      - id: report
        type: io.kestra.plugin.core.log.Log
        message: >-
          Expired history over the {{ inputs.history_rows_threshold }} row
          threshold before {{ outputs.compute_cutoff.rows[0].cutoff }}. Tables
          flagged {{ outputs.count_expired.size }}. {{
          outputs.count_expired.rows | jq('map("\(.table_name) \(.expired_rows)
          rows") | join("; ")') | first }}

      - id: route_flagged
        type: io.kestra.plugin.core.flow.If
        condition: "{{ outputs.count_expired.size > 0 }}"
        then:
          - id: alert
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: >-
              {{ {'text': 'MariaDB ' ~ inputs.jdbc_url ~ ' schema ' ~
              inputs.target_schema ~ ': ' ~ outputs.count_expired.size ~ '
              system-versioned table(s) with expired history over ' ~
              inputs.history_rows_threshold ~ ' rows before ' ~
              outputs.compute_cutoff.rows[0].cutoff ~ ' (' ~
              (outputs.count_expired.rows | jq('map("\(.table_name)
              (\(.expired_rows) rows)") | join(", ")') | first) ~ ').' ~
              (inputs.auto_remediate and not inputs.dry_run ? ' Purging expired
              history now.' : ' No change applied (auto_remediate=' ~
              inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} |
              toJson }}

          - id: plan
            type: io.kestra.plugin.core.output.OutputValues
            description: The flagged table names, one DELETE HISTORY per table.
            values:
              tables: >-
                {{ outputs.count_expired.rows | jq('map(.table_name)') | first |
                toJson }}

          - id: remediate_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
            then:
              - id: remediate
                type: io.kestra.plugin.core.flow.Loop
                description: Sequential remediation (concurrencyLimit 1). One DELETE HISTORY per
                  flagged table, since each one takes locks on its own table.
                concurrencyLimit: 1
                values: "{{ outputs.plan.values.tables }}"
                tasks:
                  - id: purge
                    type: io.kestra.plugin.jdbc.mariadb.Query
                    description: Remove only history rows whose system-time period ended before the
                      cutoff. Current rows and newer history are left in place.
                    url: "{{ inputs.jdbc_url }}"
                    username: "{{ inputs.username }}"
                    password: "{{ inputs.password }}"
                    fetchType: NONE
                    sql: >-
                      DELETE HISTORY FROM `{{ inputs.target_schema }}`.`{{
                      item.value }}` BEFORE SYSTEM_TIME '{{
                      outputs.compute_cutoff.rows[0].cutoff }}'

              - id: reverify
                type: io.kestra.plugin.jdbc.mariadb.Query
                description: Count expired history again with the same cutoff. The run fails if
                  any flagged table is still over the threshold.
                url: "{{ inputs.jdbc_url }}"
                username: "{{ inputs.username }}"
                password: "{{ inputs.password }}"
                fetchType: FETCH
                sql: >-
                  SELECT q.table_name, q.expired_rows FROM ( {{
                  outputs.build_count.values.count_sql }} ) q WHERE
                  q.expired_rows > {{ inputs.history_rows_threshold }} ORDER BY
                  q.expired_rows DESC

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.reverify.size > 0 }}"
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: "Expired history still over the threshold after DELETE HISTORY: {{
                      outputs.reverify.rows | jq('map(\"\\(.table_name)
                      \\(.expired_rows) rows\") | join(\", \")') | first }}.
                      Check that the user can DELETE on these tables."
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Purged expired history and verified. {{
                      fromJson(outputs.plan.values.tables) | length }} table(s)
                      back under the threshold before {{
                      outputs.compute_cutoff.rows[0].cutoff }}.
        else:
          - id: clean_under
            type: io.kestra.plugin.core.log.Log
            message: >-
              {{ outputs.list_versioned.size }} system-versioned table(s) in
              schema {{ inputs.target_schema }}, none with expired history over
              {{ inputs.history_rows_threshold }} rows before {{
              outputs.compute_cutoff.rows[0].cutoff }}. Nothing to purge.
    else:
      - id: clean_none
        type: io.kestra.plugin.core.log.Log
        message: >-
          No system-versioned (temporal) tables in schema {{
          inputs.target_schema }}. Nothing to check.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: >-
      {{ {'text': 'MariaDB system-versioned history 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. Always runs in safe mode (dry_run true,
      auto_remediate false).
    cron: "0 4 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true