The Materialization Storm

A hard AI Coding interview practice problem on DataDriven.

Domain
AI Coding
Difficulty
hard
Seniority
L6
Asked in
Anthropic

Problem

We added a new dataset to our Dagster repo last week. It declared 50 dynamic-partition assets that depend on shared upstream assets. Dagster scheduler CPU pegged at 100% within an hour and we're now seeing 12,000 materialization runs per hour for what should be ~50. Run tests/test_no_runaway_runs.py to reproduce. Find the cycle/fanout pattern in dagster_repo/assets.py and fix it so the partition fan-out is bounded and the run count is sane. Be ready to defend the trade-off your fix makes.

The graph started eating itself.

Worked solution and explanation

The ticket says cycle, but there is no cycle. The graph is a diamond built out of broadcasts. Any unpartitioned eager asset feeding a partitioned one is a broadcast: every time it materializes, all 50 tenant partitions get queued. user_profiles broadcasts into tenant_events and again into tenant_features, and tenant_events then forwards each of its partitions into tenant_features too. Every partition gets rebuilt once per path. Most candidates spot user_profiles and flip it to cron. The actual skill is reading how the scheduler decides what to queue. That decision belongs to the downstream asset's policy and partitioning, not the upstream's. Get that wrong and you ship a policy change that satisfies review while the storm keeps running.

Count the runs before you touch anything

Trace one cycle of simulate_run_count over 50 tenants. The two shared assets are 2 runs. tenant_events runs 150 times: 50 streaming ticks, 50 from the raw_events broadcast, and 50 from the user_profiles broadcast. tenant_features runs 200 times: 150 forwarded partition to partition, plus 50 more from the second user_profiles broadcast. tenant_recommendations inherits all 200. The total is 552, and every run past the first per partition is redundant work.

Reshape the edges

Step 1: Delete the phantom dependency on tenant_events

tenant_events filters raw_events() by tenant and never reads a profile. Its deps=[raw_events, user_profiles] is a lie that costs 50 runs per profile refresh. A declared dependency is a trigger, so declare only what the function actually reads.

Step 2: Collapse the profile broadcast into tenant_profiles

Add a partitioned projection that slices read_profiles() by context.partition_key. It absorbs the single broadcast from user_profiles. From there, tenant_features depends on [tenant_events, tenant_profiles], both with the same partitions_def, so each edge queues exactly the matching partition.

Step 3: Put user_profiles on its SLA

constraints.sql gives it 1440 minutes, and io.py says it refreshes at 04:00 UTC. AutoMaterializePolicy.cron('0 4 * * *') matches reality, so the one remaining broadcast fires once a day, not on every evaluation. Leave raw_events 'eager': its 5-minute SLA is real.

One broadcast per shared source, then partition-to-partition edges all the way down: 452 runs, linear in tenants.
  • raw_events (source)
  • user_profiles (source)
  • tenant_events (transform)
  • tenant_profiles (transform)
  • tenant_features (transform)
  • tenant_recommendations (consumer)
  • raw_events -> tenant_events
  • user_profiles -> tenant_profiles
  • tenant_events -> tenant_features
  • tenant_profiles -> tenant_features
  • tenant_features -> tenant_recommendations
The reshaped edges in `assets.py`
@asset(auto_materialize_policy=AutoMaterializePolicy.cron('0 4 * * *'))
def user_profiles():
    return read_profiles()


@asset(
    partitions_def=TENANTS,
    auto_materialize_policy=AutoMaterializePolicy.eager(),
    deps=[user_profiles],
)
def tenant_profiles(context):
    tenant = context.partition_key
    return [p for p in read_profiles() if p['tenant'] == tenant]


@asset(
    partitions_def=TENANTS,
    auto_materialize_policy=AutoMaterializePolicy.eager(),
    deps=[raw_events],
)
def tenant_events(context):
    tenant = context.partition_key
    return [e for e in raw_events() if e['tenant'] == tenant]


@asset(
    partitions_def=TENANTS,
    auto_materialize_policy=AutoMaterializePolicy.eager(),
    deps=[tenant_events, tenant_profiles],
)
def tenant_features(context):
    return {'tenant': context.partition_key, 'feature_count': 5}

tenant_recommendations is unchanged; profile data still reaches it through tenant_features.

Projection asset
tenant_profiles pays one 50-way broadcast per profile refresh. Features stay current within the 30-minute SLA. You add one asset to own, and its extra runs are cheap slices.
Join in recommendations
Drop user_profiles from tenant_features and add it to tenant_recommendations. That gives the fewest assets, but recs still take a broadcast, and features stop seeing profile data, which may be what the model needs.

Common follow-up questions