Database Internals and Storage Engines Questions

Requirements / goal: restore cluster (5 nodes) after corruption on 2 nodes, minimize data loss, verify integrity, and return service within 4 hours. Assumptions: cluster uses leader-follower replication with WAL/commit logs, backups (daily + incremental), monitoring/alerting and orchestration tools (K8s or cluster manager), ability to promote replicas.

1. Immediate detection & triage (0–10 min)

• Confirm alert source (monitoring, node errors, storage checksum failures).
• Capture scope: node IDs, shards/partitions affected, last healthy commit timestamps, leader/replica roles.
• Notify on-call, DB owner, SRE, Product lead; open incident ticket with priority and 4-hour SLA.

2. Isolation (10–20 min)

• Immediately isolate corrupted nodes (remove from load balancer, mark as read-only or cordon from cluster) to prevent further writes/read corruption.
• Quiesce any background compactions/maintenance on those nodes.

3. Assessment & decision (20–30 min)

• Determine corruption type (filesystem, block-level, logical row-level). Check checksums, file system logs.
• Identify most recent consistent replica(s) or backups for each affected shard. If majority replicas healthy, prefer replica-based recovery to minimize data loss.

4. Restore path selection

• If replicas hold up-to-date data (replication lag ≤ acceptable RPO): perform replica catch-up and re-synchronization.
• If replicas missing data or logical corruption present: restore affected shards from most recent snapshot + apply WAL/transaction logs to point-in-time up to last good commit.

5. Recovery actions (30–150 min)

• For each affected shard:
  • Promote a healthy follower to replace corrupted node’s role if leader was corrupted.
  • If restoring from snapshot:
    • Provision replacement node(s), attach clean storage.
    • Restore snapshot, then apply WALs/incremental logs to target timestamp.
  • If using replication repair:
    • Trigger full resync/repair procedure (e.g., anti-entropy / rsync / SSTable rebuild depending on DB).
• Start recovery in parallel for different shards to meet 4-hour window.

6. Consistency & integrity verification (concurrent with restores)

• Run checksums (MD5/SHA) over key SSTables/files and compare with healthy replicas.
• Run application-level consistency checks:
  • Row counts per partition, transaction sequence continuity, checksum of transaction histories.
  • Spot-check business invariants (e.g., account balances non-negative, sum of credits == sum of debits).
• Run replay of committed transactions from WAL into read-only staging cluster and verify top-N recent transactions match production indexes.

7. Client failover and service continuity

• Keep load balancer redirecting reads/writes away from isolated nodes.
• If leader(s) replaced, switch leader via controlled failover; ensure client drivers reconnect and use updated endpoint metadata.
• If reduced capacity risks SLA, scale up temporary read replicas or route read-heavy traffic to cache (Redis) where safe.

8. Validation & gradual return to full service (150–210 min)

• Smoke tests: synthetic transactions, end-to-end ingestion to verify writes land and are replicated.
• Run data reconciliation job comparing hot path recent transactions with restored data.
• Once checks pass, reintroduce nodes one-by-one into LB and cluster, monitor errors for 30–60 minutes.

9. Post-recovery steps (after service restored)

• Preserve corrupted node storage for forensic analysis; take full snapshots.
• Rotate credentials/access if needed.
• Perform root-cause analysis, produce incident report with RPO/RTO achieved, gaps, and concrete remediation (improve monitoring, more frequent snapshots, increase replication factor, automated failover runbooks).
• Update runbook with timings and commands used.

Roles & communications

• Incident commander: owns timeline and communication.
• DB owner / data engineer: directs technical restore steps.
• SRE: node provisioning, LB changes, monitoring.
• QA: run verification suites.

Key safeguards to minimize data loss

• Prefer replicas + WAL replay over bare snapshot restore.
• Always apply WALs to snapshot to reach latest consistent point.
• Validate with checksums and business invariants before marking data as trusted.

This plan focuses parallelization of restores, conservative verification, and controlled failover to meet the 4-hour SLA while ensuring integrity.

Situation/Goal: Many small Parquet files per partition are causing planner/IO overhead. We need a safe, low-downtime compaction plan, prevention controls, and atomic replacement + verification.

Plan (steps):

