Job and pipeline health

WITH job_name AS (              -- system.lakeflow.jobs is SCD2: one row per EDIT
  SELECT workspace_id, job_id, name, delete_time
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
),
runs AS (                       -- job_run_timeline emits SEVERAL rows per run
  SELECT
    workspace_id,
    job_id,
    run_id,
    MIN(period_start_time)                                              AS started_at,
    MAX(period_end_time)                                                AS ended_at,
    MAX_BY(result_state, period_end_time)                               AS result_state,
    timestampdiff(SECOND, MIN(period_start_time), MAX(period_end_time)) AS wall_secs,
    MAX_BY(execution_duration_seconds, period_end_time)                 AS exec_secs
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= current_timestamp() - INTERVAL 14 DAYS
  GROUP BY workspace_id, job_id, run_id
)
SELECT
  n.workspace_id,                  -- job_id is unique PER WORKSPACE, and the
  n.job_id,                        -- lakeflow schema is account-wide
  n.name,
  COUNT(r.run_id)                                              AS runs,
  SUM(IF(r.result_state <> 'SUCCEEDED', 1, 0))                 AS not_clean,
  ROUND(100 * AVG(IF(r.result_state <> 'SUCCEEDED', 1, 0)), 1) AS pct_not_clean,
  ROUND(AVG(r.wall_secs))                                      AS avg_wall_seconds,
  ROUND(AVG(r.exec_secs))                                      AS avg_exec_seconds,
  MAX(r.ended_at)                                              AS last_finished
FROM job_name n                    -- jobs on the LEFT, deliberately: a job that
LEFT JOIN runs r                   -- stopped running must show as runs = 0,
  ON  r.workspace_id = n.workspace_id   -- not vanish from the result
  AND r.job_id       = n.job_id
  AND r.result_state IS NOT NULL   -- terminal runs only; in-flight rows carry NULL.
                                   -- Kept in the JOIN: as a WHERE clause it turns
                                   -- the LEFT JOIN back into an inner one.
WHERE n.delete_time IS NULL        -- deleted jobs keep their run history; skip them
GROUP BY n.workspace_id, n.job_id, n.name   -- NEVER by name alone: job names are not
                                            -- unique, in one workspace or across them
ORDER BY pct_not_clean DESC NULLS LAST, avg_wall_seconds DESC;

SELECT
  workspace_id,
  pipeline_id,
  update_id,
  MIN(period_start_time)                          AS started_at,
  MAX(period_end_time)                            AS ended_at,
  MAX_BY(result_state, period_end_time)           AS result_state,
  MAX_BY(full_refresh_selection, period_end_time) AS full_refresh_selection
FROM system.lakeflow.pipeline_update_timeline
WHERE period_start_time >= current_timestamp() - INTERVAL 14 DAYS
GROUP BY workspace_id, pipeline_id, update_id   -- periods, exactly like job runs
ORDER BY started_at DESC;

-- One update emits an update_progress event PER STATE TRANSITION:
--   QUEUED -> INITIALIZING -> SETTING_UP_TABLES -> RUNNING -> COMPLETED|FAILED|CANCELED
-- so a bare "ORDER BY timestamp DESC LIMIT 20" returns two or three updates, most
-- of the rows non-terminal. Filter to terminal states AND dedupe on update_id.
SELECT
  origin.update_id                                 AS update_id,
  MAX(timestamp)                                   AS ended_at,
  MAX_BY(details:update_progress.state, timestamp) AS state
FROM event_log(TABLE(dwh_prod.silver.sales_order))
WHERE event_type = 'update_progress'
  AND details:update_progress.state IN ('COMPLETED', 'FAILED', 'CANCELED')
GROUP BY origin.update_id
ORDER BY ended_at DESC
LIMIT 20;

-- Once the event log is published to UC, the FROM clause is the only line that
-- changes: swap the TVF for the catalog.schema.name configured on the pipeline —
-- and then the rota can read it, which through the TVF they cannot.

WITH runs AS (
  SELECT
    workspace_id,                 -- job_id is unique PER WORKSPACE; system.lakeflow
    job_id,                       -- is account-wide, so dev and prod would MERGE
    run_id,
    MAX(period_end_time) AS ended_at,
    -- what the deadline cares about: queue + setup + work + cleanup
    timestampdiff(SECOND, MIN(period_start_time), MAX(period_end_time)) AS wall_secs,
    -- what the code cares about: the work alone, from the table's own column
    MAX_BY(execution_duration_seconds, period_end_time)                 AS exec_secs,
    -- the difference, itemised: this is the contention signal
    COALESCE(MAX_BY(queue_duration_seconds, period_end_time), 0)
      + COALESCE(MAX_BY(setup_duration_seconds, period_end_time), 0)    AS wait_secs
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= current_timestamp() - INTERVAL 28 DAYS
  GROUP BY workspace_id, job_id, run_id
  -- healthy runs only: a fast failure would drag the trend DOWN
  HAVING MAX_BY(result_state, period_end_time) = 'SUCCEEDED'
),
drift AS (
  SELECT
    workspace_id,
    job_id,
    median(CASE WHEN ended_at >= current_timestamp() - INTERVAL 7 DAYS
                THEN wall_secs END) AS wall_7d,
    median(wall_secs)                AS wall_28d,
    median(CASE WHEN ended_at >= current_timestamp() - INTERVAL 7 DAYS
                THEN exec_secs END) AS exec_7d,
    median(exec_secs)                AS exec_28d,
    median(CASE WHEN ended_at >= current_timestamp() - INTERVAL 7 DAYS
                THEN wait_secs END) AS wait_7d,
    median(wait_secs)                AS wait_28d
  FROM runs
  GROUP BY workspace_id, job_id
)
SELECT
  workspace_id,
  job_id,
  ROUND(wall_7d)                            AS wall_7d,
  ROUND(wall_28d)                           AS wall_28d,
  ROUND(wall_7d / NULLIF(wall_28d, 0), 2)   AS wall_ratio,  -- the deadline question
  ROUND(exec_7d / NULLIF(exec_28d, 0), 2)   AS exec_ratio,  -- is the WORK heavier?
  ROUND(wait_7d - wait_28d)                 AS wait_delta   -- queue + setup, seconds
FROM drift
WHERE wall_7d > 1.25 * wall_28d
ORDER BY wall_ratio DESC;

      health:
        rules:
          - metric: RUN_DURATION_SECONDS
            op: GREATER_THAN
            value: 5400        # 90 min. Chain starts 03:00, gold is due 06:00.