Jobs & Pipelines

WITH end_rows AS (
  SELECT workspace_id, job_id, run_id, result_state, termination_code, period_start_time
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= dateadd(day, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())   -- drop incomplete current day
    AND result_state IS NOT NULL                                   -- end row only
),
job_runs AS (
  SELECT workspace_id, job_id,
         COUNT(DISTINCT run_id) AS distinct_runs,
         COUNT(DISTINCT CASE WHEN result_state IN ('FAILED','ERROR','TIMED_OUT')
                             THEN run_id END) AS failed_runs
  FROM end_rows
  GROUP BY workspace_id, job_id
),
-- ARGMAX, not MAX: the termination_code of the LATEST failed run, per (workspace_id, job_id).
-- MAX(termination_code) would pick the lexicographically-largest code, not the most recent.
last_fail AS (
  SELECT workspace_id, job_id, termination_code AS last_failed_termination_code
  FROM end_rows
  WHERE result_state IN ('FAILED','ERROR','TIMED_OUT')
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY workspace_id, job_id ORDER BY period_start_time DESC
  ) = 1
),
-- List price effective at each usage row's time, keyed by sku_name+cloud+usage_unit.
price AS (
  SELECT sku_name, cloud, usage_unit, price_start_time, price_end_time,
         CAST(pricing.default AS DOUBLE) AS list_rate
  FROM system.billing.list_prices
),
-- DBUs billed to each job over the window. usage_metadata.job_id is unique only within a
-- workspace, so we attribute on (workspace_id, job_id). Net across ALL record_types; DBU only.
-- Pre-aggregated by exactly (workspace_id, job_id) so the LEFT JOIN below stays strictly 1:1.
job_dbus AS (
  SELECT u.workspace_id,
         u.usage_metadata.job_id AS job_id,
         SUM(u.usage_quantity)                            AS net_usage_quantity,
         SUM(u.usage_quantity * COALESCE(p.list_rate, 0)) AS est_usd_list
  FROM system.billing.usage u
  LEFT JOIN price p
    ON u.sku_name = p.sku_name AND u.cloud = p.cloud AND u.usage_unit = p.usage_unit
   AND u.usage_end_time >= p.price_start_time
   AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
  WHERE u.usage_date >= dateadd(day, -:period_days, current_date())
    AND u.usage_date < current_date()
    AND upper(u.usage_unit) = 'DBU'
    AND u.usage_metadata.job_id IS NOT NULL
  GROUP BY u.workspace_id, u.usage_metadata.job_id
)
SELECT r.workspace_id,
       r.job_id,
       r.distinct_runs,
       r.failed_runs,
       lf.last_failed_termination_code,
       COALESCE(d.net_usage_quantity, 0)                                 AS net_job_dbus,
       -- Wasted-DBU proxy: job DBUs scaled by the share of runs that failed. A proxy,
       -- not exact run-level waste (usage_metadata has no run_id) - see caveats.
       CASE WHEN r.distinct_runs > 0
            THEN COALESCE(d.net_usage_quantity, 0) * (r.failed_runs / r.distinct_runs)
            ELSE 0 END                                                   AS wasted_dbus_proxy,
       COALESCE(d.net_usage_quantity, 0)                                 AS net_dbus,
       COALESCE(d.est_usd_list, 0)                                       AS est_usd_list,
       -- status: worst-first band on failed-run count and wasted-DBU magnitude (field heuristic;
       -- :warn_failed_runs/:crit_failed_runs and :warn_wasted_dbus/:crit_wasted_dbus).
       CASE
         WHEN r.failed_runs IS NULL THEN 'NOT_ASSESSED'
         WHEN r.failed_runs >= :crit_failed_runs
           OR (CASE WHEN r.distinct_runs > 0
                    THEN COALESCE(d.net_usage_quantity, 0) * (r.failed_runs / r.distinct_runs)
                    ELSE 0 END) >= :crit_wasted_dbus
         THEN 'CRITICAL'
         WHEN r.failed_runs >= :warn_failed_runs
           OR (CASE WHEN r.distinct_runs > 0
                    THEN COALESCE(d.net_usage_quantity, 0) * (r.failed_runs / r.distinct_runs)
                    ELSE 0 END) >= :warn_wasted_dbus
         THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM job_runs r
LEFT JOIN last_fail lf
  ON r.workspace_id = lf.workspace_id AND r.job_id = lf.job_id
LEFT JOIN job_dbus d
  ON r.workspace_id = d.workspace_id AND r.job_id = d.job_id
WHERE r.failed_runs > 0
ORDER BY wasted_dbus_proxy DESC, failed_runs DESC