1. Choose strategy: batch compaction using Spark (coalesce/repartition + parquet), or use an ACID layer (Delta/Hudi/Iceberg) for built-in compaction + transactions. If possible, migrate critical tables to Delta/Iceberg for long-term safety.
2. Compaction job (safe, atomic pattern):
   • Read affected partition(s) as source (path or table snapshot).
   • Write compacted files into a staging path per partition (e.g. s3://bucket/table/_staging/date=YYYY-MM-DD/).
   • Validate file sizes and row counts in staging.
   • Atomically swap: move staging files into final partition path using an atomic rename (HDFS) or write a manifest and update table metadata (S3 requires manifest + update of metastore or use Glue/Deltalake transactions).
   • Delete old small files only after successful swap and verification.

Example Spark (PySpark) compact one partition to ~256MB files:

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

src = "s3://bucket/table/date=2025-11-21/"
staging = "s3://bucket/table/_staging/date=2025-11-21/"
df = spark.read.parquet(src)

# target ~256MB files: adjust num_partitions based on size
num_parts = max(1, int(df.rdd.mapPartitions(lambda it: [sum(1 for _ in it)]).sum() * 1 / 1000000)) 
df.repartition(num_parts).write.mode("overwrite").parquet(staging)

(For S3 use manifest + metastore update or use Delta writer: df.write.format("delta").option("overwriteSchema","true")...)

Verification:

• Compare row counts and checksums (parquet file-level or dataset-level) between source and staging.
• Check file sizes distribution (no tiny files).
• Run sample queries to verify data correctness and performance.

Atomic replacement approaches:

• HDFS: rename staging dir -> partition dir (atomic).
• S3: upload staged files, write a partition manifest file listing new files, then update metastore to point to new file list (or use Delta transactions).
• If using Hive metastore, update partition location in a single metastore transaction.

Safety and minimal downtime:

• Run compaction on a copy/snapshot; keep readers pointed at original until verification done.
• Use short maintenance window to switch metadata; or implement dual-read: readers read union of original + new until swap.
• Retain old files for N days / move to _archive; delete only after successful monitoring.

Prevention (long-term):

• Enforce writer-side batching: write larger files (coalesce before write), set rolling policies in stream jobs.
• Use partitioning strategy to avoid tiny partitions; shard by hash when necessary.
• Leverage file size thresholds via job-level checks; schedule regular incremental compaction (e.g., daily compactor).
• Adopt transactional table formats (Delta/Hudi/Iceberg) to make compaction and metadata updates atomic and query-safe.

Monitoring & Automation:

• Metrics: files-per-partition, median file size, planning time, query latency.
• Alert when small-files thresholds exceeded and auto-trigger compaction jobs.
• Keep audit logs and automated rollback procedures.

This plan ensures correctness via staging+verification, minimizes downtime by swapping metadata atomically, and prevents recurrence through writer controls and periodic compaction or transactional table adoption.

Approach: Use an LSM-like layout with immutable sorted SSTables on disk, a write-ahead log (WAL) for durability, and a small manifest (metadata file) that records SSTables and snapshots. Snapshots are represented as a manifest entry pointing to a logical sequence number (seqno) and a list of SSTables covering that seqno. Atomic writes use WAL + two-phase manifest update (write new manifest to temp + fsync + atomic rename). Range scans read SSTables visible at snapshot seqno without pausing writers.

On-disk layout:

• WAL: append-only records (key, value, seqno, tombstone)
• Memtable: in-memory sorted buffer flushes to SSTable when full
• SSTable: immutable sorted file with index and bloom filter; footer contains min/max key and max seqno
• MANIFEST: JSON/line-delimited file listing active SSTables and current global_seqno; each manifest version is written atomically via rename
• SNAPSHOT entries: manifest records like {snapshot_id, seqno, sstables[]}

Write path pseudocode:

def write(key, value):
    seqno = alloc_seqno()            # monotonic
    wal.append({seqno,key,value})
    memtable.put(key, value, seqno)
    if memtable.size() > threshold:
        sst = flush_memtable(memtable)  # creates SST with max_seqno
        add_sstable_atomically(sst)

Flush + manifest update (atomic):

def add_sstable_atomically(sst):
    # write SST to disk and fsync
    sst.write_to_disk(); fsync(sst.path)
    # build new manifest in memory
    new_manifest = current_manifest.clone()
    new_manifest.sstables.append({path: sst.path, min:key_min, max:key_max, max_seqno: sst.max_seqno})
    tmp = write_temp_manifest(new_manifest); fsync(tmp)
    os.rename(tmp, MANIFEST)  # atomic swap
    current_manifest = new_manifest
    delete_obsolete_sstables_async()

Snapshot creation and read:

• Create snapshot: capture current global_seqno and manifest version, write snapshot entry to manifest atomically.

def create_snapshot():
    snap_seq = current_global_seqno()
    snap_id = uuid()
    new_manifest = current_manifest.clone()
    new_manifest.snapshots.append({id:snap_id, seqno:snap_seq, manifest_version:current_version})
    write_tmp_then_rename(new_manifest)
    return snap_id

• Read snapshot / range scan: load manifest version for snapshot, determine SSTables with max_seqno <= snap_seq (or use per-SST tombstone semantics), then perform merge-sort across relevant SSTables + WAL/immutable memtables as needed:

def range_scan(snapshot_id, start, end):
    manifest = load_manifest_for_snapshot(snapshot_id)
    sstables = select_sstables(manifest, start, end, seqno<=manifest.snapshots[snapshot_id].seqno)
    iterators = [sst.iterator(start,end) for sst in sstables]
    return merge_sorted(iterators, choose_latest_by_seqno)

Key points / reasoning:

• WAL + memtable ensures atomic durability per write; seqno gives total order.
• Immutable SSTables make reads lock-free; manifest atomic rename publishes new state.
• Snapshots are point-in-time views by seqno; readers pick SSTables whose data max_seqno <= snapshot seqno or filter by seqno in SSTable blocks.
• Deletion: tombstones with seqno; tombstone visibility governed by seqno.
• Compaction runs in background producing new SSTables with aggregated seqno ranges; compaction updates manifest atomically; snapshots reference previous SSTables to avoid deletion until no snapshot uses them.

Edge cases:

• Crash between WAL append and memtable apply: replay WAL on startup.
• Manifest corruption: keep versioned manifests and validate checksums.
• Long-lived snapshots: prevent GC of SSTables referenced by snapshots (reference counting).

Situation: Large numbers of tombstones in our Cassandra cluster are causing CPU spikes and compaction storms, degrading read/write performance and risking outages.

Root-cause analysis (how I’d investigate)

• Query metrics: tombstone_scanned, compaction throughput, GC pause, CPU, read latency, pending compactions.
• Inspect schema and queries: wide partitions, frequent range deletes, high-TTL churn, time-series writes with upserts.
• Check compaction strategy and size-tiering vs leveled, memtable flush frequency, hinted handoffs, and repair history (is anti-entropy backlog creating deletes?).
• Sample SSTables with nodetool tablestats, sstabledump to confirm tombstone density and deletion timestamps.
• Examine GC and disk I/O: are compactions causing long GC pauses and backpressure?

Immediate mitigations (stabilize cluster now)

• Throttle compaction: nodetool setcompactionthroughput (lower), enable incremental backpressure.
• Temporarily increase read request limits and set tombstone_failure_threshold to avoid failing entire reads (careful).
• Reduce concurrent repairs/maintenance; pause nonessential jobs.
• Add capacity: bring up spare nodes or scale vertically to spread compaction load.
• Serve degraded reads: add query-level guards (limit paging, avoid wide partition scans) and increase read_request_timeout for client backoff.
• If a single table causes the storm, take it offline for controlled repair/compaction on a subset of nodes.

Long-term schema & TTL/compaction changes

• Avoid high tombstone patterns:
  • Replace frequent deletes with immutability patterns (time-bucketed tables) or write tombstone-less TTLed inserts where possible.
  • Avoid wide partitions; shard by time/window to bound partition size.
• TTL strategy:
  • Use TTL for truly ephemeral data; choose TTLs > gc_grace_seconds only if safe.
  • Align application delete semantics with gc_grace_seconds to avoid resurrected data after repair.
• Compaction:
  • For time-series with high churn, use TimeWindowCompactionStrategy (TWCS) with appropriate window size to compact similar-lifetime SSTables together and reduce tombstone spread.
  • For read-heavy hot partitions, consider LeveledCompactionStrategy (LCS).
• gc_grace_seconds:
  • Do not reduce globally; if reducing per-table, ensure consistent backups and that no nodes are down and repair windows are controlled.
• Secondary indexes/materialized views:
  • Avoid for high-churn columns; they bloat tombstones and compaction work.

Recovery plan (restore healthy performance without unacceptable data loss)

1. Controlled assessment: identify affected tables/partitions and calculate tombstone ratios; prioritize by business impact.
2. Targeted repairs/compactions:
   • For each affected table use nodetool repair with incremental repairs disabled if necessary, or run targeted nodetool scrub/sstablescrub only after snapshotting.
   • Run decommission/replace for nodes with backlog if needed, ensuring no data loss via stream throttling.
3. Tombstone tombstone cleanup:
   • After ensuring all replicas have seen deletes (repairs complete), run major compactions carefully (one node at a time, low throughput) to purge tombstones; prefer TWCS to avoid cross-window tombstone retention.
4. Schema migrations:
   • Migrate hot/wide partitions to time-bucketed tables or materialize pre-aggregated views. Use rolling deploys to avoid large write surges.
5. Validation:
   • Monitor read latency, compaction stats, tombstone_scanned reduction, and GC metrics. Run load tests and compare to baseline.
6. Prevent recurrence:
   • Add monitoring alerts for tombstone_scanned per read, tombstones per partition, compaction backlog; enforce schema guidelines in PR reviews; document TTL and repair policies.
7. If unacceptable data loss risk exists:
   • Prefer application-level replay or retention of raw events in durable append-only store (S3 / Kafka) before irreversible compaction actions.
   • Use snapshots before destructive ops; if needed, restore specific SSTables to isolated cluster for selective data extraction.

Key trade-offs and cautions

• Lowering gc_grace_seconds or deleting tombstones without full repair can cause resurrected rows across nodes.
• Throttling compaction stabilizes CPU but prolongs tombstone presence.
• Major compactions are heavy; do node-by-node during low traffic windows.

This approach stabilizes the cluster fast, reduces future tombstone generation through schema/TTL/compaction design, and recovers performance with minimal acceptable risk by combining targeted, cautious operational actions with architectural fixes.

Approach: aggregate unique users per day (DAU), then compute a 7-day rolling average using a window over dates. Run nightly into a small summary table (dau_daily) to avoid scanning the raw table repeatedly.

SQL (compute and upsert nightly):

-- 1) Aggregate DAU from events of the date range we care about (example: last 30 days)
WITH daily AS (
  SELECT
    (event_time AT TIME ZONE 'UTC')::date AS day,
    COUNT(DISTINCT user_id) AS dau
  FROM events
  WHERE event_time >= (current_date - interval '30 days') -- restrict scan
    AND event_time < (current_date + interval '1 day')
    -- optionally filter event_type IN (...)
  GROUP BY 1
)
-- 2) Compute 7-day rolling average and upsert into a summary table
INSERT INTO dau_daily (day, dau, dau_7d_avg)
SELECT
  day,
  dau,
  ROUND(AVG(dau) OVER (ORDER BY day ROWS BETWEEN 6 PRECEDING AND CURRENT ROW)::numeric,2) AS dau_7d_avg
