Learning From Failure and Continuous Improvement Questions

WITH params AS (
  SELECT CURRENT_DATE AS today,
         CURRENT_DATE - INTERVAL '29' DAY AS start_date
),
-- daily counts per job in the 30-day window
job_daily AS (
  SELECT
    job_id,
    CAST(occurred_at AS DATE) AS dt,
    COUNT(*) AS daily_error_count
  FROM pipeline_errors p
  JOIN params ON TRUE
  WHERE CAST(occurred_at AS DATE) BETWEEN params.start_date AND params.today
  GROUP BY job_id, CAST(occurred_at AS DATE)
),
-- total error-hours per job in window (assuming 1 hour per error)
job_totals AS (
  SELECT
    job_id,
    SUM(daily_error_count) AS total_error_hours
  FROM job_daily
  GROUP BY job_id
),
-- rank jobs by contribution to total error-hours
job_ranked AS (
  SELECT
    job_id,
    total_error_hours,
    RANK() OVER (ORDER BY total_error_hours DESC) AS job_rank,
    100.0 * total_error_hours / SUM(total_error_hours) OVER () AS pct_of_total_error_hours
  FROM job_totals
),
-- pick top 3 jobs
top_jobs AS (
  SELECT job_id, total_error_hours, job_rank, pct_of_total_error_hours
  FROM job_ranked
  WHERE job_rank <= 3
),
-- generate date series (ANSI SQL recursive)
dates AS (
  SELECT start_date AS dt FROM params
  UNION ALL
  SELECT dt + INTERVAL '1' DAY FROM dates
  WHERE dt + INTERVAL '1' DAY <= (SELECT today FROM params)
)
-- final: cross join top jobs with dates and left join daily counts
SELECT
  d.dt AS date,                       -- date (day)
  t.job_id AS job_id,                 -- job id (top 3)
  COALESCE(jd.daily_error_count, 0) AS daily_error_count,  -- errors on that day
  t.total_error_hours AS total_error_hours_30d,            -- total for window
  ROUND(t.pct_of_total_error_hours, 2) AS pct_of_total_error_hours, -- pct
  t.job_rank AS job_rank
FROM dates d
CROSS JOIN top_jobs t
LEFT JOIN job_daily jd
  ON jd.job_id = t.job_id AND jd.dt = d.dt
ORDER BY d.dt, t.job_rank;

from collections import defaultdict, deque

def build_adj_map(rows):
    parents = defaultdict(list)
    for r in rows:
        parents[r['job_id']] = r.get('upstream_job_ids', []) or []
    return parents

def compute_ancestors(targets, parents):
    # BFS/DFS upward to collect ancestors per target
    anc = {}
    for t in targets:
        seen = set()
        q = deque(parents.get(t, []))
        while q:
            p = q.popleft()
            if p in seen: continue
            seen.add(p)
            for pp in parents.get(p, []):
                if pp not in seen: q.append(pp)
        anc[t] = seen
    return anc

def greedy_hitting_set(ancestor_sets):
    # ancestor_sets: dict target -> set(ancestors)
    uncovered = set(ancestor_sets.keys())
    # inverted: ancestor -> set of targets it covers
    cover = defaultdict(set)
    for t, s in ancestor_sets.items():
        for a in s:
            cover[a].add(t)
    result = set()
    while uncovered:
        # pick ancestor covering most uncovered targets
        best = max(cover.items(), key=lambda kv: len(kv[1] & uncovered))[0]
        covered = cover[best] & uncovered
        if not covered: break
        result.add(best)
        uncovered -= covered
    return result

# usage:
# rows = [{job_id, start_ts, end_ts, status, upstream_job_ids}, ...]
# filter rows to time window, find failed targets (status != 'success')
# parents = build_adj_map(rows)
# ancestor_sets = compute_ancestors(failed_targets, parents)
# S = greedy_hitting_set(ancestor_sets)