WITH task_compute AS (
  SELECT workspace_id, job_id, run_id, task_key, EXPLODE(compute_ids) AS compute_id
  FROM system.lakeflow.job_task_run_timeline
  WHERE period_start_time >= dateadd(DAY, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())
    AND result_state IS NOT NULL
    AND compute_ids IS NOT NULL
),
-- Runs dropped because compute_ids was NULL/empty -> reported as NOT_ASSESSED, never as zero.
dropped AS (
  SELECT workspace_id, COUNT(DISTINCT run_id) AS dropped_runs
  FROM system.lakeflow.job_task_run_timeline
  WHERE period_start_time >= dateadd(DAY, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())
    AND result_state IS NOT NULL
    AND compute_ids IS NULL
  GROUP BY workspace_id
),
-- distinct jobs sharing each cluster in the window (denominator for the naive split).
cluster_jobs AS (
  SELECT workspace_id, compute_id, COUNT(DISTINCT job_id) AS jobs_sharing_cluster
  FROM task_compute
  GROUP BY workspace_id, compute_id
),
price AS (
  SELECT sku_name, cloud, usage_unit, price_start_time, price_end_time,
         CAST(pricing.default AS DOUBLE) AS list_rate
  FROM system.billing.list_prices
),
cost_rollup AS (
  -- Pre-aggregated cost per all-purpose cluster over the SAME window as the finding.
  SELECT u.workspace_id,
         u.usage_metadata.cluster_id                      AS cluster_id,
         SUM(u.usage_quantity)                            AS net_dbus,
         SUM(u.usage_quantity * COALESCE(p.list_rate, 0)) AS est_usd_list
  FROM system.billing.usage u
  LEFT JOIN price p
    ON u.sku_name = p.sku_name AND u.cloud = p.cloud AND u.usage_unit = p.usage_unit
   AND u.usage_end_time >= p.price_start_time
   AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
  WHERE upper(u.usage_unit) = 'DBU'
    AND u.usage_metadata.cluster_id IS NOT NULL
    AND u.usage_date >= dateadd(DAY, -:period_days, current_date())
    AND u.usage_date <  current_date()
  GROUP BY u.workspace_id, u.usage_metadata.cluster_id
),
finding AS (
  SELECT tc.workspace_id,
         CAST(tc.job_id AS STRING)         AS job_id,
         CAST(tc.compute_id AS STRING)     AS compute_id,
         c.cluster_source,
         COUNT(DISTINCT tc.run_id)          AS task_runs,
         COALESCE(MAX(cr.net_dbus), 0)      AS net_dbus,
         COALESCE(MAX(cr.est_usd_list), 0)  AS est_usd_list,
         MAX(cj.jobs_sharing_cluster)       AS jobs_sharing_cluster
  FROM task_compute tc
  LEFT JOIN (
    SELECT workspace_id, cluster_id, cluster_source
    FROM system.compute.clusters
    QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, cluster_id ORDER BY change_time DESC) = 1
  ) c
    ON tc.workspace_id = c.workspace_id AND tc.compute_id = c.cluster_id
  LEFT JOIN cost_rollup cr
    ON tc.workspace_id = cr.workspace_id AND tc.compute_id = cr.cluster_id
  LEFT JOIN cluster_jobs cj
    ON tc.workspace_id = cj.workspace_id AND tc.compute_id = cj.compute_id
  GROUP BY tc.workspace_id, tc.job_id, tc.compute_id, c.cluster_source
)
SELECT
  workspace_id, job_id, compute_id, cluster_source, task_runs,
  net_dbus, est_usd_list, jobs_sharing_cluster,
  -- naive even split of the shared cluster's list-price cost across the jobs on it.
  est_usd_list / NULLIF(jobs_sharing_cluster, 0)                        AS est_usd_list_share,
  CASE
    WHEN cluster_source IS NULL THEN 'NOT_ASSESSED'
    WHEN cluster_source IN ('UI', 'API')
         AND est_usd_list / NULLIF(jobs_sharing_cluster, 0) >= :crit_share_usd THEN 'CRITICAL'
    WHEN cluster_source IN ('UI', 'API') THEN 'WARN'
    ELSE 'OK'
  END AS status
FROM finding
UNION ALL
SELECT
  workspace_id,
  CAST(NULL AS STRING)                       AS job_id,
  CAST(NULL AS STRING)                       AS compute_id,
  'null compute_ids'                         AS cluster_source,
  dropped_runs                               AS task_runs,
  CAST(NULL AS DOUBLE)                       AS net_dbus,
  CAST(NULL AS DOUBLE)                       AS est_usd_list,
  CAST(NULL AS BIGINT)                       AS jobs_sharing_cluster,
  CAST(NULL AS DOUBLE)                       AS est_usd_list_share,
  'NOT_ASSESSED'                             AS status
FROM dropped
ORDER BY est_usd_list_share DESC
LIMIT :top_n

WITH end_rows AS (
  SELECT workspace_id, job_id, run_id, run_type, trigger_type,
         result_state, termination_code, period_start_time, period_end_time
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= dateadd(day, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())   -- drop incomplete current day
    AND result_state IS NOT NULL                                   -- end row only
)
SELECT workspace_id, run_type, trigger_type, result_state, termination_code,
       COUNT(*)              AS run_rows,
       COUNT(DISTINCT run_id) AS distinct_runs,
       SUM(CASE WHEN result_state IN ('FAILED','ERROR','TIMED_OUT') THEN 1 ELSE 0 END) AS failed_run_rows,
       -- status: worst-first band on failed run rows per group (field heuristic; :warn_failed_run_rows / :crit_failed_run_rows).
       CASE
         WHEN SUM(CASE WHEN result_state IN ('FAILED','ERROR','TIMED_OUT') THEN 1 ELSE 0 END) >= :crit_failed_run_rows THEN 'CRITICAL'
         WHEN SUM(CASE WHEN result_state IN ('FAILED','ERROR','TIMED_OUT') THEN 1 ELSE 0 END) >= :warn_failed_run_rows THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM end_rows
GROUP BY workspace_id, run_type, trigger_type, result_state, termination_code
ORDER BY failed_run_rows DESC

WITH latest_jobs AS (
  SELECT workspace_id, job_id, name, health_rules, delete_time
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY workspace_id, job_id ORDER BY change_time DESC
  ) = 1
)
SELECT workspace_id,
       COUNT(*) AS active_jobs,
       -- NULL = column not yet populated (unknown), reported as a separate not-assessed bucket
       SUM(CASE WHEN health_rules IS NULL THEN 1 ELSE 0 END) AS jobs_health_rules_null,
       -- confident "has a health rule": non-null AND a non-empty rule set
       SUM(CASE WHEN health_rules IS NOT NULL AND CARDINALITY(health_rules) > 0
                THEN 1 ELSE 0 END)                            AS jobs_with_health_rule,
       -- confident "no health rule": non-null but empty rule set (populated window, no rule)
       SUM(CASE WHEN health_rules IS NOT NULL AND CARDINALITY(health_rules) = 0
                THEN 1 ELSE 0 END)                            AS jobs_no_health_rule,
       -- status: worst-first band on health-rule coverage pct (field heuristic; :warn_coverage_pct / :crit_coverage_pct).
       CASE
         WHEN SUM(CASE WHEN health_rules IS NULL THEN 1 ELSE 0 END) = COUNT(*) THEN 'NOT_ASSESSED'
         WHEN 100.0 * SUM(CASE WHEN health_rules IS NOT NULL AND CARDINALITY(health_rules) > 0 THEN 1 ELSE 0 END)
              / NULLIF(COUNT(*) - SUM(CASE WHEN health_rules IS NULL THEN 1 ELSE 0 END), 0) < :crit_coverage_pct THEN 'CRITICAL'
         WHEN 100.0 * SUM(CASE WHEN health_rules IS NOT NULL AND CARDINALITY(health_rules) > 0 THEN 1 ELSE 0 END)
              / NULLIF(COUNT(*) - SUM(CASE WHEN health_rules IS NULL THEN 1 ELSE 0 END), 0) < :warn_coverage_pct THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM latest_jobs
