X-ray your Databricks spend before your renewal

-- BigHammer Cost X-Ray | Query 1 of 3: spend summary
-- Read-only. Last 90 days, priced at Databricks list price.
SELECT
  u.billing_origin_product,
  u.sku_name,
  u.usage_metadata.job_run_id IS NOT NULL AS traced_to_job,
  lp.sku_name IS NOT NULL                 AS priced,
  MAX(lp.currency_code)                   AS currency_code,
  ROUND(SUM(IF(u.usage_unit = 'DBU', u.usage_quantity, 0)), 2) AS dbus,
  ROUND(SUM(u.usage_quantity *
        COALESCE(lp.pricing.effective_list.default, lp.pricing.default)), 2) AS list_cost
FROM system.billing.usage AS u
LEFT JOIN system.billing.list_prices AS lp
  ON  u.sku_name   = lp.sku_name
  AND u.cloud      = lp.cloud
  AND u.usage_unit = lp.usage_unit
  AND u.usage_start_time >= lp.price_start_time
  AND (lp.price_end_time IS NULL OR u.usage_start_time < lp.price_end_time)
WHERE u.usage_date >= date_sub(current_date(), 90)
GROUP BY 1, 2, 3, 4
ORDER BY list_cost DESC NULLS LAST

-- BigHammer Cost X-Ray | Query 2 of 3: cost per job
-- Read-only. Last 90 days, priced at Databricks list price.
-- Jobs on all-purpose clusters are billed to the cluster, not the job
-- (usage_metadata.job_id is empty for them), so ap_cluster_cost estimates
-- that spend from the clusters their scheduled tasks ran on.
WITH runs AS (
  -- runs over 1 hour span several rows; take the final state of each run
  SELECT workspace_id, job_id, run_id,
         MAX_BY(result_state, period_end_time) FILTER (WHERE result_state IS NOT NULL) AS result_state,
         MAX(trigger_type) AS trigger_type
  FROM system.lakeflow.job_run_timeline
  WHERE period_start_time >= date_sub(current_date(), 90)
  GROUP BY 1, 2, 3
),
priced AS (
  SELECT u.*, u.usage_quantity * COALESCE(lp.pricing.effective_list.default, lp.pricing.default) AS cost
  FROM system.billing.usage AS u
  JOIN system.billing.list_prices AS lp
    ON  u.sku_name   = lp.sku_name
    AND u.cloud      = lp.cloud
    AND u.usage_unit = lp.usage_unit
    AND u.usage_start_time >= lp.price_start_time
    AND (lp.price_end_time IS NULL OR u.usage_start_time < lp.price_end_time)
  WHERE u.usage_date >= date_sub(current_date(), 90)
),
run_cost AS (
  -- measured: usage tagged with a job run (job compute and serverless)
  SELECT workspace_id, usage_metadata.job_id AS job_id, usage_metadata.job_run_id AS run_id,
         SUM(cost) AS list_cost,
         SUM(IF(sku_name LIKE '%ALL_PURPOSE%' AND sku_name NOT LIKE '%SERVERLESS%', cost, 0)) AS all_purpose_cost,
         SUM(IF(sku_name LIKE '%SERVERLESS%', cost, 0)) AS serverless_cost
  FROM priced
  WHERE usage_metadata.job_id IS NOT NULL
  GROUP BY 1, 2, 3
),
latest_clusters AS (
  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
),
task_compute AS (
  SELECT workspace_id, job_id, job_run_id, period_start_time, period_end_time,
         explode(compute_ids) AS cluster_id
  FROM system.lakeflow.job_task_run_timeline
  WHERE period_start_time >= date_sub(current_date(), 90)
),
ap_task_slices AS (
  -- scheduled task runs on all-purpose clusters (created in the UI or API), one row per hour touched
  SELECT tc.workspace_id, tc.job_id, tc.cluster_id,
         explode(sequence(date_trunc('HOUR', tc.period_start_time),
                          date_trunc('HOUR', tc.period_end_time), INTERVAL 1 HOUR)) AS usage_hour
  FROM task_compute AS tc
  JOIN runs AS r
    ON tc.workspace_id = r.workspace_id AND tc.job_id = r.job_id AND tc.job_run_id = r.run_id
  JOIN latest_clusters AS c
    ON tc.workspace_id = c.workspace_id AND tc.cluster_id = c.cluster_id
  WHERE r.trigger_type IN ('CRON', 'PERIODIC', 'CONTINUOUS', 'FILE_ARRIVAL', 'TABLE')
    AND c.cluster_source IN ('UI', 'API')
),
ap_hours AS (
  SELECT DISTINCT workspace_id, job_id, cluster_id, usage_hour FROM ap_task_slices
),
ap_cluster_hour_cost AS (
  SELECT workspace_id, usage_metadata.cluster_id AS cluster_id,
         date_trunc('HOUR', usage_start_time) AS usage_hour, SUM(cost) AS cost
  FROM priced
  WHERE sku_name LIKE '%ALL_PURPOSE%' AND sku_name NOT LIKE '%SERVERLESS%'
    AND usage_metadata.cluster_id IS NOT NULL
  GROUP BY 1, 2, 3
),
ap_alloc AS (
  -- estimated upper bound: the cluster's cost in each hour a scheduled task ran on it,
  -- split evenly between jobs sharing that hour so no dollar is counted twice
  SELECT h.workspace_id, h.job_id, SUM(chc.cost / h.jobs_in_hour) AS ap_cluster_cost
  FROM (SELECT *, COUNT(*) OVER (PARTITION BY workspace_id, cluster_id, usage_hour) AS jobs_in_hour
        FROM ap_hours) AS h
  JOIN ap_cluster_hour_cost AS chc
    ON h.workspace_id = chc.workspace_id AND h.cluster_id = chc.cluster_id AND h.usage_hour = chc.usage_hour
  GROUP BY 1, 2
),
job_runs AS (
  SELECT r.workspace_id, r.job_id,
    COUNT(*) AS runs,
    COUNT_IF(r.result_state IN ('FAILED', 'TIMED_OUT', 'ERROR')) AS failed_runs,
    COUNT_IF(r.trigger_type IN ('CRON', 'PERIODIC', 'CONTINUOUS', 'FILE_ARRIVAL', 'TABLE')) AS scheduled_runs,
    SUM(COALESCE(rc.list_cost, 0)) AS list_cost,
    SUM(IF(r.result_state IN ('FAILED', 'TIMED_OUT', 'ERROR'), COALESCE(rc.list_cost, 0), 0)) AS failed_cost,
    SUM(COALESCE(rc.all_purpose_cost, 0)) AS all_purpose_cost,
    SUM(COALESCE(rc.serverless_cost, 0))  AS serverless_cost
  FROM runs AS r
  LEFT JOIN run_cost AS rc
    ON r.workspace_id = rc.workspace_id AND r.job_id = rc.job_id AND r.run_id = rc.run_id
  GROUP BY 1, 2
),
jobs AS (
  -- system.lakeflow.jobs is slowly changing: keep the latest row per job
  SELECT workspace_id, job_id, name
  FROM system.lakeflow.jobs
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, job_id ORDER BY change_time DESC) = 1
)
SELECT
  jr.workspace_id,
  jr.job_id,
  COALESCE(j.name, CONCAT('job ', jr.job_id)) AS job_name,
  jr.runs,
  jr.failed_runs,
  jr.scheduled_runs,
  ROUND(jr.list_cost, 2)        AS list_cost,
  ROUND(jr.failed_cost, 2)      AS failed_cost,
  ROUND(jr.all_purpose_cost, 2) AS all_purpose_cost,
  ROUND(jr.serverless_cost, 2)  AS serverless_cost,
  ROUND(COALESCE(a.ap_cluster_cost, 0), 2) AS ap_cluster_cost
