Load and cost

SELECT
  LEFT(regexp_replace(statement_text, '\\s+', ' '), 120) AS shape,
  client_application,
  COUNT(*)                                              AS runs,
  ROUND(SUM(total_duration_ms)     / 3.6e6, 2)          AS wall_hours,
  ROUND(SUM(execution_duration_ms) / 3.6e6, 2)          AS exec_hours,
  ROUND(percentile(total_duration_ms, 0.95) / 1000, 1)  AS p95_seconds,
  ROUND(SUM(read_bytes) / POW(1024, 4), 3)              AS read_tib,
  SUM(IF(from_result_cache, 1, 0))                      AS cache_hits,
  SUM(IF(spilled_local_bytes > 0, 1, 0))                AS spilling_runs
FROM system.query.history
WHERE start_time >= current_timestamp() - INTERVAL 7 DAYS
  AND statement_type = 'SELECT'
  AND execution_status = 'FINISHED'
GROUP BY ALL
ORDER BY wall_hours DESC
LIMIT 20;

WITH priced AS (
  SELECT
    -- job compute + serverless jobs carry job_id; Lakeflow pipelines carry
    -- dlt_pipeline_id; a task on ALL-PURPOSE compute carries neither.
    COALESCE(u.usage_metadata.job_id,
             u.usage_metadata.dlt_pipeline_id,
             '(unattributed)')  AS workload_id,
    CASE
      WHEN u.usage_metadata.job_id          IS NOT NULL THEN 'job'
      WHEN u.usage_metadata.dlt_pipeline_id IS NOT NULL THEN 'pipeline'
      ELSE 'unattributed'
    END                        AS workload_kind,
    u.billing_origin_product,
    u.custom_tags['domain']    AS domain,
    u.usage_date,
    u.usage_quantity * p.pricing.effective_list.default AS list_cost
  FROM system.billing.usage u
  JOIN system.billing.list_prices p
    ON  u.cloud    = p.cloud
    AND u.sku_name = p.sku_name
    AND u.usage_start_time >= p.price_start_time
    AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
  WHERE u.usage_date >= current_date() - INTERVAL 36 DAYS
    AND u.usage_date <  current_date() - INTERVAL 1 DAYS   -- closed days only
    -- deliberately NO "job_id IS NOT NULL": that filter is what hid the two
    -- pipeline tasks of dwh_daily and every all-purpose-compute job on the estate
),
weekly AS (
  SELECT
    domain,
    workload_kind,
    workload_id,
    billing_origin_product,
    SUM(IF(usage_date >= current_date() - INTERVAL 8 DAYS, list_cost, 0))     AS last_week,
    SUM(IF(usage_date <  current_date() - INTERVAL 8 DAYS, list_cost, 0)) / 4 AS prior_avg
  FROM priced
  GROUP BY ALL
)
SELECT
  domain,
  workload_kind,
  workload_id,
  billing_origin_product,
  ROUND(last_week, 2)                                    AS last_week,
  ROUND(prior_avg, 2)                                    AS prior_weekly_avg,
  ROUND(100 * (last_week / NULLIF(prior_avg, 0) - 1), 1) AS pct_change
FROM weekly
WHERE prior_avg > 0 AND last_week > 1.25 * prior_avg
ORDER BY pct_change DESC;

WITH cluster_name AS (            -- system.compute.clusters is SCD2 as well:
  SELECT workspace_id, cluster_id, cluster_name, owned_by   -- one row per EDIT
  FROM system.compute.clusters
  QUALIFY ROW_NUMBER() OVER (
    PARTITION BY workspace_id, cluster_id ORDER BY change_time DESC) = 1
)
SELECT
  c.cluster_name,
  c.owned_by,
  ROUND(SUM(u.usage_quantity * p.pricing.effective_list.default), 2) AS list_cost
FROM system.billing.usage u
JOIN system.billing.list_prices p
  ON  u.cloud    = p.cloud
  AND u.sku_name = p.sku_name
  AND u.usage_start_time >= p.price_start_time
  AND (p.price_end_time IS NULL OR u.usage_start_time < p.price_end_time)
JOIN cluster_name c
  ON  c.workspace_id = u.workspace_id
  AND c.cluster_id   = u.usage_metadata.cluster_id
WHERE u.usage_date >= current_date() - INTERVAL 8 DAYS
  AND u.usage_date <  current_date() - INTERVAL 1 DAYS
  AND u.usage_metadata.job_id          IS NULL
  AND u.usage_metadata.dlt_pipeline_id IS NULL
  AND u.usage_metadata.cluster_id      IS NOT NULL
GROUP BY ALL
ORDER BY list_cost DESC
LIMIT 20;