Incident Management and Response Questions

SELECT partition_name, dfs_location, create_time, last_modified
FROM hive_partitions
WHERE partition_name BETWEEN '2025-12-05 10' AND '2025-12-05 14';

SHOW CREATE TABLE db.table PARTITION (dt='2025-12-05', hour='10');

INSERT INTO control.dirty_partitions VALUES ('db.table','2025-12-05','10', CURRENT_TIMESTAMP);

df.write.mode("overwrite").option("path","/data/table/_tmp/dt=...").save()
fs.rename("/data/table/_tmp/dt=...", "/data/table/dt=...")

INSERT OVERWRITE TABLE db.table PARTITION (dt='2025-12-05', hour='10') SELECT ...;

SELECT '2025-12-05 10' as part, COUNT(*) as cnt, MIN(ts), MAX(ts)
FROM db.table WHERE dt='2025-12-05' AND hour='10';

from collections import deque
from typing import Dict, Deque, Tuple, Iterable
from datetime import datetime, timedelta

WINDOW = timedelta(minutes=5)

def fingerprint(alert: dict) -> str:
    """
    Build fingerprint from metric and sorted tags.
    alert expected keys: 'metric' (str), 'tags' (dict or list of key=val)
    """
    metric = alert['metric']
    tags = alert.get('tags', {})
    # Normalize tags: accept dict or list of pairs
    if isinstance(tags, dict):
        items = sorted((str(k), str(v)) for k, v in tags.items())
    else:
        items = sorted((str(k), str(v)) for k, v in tags)
    tags_str = "|".join(f"{k}={v}" for k, v in items)
    return f"{metric}#{tags_str}"

def dedupe_alerts(stream: Iterable[dict]):
    """
    Generator: yields only unique alerts within a 5-minute sliding window.
    Each alert dict needs a 'timestamp' (datetime or ISO string), 'metric', 'tags', 'value'.
    Memory bounded by number of unique fingerprints within last 5 minutes.
    """
    seen: Dict[str, datetime] = {}  # fingerprint -> last_seen_timestamp
    q: Deque[Tuple[datetime, str]] = deque()  # ordered by arrival timestamp

    for alert in stream:
        # normalize timestamp to datetime
        ts = alert['timestamp']
        if isinstance(ts, str):
            ts = datetime.fromisoformat(ts)
        elif not isinstance(ts, datetime):
            # assume unix seconds
            ts = datetime.fromtimestamp(float(ts))
        fp = fingerprint(alert)

        # Evict fingerprints older than window relative to current alert timestamp
        cutoff = ts - WINDOW
        while q and q[0][0] <= cutoff:
            old_ts, old_fp = q.popleft()
            # Only remove from seen if the stored last_seen matches the evicted timestamp
            # (protects against newer duplicates re-inserted later)
            if seen.get(old_fp) == old_ts:
                del seen[old_fp]

        last = seen.get(fp)
        if last is None:
            # not seen in window -> yield and record
            seen[fp] = ts
            q.append((ts, fp))
            yield alert
        else:
            # seen within window -> skip
            # Optionally: update last seen to newer timestamp if arriving later
            if ts > last:
                seen[fp] = ts
                q.append((ts, fp))
            # do not yield

-- OLTP: rows and checksum for orders in window
SELECT COUNT(*) AS cnt,
       SUM(amount) AS total_amount,
       MD5(string_agg(id || '|' || amount || '|' || status, ',' ORDER BY id)) AS checksum
FROM orders
WHERE created_at >= '2025-11-01' AND created_at < '2025-11-02';

SELECT COUNT(*) AS cnt,
       SUM(amount) AS total_amount,
       MD5(string_agg(id || '|' || amount || '|' || status, ',' ORDER BY id)) AS checksum
FROM raw_orders
WHERE event_date = '2025-11-01';

SELECT COUNT(*) AS cnt,
       SUM(amount) AS total_amount,
       MD5(string_agg(id || '|' || amount || '|' || status, ',' ORDER BY id)) AS checksum
FROM dwh_orders
WHERE order_date = '2025-11-01';

-- per-partition checksum
SELECT partition_key, COUNT(*) cnt, SUM(amount) total_amount,
       MD5(string_agg(id || '|' || amount, ',' ORDER BY id)) checksum
FROM ...
GROUP BY partition_key;

-- find missing IDs present in source but not in OLAP
WITH src AS (
  SELECT id FROM orders WHERE created_at >= '2025-11-01' AND created_at < '2025-11-02'
),
dwh AS (
  SELECT id FROM dwh_orders WHERE order_date = '2025-11-01'
)
SELECT id FROM src EXCEPT SELECT id FROM dwh;

import uuid
import time
from typing import List, Callable, Dict

class StorageInterface:
    # Implementations must provide atomic conditional write semantics.
    def get_state(self) -> Dict: ...
    def compare_and_set(self, key: str, expected: Dict, new: Dict) -> bool: ...
    def write_atomic(self, key: str, value: Dict) -> None: ...

class BackfillEngine:
    def __init__(self, partitions: List[str], process_partition: Callable[[str], None],
                 storage: StorageInterface, worker_id: str = None, lease_ttl: int = 60):
        self.partitions = partitions
        self.process = process_partition
        self.storage = storage
        self.worker_id = worker_id or str(uuid.uuid4())
        self.lease_ttl = lease_ttl
        # key in storage where per-partition metadata lives
        self.state_key = "backfill_state"

    def _init_state(self):
        state = self.storage.get_state() or {"parts": {p: {"status":"pending"} for p in self.partitions}}
        self.storage.write_atomic(self.state_key, state)

    def _claim_partition(self, pname: str):
        # loop: load, modify if pending or lease expired, CAS back
        while True:
            state = self.storage.get_state()
            entry = state["parts"].get(pname)
            now = int(time.time())
            if entry["status"] == "completed":
                return False
            if entry["status"] == "pending" or entry.get("lease_expires",0) < now:
                entry["status"] = "running"
                entry["owner"] = self.worker_id
                entry["lease_expires"] = now + self.lease_ttl
                if self.storage.compare_and_set(self.state_key, state, state):
                    return True
            else:
                return False

    def _mark_completed(self, pname: str):
        while True:
            state = self.storage.get_state()
            entry = state["parts"][pname]
            entry.update({"status":"completed", "owner": None, "lease_expires": None})
            if self.storage.compare_and_set(self.state_key, state, state):
                return

    def run_single(self, pname: str):
        # safe single partition execution with idempotency assumption in process
        if not self._claim_partition(pname):
            return
        try:
            # process should be idempotent: use upserts, dedup keys, or write-ahead markers
            self.process(pname)
            self._mark_completed(pname)
        except Exception:
            # release lease so others can retry
            state = self.storage.get_state()
            entry = state["parts"][pname]
            entry.update({"status":"pending", "owner": None, "lease_expires": None})
            self.storage.write_atomic(self.state_key, state)
            raise

    def run_all(self):
        self._init_state()
        for p in self.partitions:
            self.run_single(p)