FROM job_runs AS jr
LEFT JOIN ap_alloc AS a ON jr.workspace_id = a.workspace_id AND jr.job_id = a.job_id
LEFT JOIN jobs AS j     ON jr.workspace_id = j.workspace_id AND jr.job_id = j.job_id
WHERE jr.list_cost > 0 OR a.ap_cluster_cost > 0
ORDER BY jr.list_cost + COALESCE(a.ap_cluster_cost, 0) DESC
LIMIT 1000

-- BigHammer Cost X-Ray | Query 3 of 3: cluster CPU utilisation
-- Read-only. Worker nodes only, last 90 days, per-minute samples.
WITH cpu AS (
  SELECT
    workspace_id,
    cluster_id,
    COUNT(DISTINCT date_trunc('MINUTE', start_time)) / 60.0 AS hours_observed,
    COUNT(*) / 60.0                                           AS worker_node_hours,
    AVG(cpu_user_percent + cpu_system_percent)                AS avg_cpu_percent,
    PERCENTILE_APPROX(cpu_user_percent + cpu_system_percent, 0.95) AS p95_cpu_percent
  FROM system.compute.node_timeline
  WHERE start_time >= date_sub(current_date(), 90)
    AND driver = false
  GROUP BY 1, 2
  HAVING COUNT(DISTINCT date_trunc('MINUTE', start_time)) >= 30
),
clusters AS (
  -- system.compute.clusters is slowly changing: keep the latest row per cluster
  SELECT workspace_id, cluster_id, cluster_name, cluster_source, worker_node_type
  FROM system.compute.clusters
  QUALIFY ROW_NUMBER() OVER (PARTITION BY workspace_id, cluster_id ORDER BY change_time DESC) = 1
),
cluster_cost AS (
  SELECT
    u.workspace_id,
    u.usage_metadata.cluster_id AS cluster_id,
    SUM(u.usage_quantity * COALESCE(lp.pricing.effective_list.default, lp.pricing.default)) AS list_cost
  FROM system.billing.usage AS u
  JOIN system.billing.list_prices AS lp
    ON  u.sku_name   = lp.sku_name
    AND u.cloud      = lp.cloud
    AND u.usage_unit = lp.usage_unit
    AND u.usage_start_time >= lp.price_start_time
    AND (lp.price_end_time IS NULL OR u.usage_start_time < lp.price_end_time)
  WHERE u.usage_date >= date_sub(current_date(), 90)
    AND u.usage_metadata.cluster_id IS NOT NULL
  GROUP BY 1, 2
)
SELECT
  c.workspace_id,
  c.cluster_id,
  cl.cluster_name,
  cl.cluster_source,
  cl.worker_node_type,
  ROUND(c.hours_observed, 1)    AS hours_observed,
  ROUND(c.worker_node_hours, 1) AS worker_node_hours,
  ROUND(c.avg_cpu_percent, 1)   AS avg_cpu_percent,
  ROUND(c.p95_cpu_percent, 1)   AS p95_cpu_percent,
  ROUND(COALESCE(cc.list_cost, 0), 2) AS list_cost
FROM cpu AS c
LEFT JOIN clusters AS cl
  ON c.workspace_id = cl.workspace_id AND c.cluster_id = cl.cluster_id
LEFT JOIN cluster_cost AS cc
  ON c.workspace_id = cc.workspace_id AND c.cluster_id = cc.cluster_id
ORDER BY list_cost DESC
LIMIT 1000