WHERE delete_time IS NULL
GROUP BY workspace_id
ORDER BY CASE status WHEN 'CRITICAL' THEN 1 WHEN 'NOT_ASSESSED' THEN 2 WHEN 'WARN' THEN 3 ELSE 4 END, active_jobs DESC

WITH latest_jobs AS (
  SELECT workspace_id, job_id, name, creator_user_name, run_as_user_name, delete_time
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY workspace_id, job_id ORDER BY change_time DESC
  ) = 1
),
-- normalise unavailable / redacted identity placeholders to NULL so they are never
-- counted as a real owner or a real mismatch.
norm AS (
  SELECT workspace_id, job_id, name,
         CASE WHEN creator_user_name IS NULL
                OR trim(creator_user_name) = ''
                OR upper(creator_user_name) = '__REDACTED__'
              THEN NULL ELSE creator_user_name END AS creator_user_name,
         CASE WHEN run_as_user_name IS NULL
                OR trim(run_as_user_name) = ''
                OR upper(run_as_user_name) = '__REDACTED__'
              THEN NULL ELSE run_as_user_name END  AS run_as_user_name
  FROM latest_jobs
  WHERE delete_time IS NULL
)
SELECT workspace_id,
       COUNT(*) AS active_jobs,
       SUM(CASE WHEN creator_user_name IS NULL THEN 1 ELSE 0 END) AS jobs_creator_null,
       SUM(CASE WHEN run_as_user_name  IS NULL THEN 1 ELSE 0 END) AS jobs_runas_null,
       -- confident mismatch: both identities known & different (handoff / SP run-as)
       SUM(CASE WHEN creator_user_name IS NOT NULL
                 AND run_as_user_name IS NOT NULL
                 AND creator_user_name <> run_as_user_name
                THEN 1 ELSE 0 END)                                AS jobs_owner_mismatch,
       -- orphan candidate: run-as identity unknown but creator known (no one clearly runs it)
       SUM(CASE WHEN run_as_user_name IS NULL AND creator_user_name IS NOT NULL
                THEN 1 ELSE 0 END)                                AS jobs_orphan_runas_unknown,
       -- status: worst-first band on combined ownership-risk job count (field heuristic;
       -- :warn_orphan_jobs / :crit_orphan_jobs).
       CASE
         WHEN COUNT(*) = SUM(CASE WHEN creator_user_name IS NULL THEN 1 ELSE 0 END) THEN 'NOT_ASSESSED'
         WHEN (SUM(CASE WHEN creator_user_name IS NOT NULL AND run_as_user_name IS NOT NULL AND creator_user_name <> run_as_user_name THEN 1 ELSE 0 END)
             + SUM(CASE WHEN run_as_user_name IS NULL AND creator_user_name IS NOT NULL THEN 1 ELSE 0 END)) >= :crit_orphan_jobs THEN 'CRITICAL'
         WHEN (SUM(CASE WHEN creator_user_name IS NOT NULL AND run_as_user_name IS NOT NULL AND creator_user_name <> run_as_user_name THEN 1 ELSE 0 END)
             + SUM(CASE WHEN run_as_user_name IS NULL AND creator_user_name IS NOT NULL THEN 1 ELSE 0 END)) >= :warn_orphan_jobs THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM norm
GROUP BY workspace_id
ORDER BY (jobs_owner_mismatch + jobs_orphan_runas_unknown) DESC

WITH end_rows AS (
  SELECT workspace_id, job_id, run_id, queue_duration_seconds
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= dateadd(day, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())   -- drop incomplete current day
    AND result_state IS NOT NULL                                   -- end row only
)
SELECT workspace_id, job_id,
       COUNT(DISTINCT run_id)                                   AS distinct_runs,
       SUM(queue_duration_seconds)                              AS queue_s_total,
       percentile_approx(queue_duration_seconds, 0.95)         AS queue_s_p95,
       SUM(CASE WHEN queue_duration_seconds IS NULL THEN 1 ELSE 0 END) AS runs_queue_null,
       -- status: worst-first band on p95 queue seconds (field heuristic; :warn_queue_p95_s / :crit_queue_p95_s).
       CASE
         WHEN percentile_approx(queue_duration_seconds, 0.95) IS NULL THEN 'NOT_ASSESSED'
         WHEN percentile_approx(queue_duration_seconds, 0.95) >= :crit_queue_p95_s THEN 'CRITICAL'
         WHEN percentile_approx(queue_duration_seconds, 0.95) >= :warn_queue_p95_s THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM end_rows
GROUP BY workspace_id, job_id
ORDER BY queue_s_p95 DESC