FROM daily
ON CONFLICT (day) DO UPDATE
  SET dau = EXCLUDED.dau,
      dau_7d_avg = EXCLUDED.dau_7d_avg;

Index & partitioning strategy (why and how):

• Partition the raw events table by range on event_time (daily or monthly partitions). For billions of rows, monthly partitions are often a good balance; daily partitions if retention/ingest patterns require.
  CREATE TABLE events (...) PARTITION BY RANGE (event_time);
  Create monthly partitions to allow pruning and parallel maintenance.
• Indexes:
  • On each partition, create a composite index on (event_time, user_id) or (user_id, event_time) depending on query patterns. For DAU aggregation where we filter by event_time and dedupe user_id, an index on (event_time, user_id) helps.
  • For very large partitions, use BRIN index on event_time to cheaply prune ranges; combine with a btree on (user_id) if many events per user.
  • Consider a partial index for active event_type values if you commonly filter by event_type.
• Other operational recommendations:
  • Maintain a nightly ETL job that scans only recent partitions (last N days) and writes to a compact dau_daily summary table; use that for dashboards.
  • Use parallel workers, set maintenance_work_mem appropriately, run ANALYZE after partition creation, and schedule partition attach/detach for retention (DROP old partitions).
  • For very high write volume, use COPY/bulk loads into current partition and avoid indexing hot inserts; build indexes after load or rely on BRIN for low-cost indexing.