Bronze-to-DBT Sensor Partition Error - Investigation Report

Date: December 7, 2025
Status: 🔴 Critical Issue - Sensor Failing Due to Missing PartitionsDefinition

Executive Summary

The bronze_to_dbt_sensor is now running (after being enabled), but it’s failing when trying to trigger the dbt_transform_job. The error is clear: “Job has no PartitionsDefinition”. The sensor correctly detects bronze materializations and creates RunRequests with partition keys, but the job itself doesn’t have partitions configured.

Problem Statement

Error:

dagster_shared.check.functions.CheckError: Failure condition: Job has no PartitionsDefinition

What’s Happening:

1. ✅ Sensor detects bronze materialization for partition 996523e0-aee0-452e-8ac7-35a44f2e8812
2. ✅ Sensor creates RunRequest with partition_key=partition_key
3. ❌ Dagster rejects the RunRequest because dbt_transform_job has no PartitionsDefinition
4. ❌ Sensor fails and retries every 30 seconds

Root Cause

The Issue: The dbt_transform_job is defined without an explicit partitions_def, even though:

• The dbt assets (erp_dbt_assets) DO have partitions_def=erp_connection_partitions
• The sensor is trying to trigger the job with a partition key
• Dagster requires jobs to explicitly declare partitions if they’re going to receive partitioned RunRequests

Current Code (jobs.py line 80-85):

dbt_transform_job = define_asset_job(
    name="dbt_transform",
    selection=AssetSelection.tag("dbt_project", "erp_pipeline"),
    description="Run dbt transformations (bronze → silver → gold)",
    tags={"kind": "transform"},
)

Problem: Missing partitions_def=erp_connection_partitions

Sensor Code (sensors.py line 106-114):

run_requests.append(
    RunRequest(
        run_key=f"dbt_transform_{partition_key}_{new_last_event_id}",
        partition_key=partition_key,  # ← This requires job to have partitions_def
        tags={
            "trigger": "bronze_completion_sensor",
            "connection_id": partition_key,
        },
    )
)

Evidence

Logs Show Sensor Working But Failing

12:37:59 UTC:

INFO - bronze_to_dbt_sensor - Bronze materialization for partition 996523e0-aee0-452e-8ac7-35a44f2e8812: bronze/products
INFO - bronze_to_dbt_sensor - Triggering dbt transform for 1 partitions: 996523e0-aee0-452e-8ac7-35a44f2e8812
ERROR - Sensor tick caught an error
CheckError: Failure condition: Job has no PartitionsDefinition

12:38:29 UTC (retry):

INFO - bronze_to_dbt_sensor - Bronze materialization for partition 996523e0-aee0-452e-8ac7-35a44f2e8812: bronze/products
INFO - bronze_to_dbt_sensor - Triggering dbt transform for 1 partitions: 996523e0-aee0-452e-8ac7-35a44f2e8812
ERROR - Sensor tick caught an error
CheckError: Failure condition: Job has no PartitionsDefinition

Stack Trace Analysis

The error occurs in Dagster’s job definition validation:

File ".../job_definition.py", line 856, in _get_partitions_def
    check.failed("Job has no PartitionsDefinition")

This happens when Dagster tries to resolve the RunRequest’s partition_key against the job’s partitions definition, but the job doesn’t have one.

Solution

Fix: Add PartitionsDefinition to dbt_transform_job

File: apps/dagster/erp_pipeline/jobs.py

Current:

dbt_transform_job = define_asset_job(
    name="dbt_transform",
    selection=AssetSelection.tag("dbt_project", "erp_pipeline"),
    description="Run dbt transformations (bronze → silver → gold)",
    tags={"kind": "transform"},
)

Fixed:

from erp_pipeline.partitions import erp_connection_partitions

dbt_transform_job = define_asset_job(
    name="dbt_transform",
    selection=AssetSelection.tag("dbt_project", "erp_pipeline"),
    description="Run dbt transformations (bronze → silver → gold)",
    tags={"kind": "transform"},
    partitions_def=erp_connection_partitions,  # ADD THIS
)

Why This Works

1. Job Partitioning: The job now explicitly declares it supports partitions
2. Sensor Compatibility: RunRequests with partition_key will be accepted
3. Asset Compatibility: The dbt assets already have partitions_def=erp_connection_partitions, so they’re compatible
4. Per-Tenant Runs: Each partition (connection_id) will trigger its own dbt run

Additional Considerations

typesense_sync_job

The typesense_sync_job might have the same issue if dbt_to_typesense_sensor tries to trigger it with partitions. Let me check:

Current typesense_sync_job (line 88-93):

typesense_sync_job = define_asset_job(
    name="typesense_sync",
    selection=AssetSelection.groups("typesense"),
    description="Sync all Gold layer data to Typesense (partitioned by connection)",
    tags={"kind": "search_index"},
)

dbt_to_typesense_sensor (sensors.py line 186-195):

run_requests.append(
    RunRequest(
        run_key=f"typesense_sync_gold_trigger_{partition_key}_{new_last_event_id}",
        partition_key=partition_key,  # ← Also uses partition_key
        tags={...},
    )
)

Recommendation: Add partitions_def=erp_connection_partitions to typesense_sync_job as well to prevent the same error.

Testing Plan

After fixing:

1. Verify Sensor Evaluation:

   • Check logs for “Bronze materialization for partition X”
   • Verify no “Job has no PartitionsDefinition” errors

2. Verify dbt Job Runs:

   • Check Dagster UI → Runs → Should see dbt_transform runs
   • Verify runs have partition keys set
   • Verify dbt processes only the specified connection_id

3. Verify Gold Materializations:

   • Check if gold assets materialize after dbt runs
   • Verify dbt_to_typesense_sensor triggers (if typesense job is also fixed)

4. End-to-End Test:

   • Trigger bronze sync for one connection
   • Verify: Bronze → dbt → Gold → Typesense pipeline completes

Files to Modify

1. apps/dagster/erp_pipeline/jobs.py
   • Add import: from erp_pipeline.partitions import erp_connection_partitions
   • Add partitions_def=erp_connection_partitions to dbt_transform_job
   • Add partitions_def=erp_connection_partitions to typesense_sync_job (preventive)

Related Issues

Warning About Deprecated API

Logs also show:

UserWarning: Use fetch_materializations instead of get_event_records to fetch materialization events.

Location: sensors.py line 80-87

Current:

records = context.instance.get_event_records(
    EventRecordsFilter(
        event_type=DagsterEventType.ASSET_MATERIALIZATION,
        asset_key=asset_key,
        after_cursor=last_event_id,
    ),
    limit=100,
)

Recommended: Update to use fetch_materializations() API (but this is a warning, not blocking)

Summary

Root Cause: dbt_transform_job missing partitions_def declaration

Impact: Sensor detects events but can’t trigger job runs

Fix: Add partitions_def=erp_connection_partitions to job definition

Priority: 🔴 CRITICAL - Blocks entire downstream pipeline

---

Report Generated: 2025-12-07 12:40 UTC
Investigator: AI Assistant
Status: Ready for Fix Implementation