ClickHouse Expired Partition Drop Guard

id: clickhouse-expired-partition-drop-guard
namespace: company.team

inputs:
  - id: clickhouse_url
    type: STRING
    defaults: jdbc:clickhouse://localhost:8123/default
    description: ClickHouse JDBC URL for the server that holds the table. Example
      jdbc:clickhouse://clickhouse:8123/default.

  - id: use_secret_auth
    type: BOOL
    defaults: true
    description: Leave true in production so the connection reads
      CLICKHOUSE_USERNAME and CLICKHOUSE_PASSWORD from secrets. Set false only
      to pass credentials inline in a disposable test environment.

  - id: clickhouse_username
    type: STRING
    defaults: default
    description: Username used only when use_secret_auth is false. Ignored in production.

  - id: clickhouse_password
    type: STRING
    defaults: ""
    description: Password used only when use_secret_auth is false. Ignored in production.

  - id: database
    type: STRING
    defaults: default
    description: Database that holds the MergeTree table to guard.

  - id: table
    type: STRING
    defaults: events
    description: MergeTree table partitioned by a date or datetime expression, for
      example toYYYYMM(ts), toDate(ts) or toYYYYMMDD(ts).

  - id: retention_days
    type: INT
    defaults: 90
    description: Retention window in days. A partition is expired when the newest
      row in it is older than now minus this many days.

  - id: dry_run
    type: BOOL
    defaults: true
    description: When true, report the expired partitions without dropping anything.

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: When true and dry_run is false, drop the expired partitions behind
      an approval gate, then re-read system.parts.

variables:
  detect_sql: |
    SELECT
      partition AS partition,
      partition_id AS partition_id,
      toString(greatest(max(max_time), toDateTime(max(max_date)))) AS newest_ts,
      sum(rows) AS rows_total,
      sum(bytes_on_disk) AS bytes_total,
      count() AS parts
    FROM system.parts
    WHERE database = '{{ inputs.database }}'
      AND table = '{{ inputs.table }}'
      AND active = 1
    GROUP BY partition, partition_id
    HAVING greatest(max(max_time), toDateTime(max(max_date))) < now() - INTERVAL {{ inputs.retention_days }} DAY
       AND greatest(max(max_time), toDateTime(max(max_date))) > toDateTime('1970-01-02 00:00:00')
    ORDER BY partition

