Apache Iceberg Table Snapshot Sprawl and Small File Compaction Sentinel

id: apache-iceberg-snapshot-expiration-compaction-sentinel
namespace: company.datalake

inputs:
  - id: trino_host
    type: STRING
    defaults: "trino.internal"
    description: Hostname of Trino / Starburst cluster connecting to Iceberg.
  - id: trino_port
    type: INT
    defaults: 8080
    description: Trino HTTP port.
  - id: trino_user
    type: STRING
    defaults: "kestra"
    description: Trino user.
  - id: catalog
    type: STRING
    defaults: "iceberg"
    description: Iceberg catalog name in Trino.
  - id: schema_name
    type: STRING
    description: Database / schema name.
  - id: table_name
    type: STRING
    description: Iceberg table name to audit.

tasks:
  - id: query_snapshots
    type: io.kestra.plugin.jdbc.trino.Query
    url: "jdbc:trino://{{ inputs.trino_host }}:{{ inputs.trino_port }}/{{
      inputs.catalog }}/{{ inputs.schema_name }}"
    username: "{{ inputs.trino_user }}"
    sql: |
      SELECT 
        COUNT(*) AS total_snapshots
      FROM "{{ inputs.catalog }}"."{{ inputs.schema_name }}"."{{ inputs.table_name }}$snapshots";
    fetchOne: true

  - id: evaluate_snapshot_sprawl
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.query_snapshots.row.total_snapshots | int > 100 }}"
    then:
      - id: notify_datalake_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          {
            "channel": "#datalake-alerts",
            "attachments": [
              {
                "color": "warning",
                "title": "❄️ Apache Iceberg Snapshot Sprawl Alert",
                "text": "Iceberg Table `{{ inputs.schema_name }}.{{ inputs.table_name }}` has accumulated *{{ outputs.query_snapshots.row.total_snapshots }} snapshots*! Neglected snapshot expiration increases metadata manifest read times, drastically slowing down query planning for Spark, Trino, and Flink engines.",
                "footer": "Kestra Data Lake Sentinel • Apache Iceberg"
              }
            ]
          }

errors:
  - id: alert_pipeline_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "#datalake-alerts",
        "attachments": [
          {
            "color": "danger",
            "title": "⚠️ Apache Iceberg Sentinel Failure",
            "text": "Failed to query Iceberg table snapshots via Trino. Verify Trino connectivity and table existence.",
            "footer": "Kestra Workflow Failure Sentinel"
          }
        ]
      }

triggers:
  - id: periodic_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 8 * * 1"
    disabled: true