Compute

WITH ordered AS (
    SELECT
        warehouse_id,
        event_type,
        event_time,
        LEAD(event_time) OVER (PARTITION BY warehouse_id ORDER BY event_time) AS next_event_time
    FROM system.compute.warehouse_events
    WHERE event_time >= current_timestamp() - INTERVAL :period_days DAYS
),
gaps AS (
    SELECT
        warehouse_id,
        event_type,
        -- gap (seconds) this warehouse spent in `event_type` until its NEXT event.
        -- The last event per warehouse has next_event_time = NULL -> gap NULL (open, excluded).
        CASE
            WHEN next_event_time IS NULL THEN NULL
            ELSE unix_timestamp(next_event_time) - unix_timestamp(event_time)
        END AS gap_seconds
    FROM ordered
),
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 per-warehouse DBU/$ over the SAME :period_days window as the query above.
    -- warehouse_id is a globally-unique GUID, so we key on it alone (no workspace_id).
    SELECT
        u.usage_metadata.warehouse_id                    AS warehouse_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.warehouse_id IS NOT NULL
      AND u.usage_date >= current_date() - INTERVAL :period_days DAYS
      AND u.usage_date <  current_date()
    GROUP BY u.usage_metadata.warehouse_id
)
SELECT
    g.warehouse_id,
    COUNT(*)                                                                       AS event_count,
    -- seconds the warehouse held RUNNING before the next event (the auto-stop idle tail lives here)
    SUM(CASE WHEN g.event_type = 'RUNNING'  AND g.gap_seconds IS NOT NULL THEN g.gap_seconds ELSE 0 END) AS running_seconds,
    -- seconds spent STARTING (cold-start tax) and STOPPED (suspended, not billing compute)
    SUM(CASE WHEN g.event_type = 'STARTING' AND g.gap_seconds IS NOT NULL THEN g.gap_seconds ELSE 0 END) AS starting_seconds,
    SUM(CASE WHEN g.event_type = 'STOPPED'  AND g.gap_seconds IS NOT NULL THEN g.gap_seconds ELSE 0 END) AS stopped_seconds,
    -- longest single RUNNING gap = the worst observed idle tail before a stop/next event
    MAX(CASE WHEN g.event_type = 'RUNNING'  AND g.gap_seconds IS NOT NULL THEN g.gap_seconds END)         AS max_running_gap_seconds,
    SUM(CASE WHEN g.event_type = 'RUNNING'  THEN 1 ELSE 0 END)                        AS running_events,
    SUM(CASE WHEN g.event_type = 'STARTING' THEN 1 ELSE 0 END)                        AS starting_events,
    SUM(CASE WHEN g.event_type = 'STOPPED'  THEN 1 ELSE 0 END)                        AS stopped_events,
    -- ADDED cost visibility (see header caveats): exact billed DBUs and list-price $ ESTIMATE for this warehouse
    COALESCE(cr.net_dbus, 0)     AS net_dbus,
    COALESCE(cr.est_usd_list, 0) AS est_usd_list,
    -- status: worst-first band on the single longest continuous RUNNING stretch (field heuristic; :warn_idle_hours / :crit_idle_hours).
    CASE
        WHEN MAX(CASE WHEN g.event_type = 'RUNNING' AND g.gap_seconds IS NOT NULL THEN g.gap_seconds END) IS NULL THEN 'NOT_ASSESSED'
        WHEN MAX(CASE WHEN g.event_type = 'RUNNING' AND g.gap_seconds IS NOT NULL THEN g.gap_seconds END) >= :crit_idle_hours * 3600 THEN 'CRITICAL'
        WHEN MAX(CASE WHEN g.event_type = 'RUNNING' AND g.gap_seconds IS NOT NULL THEN g.gap_seconds END) >= :warn_idle_hours * 3600 THEN 'WARN'
        ELSE 'OK'
    END AS status
FROM gaps g
LEFT JOIN cost_rollup cr
  ON g.warehouse_id = cr.warehouse_id
GROUP BY g.warehouse_id, cr.net_dbus, cr.est_usd_list
ORDER BY CASE status WHEN 'CRITICAL' THEN 0 WHEN 'WARN' THEN 1 WHEN 'OK' THEN 2 ELSE 3 END, max_running_gap_seconds DESC