triggers:
  - id: weekly_retention_check
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "0 2 * * 1"
    description: Weekly retention sweep. Disabled by default and pinned to safe mode
      so it reports without dropping until an operator enables it.
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: detect
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: Read system.parts for the table, group by partition and keep
      partitions whose newest row is older than the retention window. The epoch
      guard skips partitions with no date, so a non temporal partition key is
      never selected.
    url: "{{ inputs.clickhouse_url }}"
    username: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_USERNAME') :
      inputs.clickhouse_username }}"
    password: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_PASSWORD') :
      inputs.clickhouse_password }}"
    fetchType: FETCH
    sql: "{{ render(vars.detect_sql) }}"

  - id: summarize
    type: io.kestra.plugin.core.output.OutputValues
    description: Materialize the expired partition id list once so the logs, the
      Slack alert and the failure message all read the same value.
    values:
      partition_list: >-
        {{ outputs.detect.rows | jq('map(.partition_id) | join(", ")') | first }}

  - id: log_detect
    type: io.kestra.plugin.core.log.Log
    message: |
      ClickHouse {{ inputs.database }}.{{ inputs.table }}: {{ outputs.detect.size }} expired partition(s) older than {{ inputs.retention_days }} day(s).
      Partitions: {{ outputs.summarize.values.partition_list }}
      Mode: dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}

  - id: guard
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.detect.size > 0 }}"
    then:
      - id: findings_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alert on every detection, in dry run and in remediate mode, so a
          human sees the retention breach even when nothing is dropped.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"text": "ClickHouse expired partition guard: {{ outputs.detect.size }} partition(s) in {{ inputs.database }}.{{ inputs.table }} older than {{ inputs.retention_days }} day(s). Partitions {{ outputs.summarize.values.partition_list }}. dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}."}

      - id: mode
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approval_gate
            type: io.kestra.plugin.core.flow.Pause
            description: Hold the run until a human approves the drop. The destructive step
              never starts on its own.
            pauseDuration: PT2H
            behavior: FAIL
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Set true to drop the listed partitions. Anything else leaves them
                  in place.

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval_gate.onResume.approved ?? false }}"
            then:
              - id: drop_expired_partitions
                type: io.kestra.plugin.core.flow.Loop
                description: Remediation loop. One ALTER TABLE DROP PARTITION per expired
                  partition, concurrencyLimit 1 so the DDL runs in turn and
                  parallel drops never race on the same table.
                concurrencyLimit: 1
                values: "{{ outputs.detect.rows | jq('map(.partition_id)') | first | toJson }}"
                tasks:
                  - id: drop_one
                    type: io.kestra.plugin.jdbc.clickhouse.Query
                    description: Drop one expired partition by its id. allowFailure so one blocked
                      drop does not stop the pass and the re-read stays the real
                      gate.
                    allowFailure: true
                    url: "{{ inputs.clickhouse_url }}"
                    username: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_USERNAME') :
                      inputs.clickhouse_username }}"
                    password: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_PASSWORD') :
                      inputs.clickhouse_password }}"
                    fetchType: NONE
                    sql: "ALTER TABLE {{ inputs.database }}.{{ inputs.table }} DROP PARTITION ID '{{
                      item.value }}'"

                  - id: drop_log
                    type: io.kestra.plugin.core.log.Log
                    message: "Issued DROP PARTITION ID '{{ item.value }}' on {{ inputs.database
                      }}.{{ inputs.table }}."

              - id: reverify
                type: io.kestra.plugin.jdbc.clickhouse.Query
                description: Re-read system.parts with the same window. Any row here means a
                  drop did not land.
                url: "{{ inputs.clickhouse_url }}"
                username: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_USERNAME') :
                  inputs.clickhouse_username }}"
                password: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_PASSWORD') :
                  inputs.clickhouse_password }}"
                fetchType: FETCH
                sql: "{{ render(vars.detect_sql) }}"

              - 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: >-
                      {{ outputs.reverify.size }} expired partition(s) still
                      present after the drop pass in {{ inputs.database }}.{{
                      inputs.table }}: {{ outputs.reverify.rows |
                      jq('map(.partition_id) | join(", ")') | first }}. The
                      connecting user may lack ALTER privilege or a concurrent
                      writer re-created the partition.
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: "Dropped and verified. {{ outputs.detect.size }} expired partition(s)
                      before, {{ outputs.reverify.size }} after."
            else:
              - id: not_approved
                type: io.kestra.plugin.core.log.Log
                message: "Approval not granted, so nothing was dropped. {{ outputs.detect.size
                  }} expired partition(s) still remain in {{ inputs.database
                  }}.{{ inputs.table }}."
        else:
          - id: report_only
            type: io.kestra.plugin.core.log.Log
            message: "Report only (dry_run={{ inputs.dry_run }}, auto_remediate={{
              inputs.auto_remediate }}). {{ outputs.detect.size }} expired
              partition(s) would be dropped: {{
              outputs.summarize.values.partition_list }}. No change made."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No expired partitions in {{ inputs.database }}.{{ inputs.table }}
          older than {{ inputs.retention_days }} day(s). Nothing to drop."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post to Slack on any failure, including a connection error or
      expired partitions still present after a drop pass.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "ClickHouse expired partition guard FAILED for {{ inputs.database }}.{{ inputs.table }} (execution {{ execution.id }}). A connection error, an ALTER error or expired partitions still present after a drop pass. Check the run."}