WITH latest_tasks AS (
  SELECT workspace_id, job_id, task_key, timeout_seconds, delete_time
  FROM system.lakeflow.job_tasks
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id, task_key ORDER BY change_time DESC) = 1
),
price AS (
  SELECT sku_name, cloud, usage_unit, price_start_time, price_end_time,
         CAST(pricing.default AS DOUBLE) AS list_rate
  FROM system.billing.list_prices
),
-- Pre-aggregated cost per (workspace_id, job_id) over the billing look-back window.
cost_rollup AS (
  SELECT u.workspace_id,
         u.usage_metadata.job_id                          AS job_id,
         SUM(u.usage_quantity)                            AS net_dbus,
         SUM(u.usage_quantity * COALESCE(p.list_rate, 0)) AS est_usd_list
  FROM system.billing.usage u
  LEFT JOIN price p
    ON u.sku_name = p.sku_name AND u.cloud = p.cloud AND u.usage_unit = p.usage_unit
   AND u.usage_end_time >= p.price_start_time
   AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
  WHERE upper(u.usage_unit) = 'DBU'
    AND u.usage_metadata.job_id IS NOT NULL
    AND u.usage_date >= date_add(current_date(), -:period_days)
    AND u.usage_date <  current_date()
  GROUP BY u.workspace_id, u.usage_metadata.job_id
),
-- Jobs that have >=1 active no-timeout task, with their at-risk cost summed to workspace grain (1 row/ws).
risk_job_cost AS (
  SELECT rj.workspace_id,
         SUM(COALESCE(cr.net_dbus, 0))     AS dbus_at_risk,
         SUM(COALESCE(cr.est_usd_list, 0)) AS est_usd_at_risk
  FROM (
    SELECT DISTINCT workspace_id, job_id
    FROM latest_tasks
    WHERE delete_time IS NULL
      AND (timeout_seconds IS NULL OR timeout_seconds = 0)
  ) rj
  LEFT JOIN cost_rollup cr
    ON cr.workspace_id = rj.workspace_id AND cr.job_id = rj.job_id
  GROUP BY rj.workspace_id
)
SELECT lt.workspace_id,
       COUNT(*) AS active_tasks,
       SUM(CASE WHEN lt.timeout_seconds IS NULL OR lt.timeout_seconds = 0 THEN 1 ELSE 0 END) AS tasks_no_timeout,
       SUM(CASE WHEN lt.timeout_seconds IS NULL THEN 1 ELSE 0 END) AS tasks_timeout_null,
       COALESCE(rjc.dbus_at_risk, 0)    AS net_dbus,
       COALESCE(rjc.est_usd_at_risk, 0) AS est_usd_list,
       -- status: worst-first band on no-timeout task count (field heuristic; :warn_no_timeout_tasks / :crit_no_timeout_tasks).
       CASE
         WHEN SUM(CASE WHEN lt.timeout_seconds IS NULL OR lt.timeout_seconds = 0 THEN 1 ELSE 0 END) >= :crit_no_timeout_tasks THEN 'CRITICAL'
         WHEN SUM(CASE WHEN lt.timeout_seconds IS NULL OR lt.timeout_seconds = 0 THEN 1 ELSE 0 END) >= :warn_no_timeout_tasks THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM latest_tasks lt
LEFT JOIN risk_job_cost rjc ON rjc.workspace_id = lt.workspace_id
WHERE lt.delete_time IS NULL
GROUP BY lt.workspace_id, rjc.dbus_at_risk, rjc.est_usd_at_risk
ORDER BY tasks_no_timeout DESC

WITH latest_jobs AS (
  SELECT workspace_id, job_id, name, timeout_seconds, paused, delete_time, create_time
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
),
price AS (
  SELECT sku_name, cloud, usage_unit, price_start_time, price_end_time,
         CAST(pricing.default AS DOUBLE) AS list_rate
  FROM system.billing.list_prices
),
cost_rollup AS (
  -- Per (workspace_id, job_id) DBUs + list-price $ over the cost-lookback window. job_id is NOT
  -- globally unique -> keyed on workspace_id + job_id. Pre-aggregated (1 row per workspace_id+job_id)
  -- so the LEFT JOIN below to the deduped job set is strictly 1:1 and cannot fan out.
  SELECT u.workspace_id,
         u.usage_metadata.job_id AS job_id,
         SUM(u.usage_quantity)                            AS net_dbus,
         SUM(u.usage_quantity * COALESCE(p.list_rate, 0)) AS est_usd_list
  FROM system.billing.usage u
  LEFT JOIN price p
    ON u.sku_name = p.sku_name AND u.cloud = p.cloud AND u.usage_unit = p.usage_unit
   AND u.usage_end_time >= p.price_start_time
   AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
  WHERE upper(u.usage_unit) = 'DBU'
    AND u.usage_metadata.job_id IS NOT NULL
    AND u.usage_date >= date_sub(current_date(), :period_days)
    AND u.usage_date <  current_date()
  GROUP BY u.workspace_id, u.usage_metadata.job_id
)
SELECT lj.workspace_id,
       COUNT(*) AS active_jobs,
       SUM(CASE WHEN lj.timeout_seconds IS NULL OR lj.timeout_seconds = 0 THEN 1 ELSE 0 END) AS jobs_no_timeout,
       SUM(CASE WHEN lj.timeout_seconds IS NULL THEN 1 ELSE 0 END) AS jobs_timeout_null,
       SUM(CASE WHEN lj.create_time     IS NULL THEN 1 ELSE 0 END) AS jobs_create_time_null,
       -- cost visibility: DBUs / list-price $ attributed to the flagged subset only
       -- (jobs with no timeout: timeout_seconds IS NULL OR = 0), summed over qualifying jobs.
       SUM(CASE WHEN lj.timeout_seconds IS NULL OR lj.timeout_seconds = 0
                THEN COALESCE(cr.net_dbus, 0) ELSE 0 END)     AS net_dbus,
       SUM(CASE WHEN lj.timeout_seconds IS NULL OR lj.timeout_seconds = 0
                THEN COALESCE(cr.est_usd_list, 0) ELSE 0 END) AS est_usd_list,
       -- status: worst-first band on no-timeout job count (field heuristic; :warn_no_timeout_jobs / :crit_no_timeout_jobs).
       CASE
         WHEN SUM(CASE WHEN lj.timeout_seconds IS NULL OR lj.timeout_seconds = 0 THEN 1 ELSE 0 END) >= :crit_no_timeout_jobs THEN 'CRITICAL'
         WHEN SUM(CASE WHEN lj.timeout_seconds IS NULL OR lj.timeout_seconds = 0 THEN 1 ELSE 0 END) >= :warn_no_timeout_jobs THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM latest_jobs lj
LEFT JOIN cost_rollup cr
  ON lj.workspace_id = cr.workspace_id AND lj.job_id = cr.job_id
WHERE lj.delete_time IS NULL
GROUP BY lj.workspace_id
ORDER BY jobs_no_timeout DESC

SELECT workspace_id, job_id, termination_code,
       COUNT(DISTINCT run_id) AS never_started_runs,
       -- status: worst-first band on never-started run count (field heuristic; :warn_never_started / :crit_never_started).
       CASE
         WHEN COUNT(DISTINCT run_id) >= :crit_never_started THEN 'CRITICAL'
         WHEN COUNT(DISTINCT run_id) >= :warn_never_started THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.lakeflow.job_run_timeline