SELECT cluster_id,
       CASE WHEN cluster_name IS NULL THEN cluster_name ELSE concat(substr(cluster_name, 1, 2), '****') END AS cluster_name,
       CASE
         WHEN owned_by IS NULL OR owned_by = '__REDACTED__' THEN owned_by
         WHEN owned_by LIKE '%@%' THEN concat(substr(owned_by, 1, 2), '****@****')
         WHEN owned_by RLIKE '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN owned_by
         ELSE concat(substr(owned_by, 1, 2), '****')
       END AS owned_by,
       driver_node_type, worker_node_type, worker_count,
       min_autoscale_workers, max_autoscale_workers, auto_termination_minutes, enable_elastic_disk,
       cluster_source, dbr_version, data_security_mode, policy_id, driver_instance_pool_id,
       worker_instance_pool_id, tags, init_scripts, aws_attributes, azure_attributes, gcp_attributes,
       create_time, delete_time, change_time, workspace_id, account_id
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY cluster_id ORDER BY change_time DESC) AS rn
  FROM system.compute.clusters
)
WHERE rn = 1 AND delete_time IS NULL
ORDER BY cluster_id

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 (
    -- Pre-aggregated per cluster_id (globally-unique GUID -> workspace_id not needed for the join).
    SELECT 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, -LEAST(:period_days, 90), current_date())
      AND u.usage_date <  current_date()
    GROUP BY u.usage_metadata.cluster_id
),
finding AS (
    SELECT
        cluster_id,
        -- lexicographically-largest node_type label (representative only, not load-weighted)
        MAX(node_type)                                                       AS node_type,
        COUNT(*)                                                             AS total_slices,
        -- idle = CPU busy (user+system) below :idle_cpu_pct on that minute-slice; over ALL slices.
        SUM(CASE WHEN (cpu_user_percent + cpu_system_percent) < :idle_cpu_pct THEN 1 ELSE 0 END) AS idle_slices,
        COUNT(DISTINCT CASE WHEN driver THEN NULL ELSE node_type END)        AS worker_node_type_variants,
        -- utilization averaged over ALL slices (NOT only idle rows) so the mean is honest.
        AVG(cpu_user_percent + cpu_system_percent)                          AS avg_cpu_pct_all,
        MAX(cpu_user_percent + cpu_system_percent)                          AS peak_cpu_pct_all,
        AVG(mem_used_percent)                                               AS avg_mem_pct_all,
        MAX(mem_used_percent)                                               AS peak_mem_pct_all,
        MIN(start_time)                                                     AS first_slice,
        MAX(end_time)                                                       AS last_slice
    FROM system.compute.node_timeline
    WHERE start_time >= dateadd(DAY, -LEAST(:period_days, 90), current_date())
    GROUP BY cluster_id
)
SELECT
    f.cluster_id,
    f.node_type,
    f.total_slices,
    f.idle_slices,
    -- idle_ratio = idle minute-slices / total minute-slices (a fraction 0-1, never "hours").
    f.idle_slices / NULLIF(f.total_slices, 0)                            AS idle_ratio,
    f.worker_node_type_variants,
    f.avg_cpu_pct_all,
    f.peak_cpu_pct_all,
    f.avg_mem_pct_all,
    f.peak_mem_pct_all,
    f.first_slice,
    f.last_slice,
    COALESCE(cr.net_dbus, 0)     AS net_dbus,
    COALESCE(cr.est_usd_list, 0) AS est_usd_list,
    -- directional wasted-$ overlay: list-price cost scaled by the idle fraction.
    COALESCE(cr.est_usd_list, 0) * (f.idle_slices / NULLIF(f.total_slices, 0)) AS est_wasted_usd_list,
    -- status: worst-first band on idle_ratio (field heuristic). Too few slices -> NOT_ASSESSED.
    CASE
      WHEN f.total_slices < :min_slices                                        THEN 'NOT_ASSESSED'
      WHEN f.idle_slices / NULLIF(f.total_slices, 0) >= :crit_idle_ratio       THEN 'CRITICAL'
      WHEN f.idle_slices / NULLIF(f.total_slices, 0) >= :warn_idle_ratio       THEN 'WARN'
      ELSE 'OK'
    END AS status
FROM finding f
LEFT JOIN cost_rollup cr
  ON f.cluster_id = cr.cluster_id
