DB2 Check-Pending Table Recovery Guard

id: db2-set-integrity-pending-check-guard
namespace: company.team

description: |
  Find DB2 tables stuck in check-pending state and bring them back online. After a LOAD or after
  a constraint is added, DB2 can leave a table with SYSCAT.TABLES.STATUS set to 'C'. In that state the
  table rejects every SELECT and every DML statement with SQL0668N, so an application that reads it
  is simply down. It stays down until somebody runs SET INTEGRITY. This flow reads
  SYSCAT.TABLES.STATUS for a schema, orders the pending tables parents first from SYSCAT.REFERENCES
  because a child cannot be checked while its parent is still pending, then behind a dry run and a
  Pause runs SET INTEGRITY FOR ... IMMEDIATE CHECKED one table at a time. It reads STATUS again and
  fails the run if any target is still not 'N', then counts rows through the tables it cleared to
  prove they answer queries again. Read-only by default.

inputs:
  - id: jdbc_url
    type: STRING
    displayName: JDBC URL
    defaults: "jdbc:db2://localhost:50000/testdb"

  - id: jdbc_username
    type: STRING
    displayName: Database user
    defaults: "db2inst1"

  - id: jdbc_password
    type: STRING
    displayName: Database password
    description: Defaults to the DB2_PASSWORD secret. Override per run only for
      local testing.
    defaults: "{{ secret('DB2_PASSWORD') }}"

  - id: table_schema
    type: STRING
    displayName: Schema to scan
    description: Schema whose tables are read from SYSCAT.TABLES. Folded to upper
      case before the query runs.
    defaults: "DB2INST1"

  - id: violation_handling
    type: SELECT
    displayName: What to do with rows that break the constraint
    description: >-
      fail lets SET INTEGRITY fail the statement on the first bad row so nothing
      is moved and the table stays pending. divert_to_exception_tables adds FOR
      EXCEPTION IN, which moves each bad row into a per table exception table
      and brings the table online. Rows are never discarded either way.
    values:
      - fail
      - divert_to_exception_tables
    defaults: fail

  - id: exception_table_suffix
    type: STRING
    displayName: Exception table suffix
    description: Appended to the table name to name its exception table in divert
      mode. ORDERS becomes ORDERS_EXCEPTIONS.
    defaults: "_EXCEPTIONS"

  - id: max_tables
    type: INT
    displayName: Tables to check per run
    description: >-
      Ceiling on how many tables one run brings back. SET INTEGRITY reads every
      row of the table to validate it, so a wide run can hold the schema for a
      long time. The lowest dependency depth is taken first, which means
      parents, so a truncated run leaves children for the next one.
    defaults: 10

  - 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, before SET INTEGRITY runs.
    defaults: true

variables:
  detect_sql: |
    WITH PEND AS (
      SELECT TABSCHEMA, TABNAME, STATUS, CONST_CHECKED
      FROM SYSCAT.TABLES
      WHERE TABSCHEMA = UPPER(CAST(:schema AS VARCHAR(128)))
        AND TYPE = 'T' AND STATUS <> 'N'
    ),
    EDGE AS (
      SELECT R.TABSCHEMA AS CS, R.TABNAME AS CT, R.REFTABSCHEMA AS PS, R.REFTABNAME AS PT
      FROM SYSCAT.REFERENCES R
      WHERE (R.TABSCHEMA, R.TABNAME) IN (SELECT TABSCHEMA, TABNAME FROM PEND)
        AND (R.REFTABSCHEMA, R.REFTABNAME) IN (SELECT TABSCHEMA, TABNAME FROM PEND)
        AND (R.TABSCHEMA <> R.REFTABSCHEMA OR R.TABNAME <> R.REFTABNAME)
    ),
    LVL (TABSCHEMA, TABNAME, DEPTH) AS (
      SELECT TABSCHEMA, TABNAME, CAST(0 AS INTEGER) FROM PEND
      UNION ALL
      SELECT E.CS, E.CT, L.DEPTH + 1
      FROM LVL L, EDGE E
      WHERE E.PS = L.TABSCHEMA AND E.PT = L.TABNAME AND L.DEPTH < 8
    )
    SELECT RTRIM(P.TABSCHEMA) AS TABSCHEMA, RTRIM(P.TABNAME) AS TABNAME, P.STATUS AS STATUS,
           RTRIM(P.CONST_CHECKED) AS CONST_CHECKED,
           (SELECT MAX(L.DEPTH) FROM LVL L
             WHERE L.TABSCHEMA = P.TABSCHEMA AND L.TABNAME = P.TABNAME) AS DEPTH,
           COALESCE((SELECT LISTAGG(RTRIM(E.PS) CONCAT '.' CONCAT RTRIM(E.PT), ', ')
                     FROM EDGE E WHERE E.CS = P.TABSCHEMA AND E.CT = P.TABNAME), '') AS PENDING_PARENTS,
           (SELECT COUNT(*) FROM SYSCAT.TABLES X
             WHERE X.TABSCHEMA = P.TABSCHEMA
               AND X.TABNAME = RTRIM(P.TABNAME) CONCAT UPPER(CAST(:sfx AS VARCHAR(128)))) AS EXC_EXISTS
    FROM PEND P
    ORDER BY 5, 1, 2

  verify_sql: |
    SELECT RTRIM(TABSCHEMA) AS TABSCHEMA, RTRIM(TABNAME) AS TABNAME, STATUS
    FROM SYSCAT.TABLES
    WHERE TABSCHEMA = UPPER(CAST(:schema AS VARCHAR(128)))
      AND TYPE = 'T' AND STATUS <> 'N'