WHERE period_start_time >= dateadd(day, -:period_days, current_date())
  AND period_end_time < date_trunc('DAY', current_timestamp())
  AND result_state IS NOT NULL
  AND period_start_time = period_end_time     -- never began execution
GROUP BY workspace_id, job_id, termination_code
ORDER BY never_started_runs DESC

SELECT workspace_id, job_id,
       COUNT(DISTINCT run_id)                    AS runs,
       SUM(setup_duration_seconds)               AS setup_s_total,
       SUM(queue_duration_seconds)               AS queue_s_total,
       SUM(execution_duration_seconds)           AS execution_s_total,
       SUM(cleanup_duration_seconds)             AS cleanup_s_total,
       SUM(run_duration_seconds)                 AS run_s_total,
       PERCENTILE(setup_duration_seconds, 0.95)  AS setup_s_p95,
       PERCENTILE(queue_duration_seconds, 0.95)  AS queue_s_p95,
       SUM(CASE WHEN setup_duration_seconds IS NULL THEN 1 ELSE 0 END) AS rows_setup_null,
       -- status: worst-first band on p95 setup (cold-start) seconds (field heuristic; :warn_setup_p95_s / :crit_setup_p95_s).
       CASE
         WHEN PERCENTILE(setup_duration_seconds, 0.95) IS NULL THEN 'NOT_ASSESSED'
         WHEN PERCENTILE(setup_duration_seconds, 0.95) >= :crit_setup_p95_s THEN 'CRITICAL'
         WHEN PERCENTILE(setup_duration_seconds, 0.95) >= :warn_setup_p95_s THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.lakeflow.job_run_timeline
WHERE period_start_time >= dateadd(day, -:period_days, current_date())
  AND period_end_time < date_trunc('DAY', current_timestamp())
  AND result_state IS NOT NULL     -- end row carries the final durations
GROUP BY workspace_id, job_id
ORDER BY setup_s_p95 DESC

WITH pipe_dbus AS (
  -- DBUs billed to each pipeline over the window. Net across ALL record_types; DBU only.
  -- Split maintenance DBUs out via dlt_maintenance_id so they are reported separately.
  SELECT workspace_id,
         usage_metadata.dlt_pipeline_id AS pipeline_id,
         SUM(usage_quantity) AS net_pipeline_dbus,
         SUM(CASE WHEN usage_metadata.dlt_maintenance_id IS NOT NULL
                  THEN usage_quantity ELSE 0 END) AS net_maintenance_dbus
  FROM system.billing.usage
  WHERE usage_date >= dateadd(day, -:period_days, current_date())
    AND usage_date < current_date()
    AND upper(usage_unit) = 'DBU'
    AND usage_metadata.dlt_pipeline_id IS NOT NULL
  GROUP BY workspace_id, usage_metadata.dlt_pipeline_id
),
pipe_list AS (
  -- LIST-$ estimate only (pre-discount, CAST-unverified): SUM(usage_quantity * list_rate)
  -- per pipeline. Apply your negotiated discount on top of this; never a billed dollar.
  SELECT u.workspace_id,
         u.usage_metadata.dlt_pipeline_id AS pipeline_id,
         MAX(lp.currency_code) AS currency_code,
         SUM(u.usage_quantity * lp.list_rate) AS net_list_cost
  FROM system.billing.usage u
  LEFT JOIN (
    SELECT sku_name, cloud, currency_code, usage_unit, price_start_time, price_end_time,
           CAST(pricing.effective_list.default AS DOUBLE) AS list_rate   -- <-- UNVERIFIED path
    FROM system.billing.list_prices
  ) lp
    ON u.sku_name = lp.sku_name
   AND u.cloud    = lp.cloud
   AND u.usage_end_time >= lp.price_start_time
   AND (lp.price_end_time IS NULL OR u.usage_end_time < lp.price_end_time)
  WHERE u.usage_date >= dateadd(day, -:period_days, current_date())
    AND u.usage_date < current_date()
    AND upper(u.usage_unit) = 'DBU'
    AND u.usage_metadata.dlt_pipeline_id IS NOT NULL
  GROUP BY u.workspace_id, u.usage_metadata.dlt_pipeline_id
),
pipe_meta AS (
  -- SCD2 latest row per (workspace_id, pipeline_id): names-not-ids + type.
  SELECT workspace_id, pipeline_id, name AS pipeline_name, pipeline_type
  FROM system.lakeflow.pipelines
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY workspace_id, pipeline_id ORDER BY change_time DESC
  ) = 1
),
pipe_updates AS (
  -- Over-refresh signal: distinct updates + total active seconds per pipeline.
  SELECT workspace_id, pipeline_id,
         COUNT(DISTINCT update_id) AS updates,
         SUM(unix_timestamp(period_end_time) - unix_timestamp(period_start_time)) AS active_seconds_total
  FROM system.lakeflow.pipeline_update_timeline
  WHERE period_start_time >= dateadd(day, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())
    AND result_state IS NOT NULL   -- update end row only
  GROUP BY workspace_id, pipeline_id
)
SELECT d.workspace_id,
       d.pipeline_id,
       CASE WHEN m.pipeline_name IS NULL THEN m.pipeline_name ELSE concat(substr(m.pipeline_name, 1, 2), '****') END AS pipeline_name,
       m.pipeline_type,
       d.net_pipeline_dbus,
       d.net_maintenance_dbus,
       l.currency_code,
       l.net_list_cost,
       COALESCE(up.updates, 0)              AS updates,
       COALESCE(up.active_seconds_total, 0) AS active_seconds_total,
       -- status: worst-first band on net pipeline DBUs in the window (field heuristic; :warn_pipeline_dbus / :crit_pipeline_dbus).
       CASE
         WHEN d.net_pipeline_dbus >= :crit_pipeline_dbus THEN 'CRITICAL'
         WHEN d.net_pipeline_dbus >= :warn_pipeline_dbus THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM pipe_dbus d
LEFT JOIN pipe_list l
  ON d.workspace_id = l.workspace_id AND d.pipeline_id = l.pipeline_id