ORDER BY est_wasted_usd_list DESC
LIMIT :top_n

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.usage_metadata.warehouse_id                     AS warehouse_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.warehouse_id IS NOT NULL
      AND u.usage_date >= current_date() - INTERVAL :period_days DAYS
      AND u.usage_date <  current_date()
    GROUP BY u.usage_metadata.warehouse_id
)
SELECT
    we.warehouse_id,
    SUM(CASE WHEN event_type = 'SCALED_UP'   THEN 1 ELSE 0 END)        AS scaled_up_events,
    SUM(CASE WHEN event_type = 'SCALED_DOWN' THEN 1 ELSE 0 END)        AS scaled_down_events,
    SUM(CASE WHEN event_type IN ('SCALED_UP','SCALED_DOWN') THEN 1 ELSE 0 END) AS scaling_events,
    COUNT(*)                                                          AS total_events,
    MIN(event_time)                                                   AS first_event_time,
    MAX(event_time)                                                   AS last_event_time,
    -- observed span in hours (first..last event in-window); the driver rate below is scaling_events / this.
    (unix_timestamp(MAX(event_time)) - unix_timestamp(MIN(event_time))) / 3600.0 AS observed_hours,
    MAX(cluster_count)                                                AS max_cluster_count,
    AVG(cluster_count)                                                AS avg_cluster_count,
    COALESCE(cr.net_dbus, 0)     AS net_dbus,
    COALESCE(cr.est_usd_list, 0) AS est_usd_list,
    -- status: worst-first band on scaling events per observed hour (field heuristic; :warn_churn_per_hour / :crit_churn_per_hour).
    CASE
        WHEN (unix_timestamp(MAX(event_time)) - unix_timestamp(MIN(event_time))) / 3600.0 < :min_observed_hours THEN 'NOT_ASSESSED'
        WHEN SUM(CASE WHEN event_type IN ('SCALED_UP','SCALED_DOWN') THEN 1 ELSE 0 END)
             / ((unix_timestamp(MAX(event_time)) - unix_timestamp(MIN(event_time))) / 3600.0) >= :crit_churn_per_hour THEN 'CRITICAL'
        WHEN SUM(CASE WHEN event_type IN ('SCALED_UP','SCALED_DOWN') THEN 1 ELSE 0 END)
             / ((unix_timestamp(MAX(event_time)) - unix_timestamp(MIN(event_time))) / 3600.0) >= :warn_churn_per_hour THEN 'WARN'
        ELSE 'OK'
    END AS status
FROM system.compute.warehouse_events we
LEFT JOIN cost_rollup cr
    ON we.warehouse_id = cr.warehouse_id
WHERE event_time >= current_timestamp() - INTERVAL :period_days DAYS
GROUP BY we.warehouse_id, cr.net_dbus, cr.est_usd_list
ORDER BY CASE status WHEN 'CRITICAL' THEN 0 WHEN 'WARN' THEN 1 WHEN 'OK' THEN 2 ELSE 3 END, scaling_events DESC

SELECT workspace_id, node_type, availability_type, state, event_type,
       COUNT(*) AS event_count,
       COUNT(DISTINCT instance_id) AS instance_count,
       MIN(event_time) AS first_event_time, MAX(event_time) AS last_event_time
FROM system.compute.instance_events
WHERE event_time >= current_timestamp() - INTERVAL :period_days DAYS
GROUP BY workspace_id, node_type, availability_type, state, event_type
ORDER BY event_count DESC, instance_count 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.instance_pool_id AS instance_pool_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.instance_pool_id IS NOT NULL
    AND u.usage_date >= date_add(current_date(), -CAST(:period_days AS INT))
    AND u.usage_date <  current_date()
  GROUP BY u.workspace_id, u.usage_metadata.instance_pool_id
)
SELECT p.instance_pool_id,
       CASE WHEN p.instance_pool_name IS NULL THEN p.instance_pool_name ELSE concat(substr(p.instance_pool_name, 1, 2), '****') END AS instance_pool_name,
       p.node_type, p.min_idle_instances, p.max_capacity,
       p.idle_instance_autotermination_minutes, p.enable_elastic_disk, p.preloaded_spark_version,
       p.preloaded_docker_images, p.tags, p.aws_attributes, p.azure_attributes, p.gcp_attributes, p.disk_spec,
       p.create_time, p.delete_time, p.change_time, p.workspace_id, p.account_id,
       COALESCE(cr.net_dbus, 0)     AS net_dbus,
       COALESCE(cr.est_usd_list, 0) AS est_usd_list,
       -- status: worst-first band on the standing idle-instance floor (field heuristic; :warn_min_idle_instances / :crit_min_idle_instances).
       CASE
         WHEN p.min_idle_instances IS NULL THEN 'NOT_ASSESSED'
         WHEN p.min_idle_instances >= :crit_min_idle_instances THEN 'CRITICAL'
         WHEN p.min_idle_instances >= :warn_min_idle_instances THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY instance_pool_id ORDER BY change_time DESC) AS rn
  FROM system.compute.instance_pools
) p
LEFT JOIN cost_rollup cr
  ON cr.workspace_id = p.workspace_id
 AND cr.instance_pool_id = p.instance_pool_id
