Problem Framing

Your real-time analytics dashboards are slow, but it's not always the raw query engine that's bottlenecking. Often, high-cardinality joins are fundamentally modeling mistakes that no amount of query tuning can fix at scale.

Problem Framing

Imagine you run an e-commerce platform and need to analyze user behavior in real-time. You have a clicks table (a fact table) logging every user interaction, and a users table (a dimension table) with rich profile data like user_id, age_group, region, device_type, and last_login_ip. The clicks table grows by millions of rows per minute. The users table has hundreds of millions of rows, with user_id being the high-cardinality key.

You want to answer questions like: "How many users from the APAC region using mobile devices clicked on product category X in the last 5 minutes?"

A naive SQL query would look something like this:

SELECT
    u.region,
    u.device_type,
    COUNT(c.click_id)
FROM
    clicks c
JOIN
    users u ON c.user_id = u.user_id
WHERE
    c.timestamp >= NOW() - INTERVAL '5 minutes'
    AND c.product_category = 'X'
GROUP BY
    u.region, u.device_type;

In a distributed analytical database (like Presto/Trino, Spark, ClickHouse), this query quickly becomes a nightmare. To perform the join, data for matching user_ids from both clicks and users tables needs to be co-located on the same node. For hundreds of millions of distinct user_ids, this means a massive data shuffle across your cluster. Nodes spend more time transferring data over the network than actually processing it, leading to query latencies of 30+ seconds, frequent timeouts, or even cluster instability. Even with perfect indexing, the sheer volume of data movement kills performance.

Core Concept

The fundamental problem isn't the "high-cardinality" itself, but the volume of data that needs to be shuffled across a distributed system to resolve a join on a high-cardinality key. If you're joining a fact table with billions of rows against a dimension table with hundreds of millions of rows, and both tables are large, the join key (e.g., user_id) drives immense data movement.

The solution is to denormalize relevant, low-cardinality attributes from the dimension table into the fact table at ingestion time. Instead of joining large tables at query time, you "pre-join" or enrich your event data as it flows into your analytical store. This pushes the computational cost from query execution to your data ingestion pipeline, which is typically designed for high throughput and can scale horizontally.

Here's how it works:

              ┌───────────────────────┐                                 ┌───────────────────────┐
              │ User Profiles DB      │                                 │ Clicks Event Stream   │
              │ (e.g., PostgreSQL)    │                                 │ (e.g., Kafka)         │
              │                       │                                 │                       │
              │ - user_id (PK)        │                                 │ - click_id            │
              │ - age_group           │                                 │ - user_id (FK)        │
              │ - region              │                                 │ - product_category    │
              │ - device_type         │                                 │ - timestamp           │
              └─────────┬─────────────┘                                 └─────────┬─────────────┘
                        │ Change Data Capture (CDC) or                   │
                        │ Batch Extract/Transform                        │
                        │                                                │
                        ▼                                                ▼
              ┌───────────────────────────────────────────────────────────────────────────┐
              │ STREAM PROCESSING / ETL PIPELINE (e.g., Flink, Spark Structured Streaming)│
              │ - Joins/Enriches click events with relevant user attributes from profiles │
              │ - Filters for low-cardinality attributes (age_group, region, device_type)│
              └────────────────────────┬──────────────────────────────────────────────────┘
                                       │
                                       ▼
              ┌───────────────────────────────────────────┐
              │ REAL-TIME ANALYTICS DB (Fact Table)       │
              │ (e.g., ClickHouse, Apache Druid, Parquet/Orc on S3) │
              │                                           │
              │ - click_id                                │
              │ - user_id (Denormalized)                  │
              │ - product_category                        │
              │ - timestamp                               │
              │ - age_group (DENORMALIZED FROM USERS)     │
              │ - region (DENORMALIZED FROM USERS)        │
              │ - device_type (DENORMALIZED FROM USERS)   │
              └───────────────────────────────────────────┘

Now, your query becomes:

SELECT
    region,
    device_type,
    COUNT(click_id)
FROM
    clicks_enriched
WHERE
    timestamp >= NOW() - INTERVAL '5 minutes'
    AND product_category = 'X'
GROUP BY
    region, device_type;

This query no longer requires a JOIN, drastically reducing network I/O and CPU overhead in your analytics database. It performs a simple scan and aggregation on a single, albeit wider, table.

Real-World Application