LEFT JOIN pipe_meta m
  ON d.workspace_id = m.workspace_id AND d.pipeline_id = m.pipeline_id
LEFT JOIN pipe_updates up
  ON d.workspace_id = up.workspace_id AND d.pipeline_id = up.pipeline_id
WHERE d.net_pipeline_dbus > 0
ORDER BY d.net_pipeline_dbus DESC

WITH price AS (
  SELECT sku_name, cloud, usage_unit, price_start_time, price_end_time,
         CAST(pricing.default AS DOUBLE) AS list_rate
  FROM system.billing.list_prices
),
cost_rollup AS (
  SELECT u.workspace_id,
         u.usage_metadata.dlt_pipeline_id AS dlt_pipeline_id,
         SUM(u.usage_quantity)                            AS net_dbus,
         SUM(u.usage_quantity * COALESCE(p.list_rate, 0)) AS est_usd_list
  FROM system.billing.usage u
  LEFT JOIN price p
    ON u.sku_name = p.sku_name AND u.cloud = p.cloud AND u.usage_unit = p.usage_unit
   AND u.usage_end_time >= p.price_start_time
   AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
  WHERE upper(u.usage_unit) = 'DBU'
    AND u.usage_metadata.dlt_pipeline_id IS NOT NULL
    AND u.usage_date >= dateadd(day, -:period_days, current_date())
    AND u.usage_date <  current_date()
  GROUP BY u.workspace_id, u.usage_metadata.dlt_pipeline_id
)
-- NEEDS CONFIRMATION: settings.<key> dot-access is UNVERIFIED (struct vs map).
SELECT u.workspace_id, u.pipeline_id, p.pipeline_type, p.setting_continuous, p.setting_development,
       COUNT(DISTINCT u.update_id) AS updates,
       SUM(unix_timestamp(u.period_end_time) - unix_timestamp(u.period_start_time)) AS active_seconds_total,
       COALESCE(MAX(cr.net_dbus), 0)     AS net_dbus,
       COALESCE(MAX(cr.est_usd_list), 0) AS est_usd_list,
       -- status: worst-first band on net DBUs for a non-continuous pipeline (field heuristic; :warn_idle_dbus / :crit_idle_dbus).
       CASE
         WHEN p.setting_continuous = true THEN 'NOT_ASSESSED'
         WHEN COALESCE(MAX(cr.net_dbus), 0) >= :crit_idle_dbus THEN 'CRITICAL'
         WHEN COALESCE(MAX(cr.net_dbus), 0) >= :warn_idle_dbus THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.lakeflow.pipeline_update_timeline u
LEFT JOIN (
  SELECT workspace_id, pipeline_id, pipeline_type,
         settings.continuous  AS setting_continuous,
         settings.development AS setting_development
  FROM system.lakeflow.pipelines
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id ORDER BY change_time DESC) = 1
) p
  ON u.workspace_id = p.workspace_id AND u.pipeline_id = p.pipeline_id
LEFT JOIN cost_rollup cr
  ON u.workspace_id = cr.workspace_id AND u.pipeline_id = cr.dlt_pipeline_id
WHERE u.period_start_time >= dateadd(day, -:period_days, current_date())
  AND u.period_end_time < date_trunc('DAY', current_timestamp())
  AND u.result_state IS NOT NULL
GROUP BY u.workspace_id, u.pipeline_id, p.pipeline_type, p.setting_continuous, p.setting_development
ORDER BY net_dbus DESC

WITH end_rows AS (
  SELECT workspace_id, pipeline_id, update_id, request_id, update_type,
         trigger_type, result_state, period_start_time, period_end_time
  FROM system.lakeflow.pipeline_update_timeline
  WHERE period_start_time >= dateadd(day, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())
    AND result_state IS NOT NULL          -- end row only for updates >1h
)
SELECT e.workspace_id, e.pipeline_id, e.update_type, e.trigger_type, e.result_state,
       COUNT(DISTINCT e.update_id) AS updates,
       SUM(CASE WHEN e.result_state = 'FAILED'             THEN 1 ELSE 0 END) AS failed_update_rows,
       SUM(CASE WHEN e.trigger_type = 'RETRY_ON_FAILURE'   THEN 1 ELSE 0 END) AS retry_triggered_rows,
       -- status: worst-first band on failed update rows per group (field heuristic; :warn_failed_updates / :crit_failed_updates).
       CASE
         WHEN SUM(CASE WHEN e.result_state = 'FAILED' THEN 1 ELSE 0 END) >= :crit_failed_updates THEN 'CRITICAL'
         WHEN SUM(CASE WHEN e.result_state = 'FAILED' THEN 1 ELSE 0 END) >= :warn_failed_updates THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM end_rows e
GROUP BY e.workspace_id, e.pipeline_id, e.update_type, e.trigger_type, e.result_state
ORDER BY failed_update_rows DESC

-- NEEDS CONFIRMATION: settings.<key> dot-access vs settings['<key>'] map-access is UNVERIFIED.
WITH latest_pipelines AS (
  SELECT workspace_id, pipeline_id, pipeline_type, name, created_by, run_as, settings, delete_time,
         settings.serverless  AS setting_serverless,
         settings.development AS setting_development,
         settings.continuous  AS setting_continuous,
         settings.photon      AS setting_photon,
         settings.edition     AS setting_edition,
         settings.channel     AS setting_channel
  FROM system.lakeflow.pipelines
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, pipeline_id ORDER BY change_time DESC) = 1
)
SELECT workspace_id, pipeline_type, setting_serverless, setting_development,
       setting_continuous, setting_edition,
       COUNT(*) AS pipelines
FROM latest_pipelines
WHERE delete_time IS NULL
GROUP BY workspace_id, pipeline_type, setting_serverless, setting_development, setting_continuous, setting_edition
ORDER BY workspace_id, pipeline_type

