Design Walkthrough 02 — Scaling a Metadata Service (Hive Metastore-Style)

Problem statement: A Hive-style metastore (databases → tables → partitions, schemas, stats, locations) backed by one RDBMS serves a company that grew from 10K to 5M partitions and from 50 to 5,000 queries/minute. getPartitions calls now take 30+ seconds, planner latency dominates short queries, and the DBA is paged weekly. Fix it without breaking the thousands of jobs using the Thrift API.

Table of Contents

Step 1: Why metastores melt

Name the actual pathologies before designing — they determine which fixes work:

  1. Partition explosion: partition-per-hour × years × thousands of tables ⇒ tens of millions of rows in PARTITIONS + per-partition key/value tables; ORM-generated joins over them are super-linear.
  2. Chatty planning: a single query plan can issue thousands of metastore calls (getTable, getPartitionsByFilter, per-partition stats). Latency multiplies.
  3. Listing as a primitive: getAllPartitions(table) returns megabytes; clients call it because the filtered API is historically unreliable — an API-shape problem, not a database problem.
  4. Write amplification on stats: per-partition stats updates after every load job.

Step 2: Constraints

  • API compatibility is the hard wall: thousands of jobs speak the Thrift API; clients embed old versions for years. Anything that changes call semantics needs versioned endpoints and long deprecation (Phase 11 discipline).
  • Reads ≫ writes (~100:1), and reads are partition-metadata-shaped.
  • Planner correctness tolerates bounded staleness for stats, but not for schema/location (wrong location = wrong data read).

Step 3: Alternatives

A. Scale the RDBMS (bigger box, read replicas, drop the ORM for hand-written SQL, proper indexes). Cheap, buys 5–10×, no compat risk. Rejected as the answer but accepted as step one — hand-written SQL via directSQL was exactly Hive's own move.

B. Cache layer in front (per-instance + distributed cache, TTL + invalidation). Buys read latency; the hard part is invalidation across multiple metastore instances — needs a change journal. Accepted as step two.

C. Shard the metastore by database/table hash. Solves table-count scale, but cross-shard ops (cross-db queries, global listing) get painful, and partition explosion within one table is unsolved. Deferred.

D. Change the data model — partition metadata out of the RDBMS into the table format itself (Iceberg/Delta-style manifests in object storage; the metastore keeps only the table pointer). Solves the root cause: planning reads manifests directly, scaling with the table not the service. This is the industry's actual answer, but it is a multi-year migration, not a fix. Accepted as the destination.

The staff-level move is sequencing A → B → D with compat preserved throughout, not picking one.

Step 4: Chosen design

Phase 1 — stop the bleeding (weeks): hand-written SQL paths for the hot calls; covering indexes for getPartitionsByFilter; server-side pagination + result-size caps on listing calls (new API variants, old ones kept but capped with loud deprecation warnings); connection pooling and statement caching audits.

Phase 2 — read path (months): stateless metastore instances behind LB; write-through journal table (monotonic change id per object); per-instance caches subscribe to the journal and invalidate; getPartitionsByFilter served from cache keyed by (table, filter-normal-form) with change-id validity. Stats reads get TTL staleness (planner-safe); schema/location reads validate change-id (correctness-critical).

Phase 3 — data model (quarters): new tables created as Iceberg; metastore stores the metadata pointer only; partition APIs for those tables are served by a shim that reads manifests — old clients keep calling getPartitionsByFilter and get correct answers without the RDBMS holding per-partition rows. Existing tables migrate by snapshot + catchup, heaviest first (top 1% of tables typically hold >50% of partitions).

Step 5: Consistency details

  • Journal as the ordering primitive: every metadata write commits the object change and journal row in one RDBMS transaction; cache validity = "my change-id for this object ≥ journal head for this object". Avoids distributed invalidation races without distributed transactions.
  • Read-your-writes for the writing client: client carries the change-id returned by its write; reads route with min_change_id so a stale cache either catches up or forwards to the source.
  • Concurrent DDL: optimistic — writes carry expected version, conflict ⇒ retry with rebase or fail to the user (matching Iceberg's optimistic commit protocol at the table layer).

Step 6: Migration plan

Compatibility-first sequencing, the part interviewers actually probe:

  1. Every new behavior behind a server-side flag, default off; canary instances first.
  2. Shadow-read validation: new path computes, old path serves, diffs logged (count and content hashes) until diff rate < 1e-6 for two weeks.
  3. API deprecations announced with metrics: instrument who still calls the unpaginated listing (user-agent/job tags), chase the top offenders, then cap.
  4. Rollback story at each step: caches can be bypassed by flag; Iceberg-pointer tables keep a frozen RDBMS snapshot of their last pre-migration state for emergency fallback (read-only).

Interviewer follow-ups

  1. Why not just put the metastore in a NoSQL store? — the workload is relational (filters over typed partition keys, transactional DDL); the problem is the data model keeping per-partition rows centrally, not the engine. Moving engines re-platforms the pain.
  2. How do you size the cache and what's the eviction story? — partition metadata is highly skewed (hot tables); LRU with per-table weight caps; measure hit rate per call type, not globally.
  3. What breaks first at another 10×? — the journal table write rate and the single RDBMS as the write master ⇒ that is when sharding (C) returns, by database, with the journal per shard.
  4. How does this compare to what actually happened in the ecosystem? — Hive directSQL + AWS Glue-style hosted metastores + the Iceberg/Delta migration; HMS remains as a catalog-of-pointers (or is replaced by REST catalogs).

References