WHERE p.rn = 1 AND p.delete_time IS NULL
ORDER BY p.min_idle_instances DESC

SELECT cluster_id, node_type, driver,
       COUNT(*) AS minute_rows,
       MIN(start_time) AS first_minute, MAX(end_time) AS last_minute,
       AVG(cpu_user_percent + cpu_system_percent) AS avg_cpu_pct,
       MAX(cpu_user_percent + cpu_system_percent) AS peak_cpu_pct,
       AVG(mem_used_percent) AS avg_mem_pct, MAX(mem_used_percent) AS peak_mem_pct,
       AVG(cpu_wait_percent) AS avg_cpu_wait_pct,
       SUM(network_sent_bytes)     AS total_network_sent_bytes,
       SUM(network_received_bytes) AS total_network_received_bytes,
       -- status: oversized-candidate band (field heuristic). Too few slices -> NOT_ASSESSED.
       CASE
         WHEN COUNT(*) < :min_slices THEN 'NOT_ASSESSED'
         WHEN AVG(cpu_user_percent + cpu_system_percent) < :oversized_cpu_pct
          AND AVG(mem_used_percent) < :oversized_mem_pct THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.compute.node_timeline
WHERE start_time >= dateadd(DAY, -LEAST(:period_days, 90), current_date())
GROUP BY cluster_id, node_type, driver
ORDER BY avg_cpu_pct ASC
LIMIT :top_n

SELECT node_type, core_count, memory_mb, gpu_count, account_id
FROM system.compute.node_types
ORDER BY node_type

SELECT warehouse_id, CASE WHEN warehouse_name IS NULL THEN warehouse_name ELSE concat(substr(warehouse_name, 1, 2), '****') END AS warehouse_name, workspace_id, account_id, warehouse_type, warehouse_channel,
       warehouse_size, min_clusters, max_clusters, auto_stop_minutes, tags, change_time, delete_time
FROM (
  SELECT warehouse_id, warehouse_name, workspace_id, account_id, warehouse_type, warehouse_channel,
         warehouse_size, min_clusters, max_clusters, auto_stop_minutes, tags, change_time, delete_time,
         ROW_NUMBER() OVER (PARTITION BY warehouse_id ORDER BY change_time DESC) AS rn
  FROM system.compute.warehouses
)
WHERE rn = 1 AND delete_time IS NULL
ORDER BY warehouse_id

SELECT warehouse_id, event_type,
       COUNT(*)            AS event_count,
       MAX(event_time)     AS last_event_time,
       MIN(event_time)     AS first_event_time,
       MAX(cluster_count)  AS max_cluster_count,
       AVG(cluster_count)  AS avg_cluster_count,
       -- status: only meaningful on RUNNING/STARTING rows - days since last RUNNING/STARTING event (field heuristic; :warn_stale_days / :crit_stale_days).
       CASE
         WHEN event_type NOT IN ('RUNNING', 'STARTING') THEN 'NOT_ASSESSED'
         WHEN (unix_timestamp(current_timestamp()) - unix_timestamp(MAX(event_time))) / 86400.0 >= :crit_stale_days THEN 'CRITICAL'
         WHEN (unix_timestamp(current_timestamp()) - unix_timestamp(MAX(event_time))) / 86400.0 >= :warn_stale_days THEN 'WARN'
         ELSE 'OK'
       END AS status
FROM system.compute.warehouse_events
WHERE event_time >= current_timestamp() - INTERVAL :period_days DAYS
GROUP BY warehouse_id, event_type
ORDER BY CASE status WHEN 'CRITICAL' THEN 0 WHEN 'WARN' THEN 1 WHEN 'OK' THEN 2 ELSE 3 END, last_event_time ASC