WITH end_rows AS (
  SELECT workspace_id, job_id, run_id
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= dateadd(day, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())
    AND result_state IS NOT NULL          -- one non-NULL result_state row per attempt
),
per_run AS (
  SELECT workspace_id, job_id, run_id, COUNT(*) AS attempt_rows
  FROM end_rows GROUP BY workspace_id, job_id, run_id
),
price AS (
  SELECT sku_name, cloud, usage_unit, price_start_time, price_end_time,
         CAST(pricing.default AS DOUBLE) AS list_rate
  FROM system.billing.list_prices
),
cost_rollup AS (
  -- Pre-aggregated cost per (workspace_id, job_id). job_id is not globally unique, so we key on
  -- workspace_id + job_id. Window mirrors this query's trailing window via usage_date.
  SELECT u.workspace_id,
         u.usage_metadata.job_id                          AS job_id,
         SUM(u.usage_quantity)                            AS net_dbus,
         SUM(u.usage_quantity * COALESCE(p.list_rate, 0)) AS est_usd_list
  FROM system.billing.usage u
  LEFT JOIN price p
    ON u.sku_name = p.sku_name AND u.cloud = p.cloud AND u.usage_unit = p.usage_unit
   AND u.usage_end_time >= p.price_start_time
   AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
  WHERE upper(u.usage_unit) = 'DBU'
    AND u.usage_metadata.job_id IS NOT NULL
    AND u.usage_date >= dateadd(day, -:period_days, current_date())
    AND u.usage_date <  current_date()
  GROUP BY u.workspace_id, u.usage_metadata.job_id
)
SELECT pr.workspace_id, pr.job_id,
       COUNT(*)             AS distinct_runs,
       SUM(pr.attempt_rows)    AS total_attempt_rows,
       SUM(pr.attempt_rows - 1) AS total_retries,
       SUM(CASE WHEN pr.attempt_rows > 1 THEN 1 ELSE 0 END) AS runs_with_retry,
       COALESCE(MAX(cr.net_dbus), 0)     AS net_dbus,
       COALESCE(MAX(cr.est_usd_list), 0) AS est_usd_list,
       -- status: worst-first band on total retries in the window (field heuristic; :warn_total_retries / :crit_total_retries).
       CASE
         WHEN SUM(pr.attempt_rows - 1) >= :crit_total_retries THEN 'CRITICAL'
         WHEN SUM(pr.attempt_rows - 1) >= :warn_total_retries THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM per_run pr
LEFT JOIN cost_rollup cr
  ON pr.workspace_id = cr.workspace_id AND pr.job_id = cr.job_id
GROUP BY pr.workspace_id, pr.job_id
ORDER BY total_retries DESC

WITH latest_jobs AS (
  SELECT workspace_id, job_id, name, run_as, creator_id, delete_time
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
),
last_run AS (
  SELECT workspace_id, job_id, MAX(period_start_time) AS last_run_start
  FROM system.lakeflow.job_run_timeline
  GROUP BY workspace_id, job_id
),
price AS (
  SELECT sku_name, cloud, usage_unit, price_start_time, price_end_time,
         CAST(pricing.default AS DOUBLE) AS list_rate
  FROM system.billing.list_prices
),
cost_rollup AS (
  SELECT u.workspace_id,
         u.usage_metadata.job_id AS job_id,
         SUM(u.usage_quantity)                            AS net_dbus,
         SUM(u.usage_quantity * COALESCE(p.list_rate, 0)) AS est_usd_list
  FROM system.billing.usage u
  LEFT JOIN price p
    ON u.sku_name = p.sku_name AND u.cloud = p.cloud AND u.usage_unit = p.usage_unit
   AND u.usage_end_time >= p.price_start_time
   AND (p.price_end_time IS NULL OR u.usage_end_time < p.price_end_time)
  WHERE upper(u.usage_unit) = 'DBU'
    AND u.usage_metadata.job_id IS NOT NULL
    AND u.usage_date >= dateadd(day, -:period_days, current_date())
    AND u.usage_date <  current_date()
  GROUP BY u.workspace_id, u.usage_metadata.job_id
)
SELECT j.workspace_id, j.job_id,
       CASE WHEN j.name IS NULL THEN j.name ELSE concat(substr(j.name, 1, 2), '****') END AS name,
       r.last_run_start,
       CASE WHEN r.last_run_start IS NULL
              OR r.last_run_start < dateadd(day, -:stale_days, current_date())
            THEN 1 ELSE 0 END AS is_stale,
       COALESCE(cr.net_dbus, 0)     AS net_dbus,
       COALESCE(cr.est_usd_list, 0) AS est_usd_list,
       -- status: worst-first band on days since last run (field heuristic; :stale_days / :crit_stale_days).
       CASE
         WHEN r.last_run_start IS NULL THEN 'WARN'   -- no run ever recorded for this job in job_run_timeline history
         WHEN r.last_run_start < dateadd(day, -:crit_stale_days, current_date()) THEN 'CRITICAL'
         WHEN r.last_run_start < dateadd(day, -:stale_days, current_date()) THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM latest_jobs j
LEFT JOIN last_run r ON j.workspace_id = r.workspace_id AND j.job_id = r.job_id
LEFT JOIN cost_rollup cr ON j.workspace_id = cr.workspace_id AND j.job_id = cr.job_id
WHERE j.delete_time IS NULL
ORDER BY last_run_start ASC

WITH job_end AS (
  SELECT workspace_id, job_id, run_id AS job_run_id, result_state AS job_result_state
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= dateadd(day, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())
    AND result_state IS NOT NULL
),
task_end AS (
  SELECT workspace_id, job_id, job_run_id, task_key, result_state AS task_result_state
  FROM system.lakeflow.job_task_run_timeline
  WHERE period_start_time >= dateadd(day, -:period_days, current_date())
    AND result_state IS NOT NULL
)
SELECT j.workspace_id, j.job_id,
       COUNT(DISTINCT j.job_run_id) AS succeeded_runs,
       COUNT(DISTINCT CASE WHEN t.task_result_state IN ('FAILED','ERROR','TIMED_OUT')
                           THEN j.job_run_id END) AS succeeded_runs_with_failed_task,
       -- status: worst-first band on succeeded-but-had-a-failed-task run count (field heuristic;
       -- :warn_succeeded_with_failed / :crit_succeeded_with_failed).
       CASE
         WHEN COUNT(DISTINCT CASE WHEN t.task_result_state IN ('FAILED','ERROR','TIMED_OUT') THEN j.job_run_id END) >= :crit_succeeded_with_failed THEN 'CRITICAL'
         WHEN COUNT(DISTINCT CASE WHEN t.task_result_state IN ('FAILED','ERROR','TIMED_OUT') THEN j.job_run_id END) >= :warn_succeeded_with_failed THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM job_end j