Uber faces this problem constantly with its massive ride-hailing and delivery platforms. Consider their analytics for ride matching or driver performance. They have granular event data (pickup/dropoff events, driver GPS updates, rating events) which needs to be analyzed alongside high-cardinality dimension data like driver_id and rider_id, each with associated profiles (driver ratings, vehicle type, rider preferences).

Uber leverages a robust stream processing architecture, often built on Apache Kafka, Flink, and Spark, to handle this. As ride events flow through Kafka, they are enriched in real-time. For example, a "ride completed" event might be joined with the driver's current driver_rating, vehicle_type, and the rider's loyalty_tier before being written to their analytical data lake (e.g., Apache Hudi or Iceberg tables on S3, queried by Presto/Trino or Spark).

This denormalization at ingest time allows real-time dashboards to query billions of enriched events without performing expensive joins against driver or rider profile tables containing 100M+ entries. A query that previously might have shuffled petabytes of data for a join and taken minutes, now becomes a simple scan and aggregation over an already flattened table, completing in milliseconds. This enables critical business operations like dynamic pricing adjustments, fraud detection, and operational monitoring.

Common Mistakes

1. Over-denormalizing everything: The most common mistake is to copy all columns from a dimension table into the fact table. This bloats storage, makes schema changes cumbersome, and often copies data that is rarely used for analytical queries. Only denormalize the specific, low-cardinality attributes that are frequently used for filtering or grouping. If you need a rarely used, high-cardinality attribute (e.g., last_login_ip), keep it in the dimension table and perform an occasional, slower join or use a separate lookup service.
2. Ignoring eventual consistency: When you denormalize data, you're making a copy. If the source dimension table (e.g., users.region) changes, the denormalized region in your clicks_enriched table will be stale until the event is reprocessed or a new enrichment cycle runs. For real-time analytics, this latency is often acceptable (e.g., a few minutes), but it's a critical trade-off to acknowledge. If strict real-time consistency is paramount, you might need a different strategy (e.g., pre-computing aggregates and storing them in a key-value store, or using a materialized view with refresh guarantees).
3. Trying to optimize traditional relational databases for this problem: While indexes help, scaling traditional OLTP relational databases (like PostgreSQL, MySQL) to handle high-cardinality joins on billions of rows for real-time analytics is fighting an uphill battle. These systems are optimized for transactional integrity, not massive analytical scans and shuffles. The architectural shift to stream processing and OLAP-optimized stores is necessary for true scale.

Interview Angle

Interviewers often probe your understanding of distributed systems and data modeling when discussing performance.

Common Follow-up Questions:

• "How would you handle a user demographics table with billions of rows needing to be joined with a real-time event stream of user actions?"
• "What are the trade-offs of your proposed solution?"
• "What if the user's demographic data changes frequently? How do you ensure the analytics are up-to-date?"

Strong Answers:

1. Identify the root cause: Start by explaining that a direct join on user_id across billions of records in a distributed analytics system will lead to prohibitive data shuffling and network I/O. The problem is not simply slow queries, but an architectural mismatch.
2. Propose denormalization/pre-aggregation: Suggest enriching the event stream with relevant, low-cardinality user attributes (e.g., region, age_group, device_type) at ingestion time. Mention specific tools like Kafka Streams, Flink, or Spark Structured Streaming for this enrichment step.
3. Discuss trade-offs:
   • Pros: Significantly faster query performance for analytical queries, reduced load on the analytics database, simpler query logic.
   • Cons: Increased storage footprint (wider fact tables), potential for data staleness (eventual consistency), increased complexity in the ingestion pipeline, more challenging schema evolution.
4. Address data freshness:
   • CDC (Change Data Capture): For slowly changing dimensions, use CDC from the source users table to update a lookup service (e.g., Redis, a key-value store) or re-process historical data.
   • Slowly Changing Dimensions (SCD Type 2): If historical accuracy is needed (e.g., "what region was the user in when this event happened?"), implement SCD Type 2 logic in your dimension table and join to the correct version, or snapshot dimension attributes at the time of event.
   • Re-processing: For larger changes, you might need to re-process historical event streams with updated dimension data, if your data lake supports it (e.g., Apache Hudi's MERGE INTO or Apache Iceberg's table evolution).

Want to dive deeper into practical system design challenges? Let's connect for a 1:1 session. Book a slot on Topmate to discuss your specific engineering career goals.

---

Want to Go Deeper?

I do 1:1 sessions on system design, backend architecture, and interview prep.
If you're preparing for a Staff/Senior role or cracking FAANG rounds — book a session here.