tasks:
  - id: detect
    type: io.kestra.plugin.jdbc.db2.Query
    description: >-
      Read-only. Lists every table in the schema whose STATUS is not 'N', with
      its dependency depth from SYSCAT.REFERENCES, the pending parents that must
      be checked first and whether an exception table exists for it.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ inputs.jdbc_username }}"
    password: "{{ inputs.jdbc_password }}"
    fetchType: FETCH
    sql: "{{ render(vars.detect_sql) }}"
    parameters:
      schema: "{{ inputs.table_schema }}"
      sfx: "{{ inputs.exception_table_suffix }}"

  - id: assess
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      Split check-pending tables from inoperative ones, order them parents
      first, apply the per run ceiling, build one SET INTEGRITY statement per
      target and the row count query that proves the tables answer reads again.
    values:
      items: >-
        {{ {'pend': outputs.detect.rows, 'schema': (inputs.table_schema |
        upper), 'mode': inputs.violation_handling,
            'sfx': (inputs.exception_table_suffix | upper), 'max': inputs.max_tables,
            'ar': inputs.auto_remediate, 'dr': inputs.dry_run,
            'armed': (inputs.auto_remediate and not inputs.dry_run), 'url': inputs.jdbc_url} | jq('
          . as $i
          | [ ($i.pend // [])[] | {schema: .TABSCHEMA, table: .TABNAME, status: .STATUS,
              const_checked: (.CONST_CHECKED // ""), depth: ((.DEPTH // 0) | tonumber),
              parents: (.PENDING_PARENTS // ""), exc: (((.EXC_EXISTS // 0) | tonumber) > 0),
              name: (.TABSCHEMA + "." + .TABNAME),
              qname: ("\"" + .TABSCHEMA + "\".\"" + .TABNAME + "\""),
              exc_name: (.TABSCHEMA + "." + .TABNAME + $i.sfx),
              exc_qname: ("\"" + .TABSCHEMA + "\".\"" + .TABNAME + $i.sfx + "\"")} ] as $all
          | ($i.mode == "divert_to_exception_tables") as $divert
          | [ $all[] | select(.status == "C") ] as $cp
          | [ $all[] | select(.status != "C") ] as $other
          | ($cp | sort_by(.depth)) as $ord
          | ($ord | length) as $cpn
          | ($ord[0:$i.max]) as $sel
          | ($cpn > ($sel | length)) as $truncated
          | [ $sel[] | select($divert and (.exc | not)) | .exc_name ] as $missing
          | [ $sel[] | . + {stmt: ("SET INTEGRITY FOR " + .qname + " IMMEDIATE CHECKED"
              + (if $divert then " FOR EXCEPTION IN " + .qname + " USE " + .exc_qname else "" end))} ] as $tg
          | [ $tg[] | .stmt ] as $stmts
          | [ $tg[] | .name ] as $names
          | ($names + (if $divert then [ $tg[] | .exc_name ] else [] end)) as $labels
          | ([ $tg[] | .qname ] + (if $divert then [ $tg[] | .exc_qname ] else [] end)) as $qall
          | (($qall | to_entries | map("SELECT " + ((.key + 1) | tostring) + " AS SEQ, COUNT(*) AS ROW_COUNT FROM " + .value) | join(" UNION ALL "))
             + (if ($qall | length) > 0 then " ORDER BY 1" else "" end)) as $count_sql
          | (if ($all | length) == 0 then "OK" elif ($missing | length) > 0 then "BLOCKED" elif $cpn == 0 then "BLOCKED" else "ACTION" end) as $verdict
          | ($verdict == "ACTION" and $i.armed) as $will_change
          | ($verdict != "OK") as $attention
          | ("DB2 check-pending scan of " + $i.schema + " on " + $i.url + ": \($all | length) table(s) with STATUS not N, \($cpn) of them check pending."
             + (if $cpn > 0 then " Check pending: " + ([ $ord[] | "\(.name) (depth \(.depth)\(if .parents == "" then "" else ", after \(.parents)" end))" ] | join("; ")) + "." else "" end)
             + (if ($other | length) > 0 then " Not fixable by SET INTEGRITY: " + ([ $other[] | "\(.name) STATUS \(.status)" ] | join("; ")) + "." else "" end)
             + (if $truncated then " Only the first \($tg | length) are in scope this run (max_tables \($i.max)), lowest dependency depth first." else "" end)) as $core
          | (if ($missing | length) > 0 then "Blocked: divert mode needs an exception table per target and these are missing: " + ($missing | join(", ")) + ". Create one per target first. The other option is violation_handling fail."
             elif $verdict == "BLOCKED" then "Nothing SET INTEGRITY can fix here. An inoperative table has to be dropped and recreated."
             elif $will_change then "Running " + ($tg | length | tostring) + " SET INTEGRITY statement(s) now, parents first, one table at a time."
             elif $verdict == "ACTION" then "Fix available: " + ($stmts | join("; ")) + ". No change applied (auto_remediate=\($i.ar) dry_run=\($i.dr))."
             else "Every table in " + $i.schema + " reads STATUS N." end) as $action
          | {verdict: $verdict, pending_total: ($all | length), check_pending: $cpn,
             other_count: ($other | length), target_count: ($tg | length), truncated: $truncated,
             divert: $divert, missing_exception_tables: $missing, stmts: $stmts, names: $names,
             labels: $labels, count_sql: $count_sql, will_change: $will_change, attention: $attention,
             targets: [ $tg[] | {name: .name, depth: .depth, parents: .parents, status: .status, const_checked: .const_checked} ],
             report: ($core + " Verdict " + $verdict + ". " + $action),
             slack: ($core + " Verdict " + $verdict + ". " + $action)}
          ') | first | toJson }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: "{{ fromJson(outputs.assess.values.items).report }}"

  - id: route
    type: io.kestra.plugin.core.flow.If
    description: Alert and consider remediation when a table is not in STATUS N.
      Otherwise log and stop.
    condition: "{{ fromJson(outputs.assess.values.items) | jq('if .attention then 1
      else 0 end') | first > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': fromJson(outputs.assess.values.items).slack} | toJson }}

      - id: planned
        type: io.kestra.plugin.core.log.Log
        description: The exact statements a remediation run would execute, printed on
          every finding including a dry one.
        message: >-
          {{ {'s': fromJson(outputs.assess.values.items).stmts, 'w':
          fromJson(outputs.assess.values.items).will_change} | jq('
            if (.s | length) == 0 then "No SET INTEGRITY statement to run."
            else "Statements in order: " + (.s | join("; ")) + ". " + (if .w then "Running them now." else "Not running them this run. The verdict line above says why." end) end') | first }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        description: SET INTEGRITY runs only on verdict ACTION with auto_remediate true
          and dry_run false.
        condition: "{{ fromJson(outputs.assess.values.items) | jq('if .will_change then
          1 else 0 end') | first > 0 }}"
        then:
          - id: hold
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              SET INTEGRITY reads every row of the table to validate it, so on a
              large table it is a long scan that holds the table. A human
              approves the window before it starts.
            onResume:
              - id: approved
                type: BOOL
                displayName: Approve SET INTEGRITY on the tables listed above
                required: false

          - id: approval_gate
            type: io.kestra.plugin.core.flow.If
            description: An unanswered resume leaves approved null, which is read as a
              refusal.
            condition: "{{ outputs.hold.onResume.approved ?? false }}"
            then:
              - id: remediate
                type: io.kestra.plugin.core.flow.Loop
                description: >-
                  Sequential remediation (concurrencyLimit 1). One table at a
                  time, parents before children, because a child cannot be
                  checked while its parent is still pending.
                concurrencyLimit: 1
                values: "{{ fromJson(outputs.assess.values.items).stmts | toJson }}"
                tasks:
                  - id: log_statement
                    type: io.kestra.plugin.core.log.Log
                    message: "Running {{ item.value }}"

                  - id: set_integrity
                    type: io.kestra.plugin.jdbc.db2.Query
                    description: >-
                      Validates every row of the table and takes it out of
                      check-pending state. In fail mode a row that breaks the
                      constraint fails this statement and the table stays
                      pending, which the re-verify step turns into a failed run.
                    url: "{{ inputs.jdbc_url }}"
                    username: "{{ inputs.jdbc_username }}"
                    password: "{{ inputs.jdbc_password }}"
                    fetchType: NONE
                    sql: "{{ item.value }}"
                    allowFailure: true

              - id: reverify
                type: io.kestra.plugin.jdbc.db2.Query
                description: Read SYSCAT.TABLES.STATUS again for the whole schema. The run fails
                  if a target is still not 'N'.
                url: "{{ inputs.jdbc_url }}"
                username: "{{ inputs.jdbc_username }}"
                password: "{{ inputs.jdbc_password }}"
                fetchType: FETCH
                sql: "{{ render(vars.verify_sql) }}"
                parameters:
                  schema: "{{ inputs.table_schema }}"

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: >-
                  {{ {'rows': (outputs.reverify.rows ?? []), 'names':
                  fromJson(outputs.assess.values.items).names} | jq('
                    [ (.rows // [])[] | .TABSCHEMA + "." + .TABNAME ] as $left
                    | [ .names[] | select((([.] - $left) | length) == 0) ] | length') | first > 0 }}
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      {{ {'rows': (outputs.reverify.rows ?? []), 'names':
                      fromJson(outputs.assess.values.items).names} | jq('
                        [ (.rows // [])[] | {n: (.TABSCHEMA + "." + .TABNAME), s: .STATUS} ] as $left
                        | [ .names[] as $n | $left[] | select(.n == $n) | "\(.n) STATUS \(.s)" ] as $stuck
                        | "Still not in STATUS N after SET INTEGRITY: " + ($stuck | join("; "))
                          + ". A row that breaks the constraint fails the statement in fail mode, so either delete the offending rows or rerun with violation_handling divert_to_exception_tables. A table whose parent is still pending cannot be checked, so clear the parent first."') | first }}
                else:
                  - id: prove_reads
                    type: io.kestra.plugin.jdbc.db2.Query
                    description: >-
                      Count rows through every table that was cleared, which is
                      the proof it serves queries again rather than rejecting
                      them with SQL0668N. In divert mode the exception tables
                      are counted too, so diverted rows are reported and not
                      lost.
                    url: "{{ inputs.jdbc_url }}"
                    username: "{{ inputs.jdbc_username }}"
                    password: "{{ inputs.jdbc_password }}"
                    fetchType: FETCH
                    sql: "{{ fromJson(outputs.assess.values.items).count_sql }}"

                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      {{ {'rows': outputs.prove_reads.rows, 'labels':
                      fromJson(outputs.assess.values.items).labels, 'd':
                      fromJson(outputs.assess.values.items).divert} | jq('
                        . as $i
                        | [ ($i.rows // [])[] | {seq: ((.SEQ // 0) | tonumber), c: ((.ROW_COUNT // 0) | tonumber)} ] as $r
                        | [ $r[] | "\($i.labels[.seq - 1]) \(.c) row(s)" ] as $lines
                        | "Every target reads STATUS N again and answers queries: " + ($lines | join("; "))
                          + (if $i.d then ". Counts for the _EXCEPTIONS tables are the rows diverted out of the base tables." else "" end)') | first }}
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: "SET INTEGRITY was not approved at the pause, so no table was changed.
                  The schema is still reporting {{
                  fromJson(outputs.assess.values.items).check_pending }}
                  check-pending table(s)."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "{{ fromJson(outputs.assess.values.items).report }}"

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

triggers:
  - id: every_15_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: >-
      Shipped disabled. Always runs in safe mode. A check-pending table is an
      outage, so a short interval matters more here than it does for a
      performance check.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true