LEFT JOIN task_end t
  ON j.workspace_id = t.workspace_id AND j.job_id = t.job_id AND j.job_run_id = t.job_run_id
WHERE j.job_result_state = 'SUCCEEDED'
GROUP BY j.workspace_id, j.job_id
ORDER BY succeeded_runs_with_failed_task DESC

WITH latest_tasks AS (
  SELECT workspace_id, job_id, task_key, timeout_seconds, delete_time
  FROM system.lakeflow.job_tasks
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY workspace_id, job_id, task_key ORDER BY change_time DESC
  ) = 1
),
task_end AS (
  SELECT workspace_id, job_id, task_key, run_id, execution_duration_seconds
  FROM system.lakeflow.job_task_run_timeline
  WHERE period_start_time >= dateadd(day, -:period_days, current_date())
    AND period_end_time < date_trunc('DAY', current_timestamp())   -- drop incomplete current day
    AND result_state IS NOT NULL                                   -- task end row only
)
SELECT t.workspace_id, t.job_id, t.task_key,
       lt.timeout_seconds,
       COUNT(DISTINCT t.run_id)                                          AS task_runs,
       MAX(t.execution_duration_seconds)                                 AS max_exec_s,
       percentile_approx(t.execution_duration_seconds, 0.95)            AS exec_s_p95,
       -- runs whose execution reached >= :near_timeout_ratio of a CONFIGURED (non-null, >0) timeout
       SUM(CASE WHEN lt.timeout_seconds IS NOT NULL AND lt.timeout_seconds > 0
                 AND t.execution_duration_seconds >= :near_timeout_ratio * lt.timeout_seconds
                THEN 1 ELSE 0 END)                                       AS runs_near_timeout,
       -- runs that ran AT/PAST the configured timeout (should have been killed)
       SUM(CASE WHEN lt.timeout_seconds IS NOT NULL AND lt.timeout_seconds > 0
                 AND t.execution_duration_seconds >= lt.timeout_seconds
                THEN 1 ELSE 0 END)                                       AS runs_over_timeout,
       -- degradation buckets: a NULL/zero timeout is "no bound configured" (not a finding);
       -- a NULL execution duration is "column not yet populated" (not assessed).
       SUM(CASE WHEN lt.timeout_seconds IS NULL OR lt.timeout_seconds = 0
                THEN 1 ELSE 0 END)                                       AS runs_no_task_timeout,
       SUM(CASE WHEN t.execution_duration_seconds IS NULL THEN 1 ELSE 0 END) AS runs_exec_null,
       -- status: worst-first band on near/over-timeout run count (field heuristic;
       -- :warn_near_timeout_runs / :crit_near_timeout_runs).
       CASE
         WHEN (SUM(CASE WHEN lt.timeout_seconds IS NOT NULL AND lt.timeout_seconds > 0
                    AND t.execution_duration_seconds >= :near_timeout_ratio * lt.timeout_seconds THEN 1 ELSE 0 END)
             + SUM(CASE WHEN lt.timeout_seconds IS NOT NULL AND lt.timeout_seconds > 0
                    AND t.execution_duration_seconds >= lt.timeout_seconds THEN 1 ELSE 0 END)) >= :crit_near_timeout_runs THEN 'CRITICAL'
         WHEN (SUM(CASE WHEN lt.timeout_seconds IS NOT NULL AND lt.timeout_seconds > 0
                    AND t.execution_duration_seconds >= :near_timeout_ratio * lt.timeout_seconds THEN 1 ELSE 0 END)
             + SUM(CASE WHEN lt.timeout_seconds IS NOT NULL AND lt.timeout_seconds > 0
                    AND t.execution_duration_seconds >= lt.timeout_seconds THEN 1 ELSE 0 END)) >= :warn_near_timeout_runs THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM task_end t
LEFT JOIN latest_tasks lt
  ON  t.workspace_id = lt.workspace_id
  AND t.job_id       = lt.job_id
  AND t.task_key     = lt.task_key
GROUP BY t.workspace_id, t.job_id, t.task_key, lt.timeout_seconds
ORDER BY (runs_near_timeout + runs_over_timeout) DESC

SELECT workspace_id, termination_code,
       COUNT(*)              AS run_rows,
       COUNT(DISTINCT run_id) AS distinct_runs
FROM system.lakeflow.job_run_timeline
WHERE period_start_time >= dateadd(day, -:period_days, current_date())
  AND period_end_time < date_trunc('DAY', current_timestamp())
  AND result_state IS NOT NULL          -- end row only
  AND termination_code IS NOT NULL
GROUP BY workspace_id, termination_code
ORDER BY workspace_id, run_rows DESC

-- NEEDS CONFIRMATION: termination_type column exists but its value enum is UNVERIFIED.
-- This probe discovers distinct values at runtime; confirm the column populates on the account.
SELECT workspace_id, termination_type, COUNT(*) AS run_rows
FROM system.lakeflow.job_run_timeline
WHERE period_start_time >= dateadd(day, -:period_days, current_date())
  AND result_state IS NOT NULL
GROUP BY workspace_id, termination_type
ORDER BY workspace_id, run_rows DESC

SELECT workspace_id, run_type, trigger_type,
       date_trunc('DAY', period_start_time) AS run_day,
       COUNT(DISTINCT run_id) AS distinct_runs,
       SUM(CASE WHEN result_state IS NOT NULL THEN 1 ELSE 0 END) AS completed_run_rows,
       SUM(execution_duration_seconds) AS execution_s_total
FROM system.lakeflow.job_run_timeline
WHERE period_start_time >= dateadd(day, -:period_days, current_date())
  AND period_start_time < date_trunc('DAY', current_timestamp())
GROUP BY workspace_id, run_type, trigger_type, date_trunc('DAY', period_start_time)
ORDER BY workspace_id, run_day DESC