Pipeline Chain Mislabeling Audit - Complete Report

Date: December 7, 2025
Status: 🔴 Multiple Issues Identified

Executive Summary

Comprehensive audit of the entire ERP pipeline chain (Bronze → dbt → Gold → Typesense) revealed 4 critical mislabeling issues that prevent proper asset selection, partition inference, and sensor triggering.

Pipeline Chain Overview

Bronze Assets (Python) → dbt Transform → Gold Assets (dbt) → Typesense Assets
     ↓                        ↓                    ↓                  ↓
["bronze", "products"]  ["gold", "dim_products"]  ["gold", "dim_products"]  ["typesense", "products"]
group_name="bronze"      (multi-asset, no group)  (multi-asset, no group)   group_name="typesense"

Issues Identified

🔴 Issue 1: `dbt_transform_job` - Invalid Asset Selection

Location: apps/dagster/erp_pipeline/jobs.py:82

Problem:

dbt_transform_job = define_asset_job(
    name="dbt_transform",
    selection=AssetSelection.tag("dbt_project", "erp_pipeline"),  # ← Finds 0 assets!
    ...
)

Root Cause:

dbt assets don’t have tag "dbt_project": "erp_pipeline"
dbt assets are a multi-asset with keys like ["gold", "dim_products"], ["silver", "stg_products"], etc.
No group_name attribute (multi-assets don’t support groups)
Job finds 0 assets → Can’t infer partitions → Sensor fails with “Job has no PartitionsDefinition”

Impact: ⚠️ CRITICAL - Blocks entire dbt transformation pipeline

Fix Required: Select dbt assets by explicit asset keys:

dbt_transform_job = define_asset_job(
    name="dbt_transform",
    selection=(
        # Gold layer
        AssetSelection.assets(AssetKey(["gold", "dim_products"]))
        | AssetSelection.assets(AssetKey(["gold", "dim_customers"]))
        | AssetSelection.assets(AssetKey(["gold", "dim_shipping_addresses"]))
        | AssetSelection.assets(AssetKey(["gold", "xref_products"]))
        # Silver layer
        | AssetSelection.assets(AssetKey(["silver", "stg_products"]))
        | AssetSelection.assets(AssetKey(["silver", "stg_customers"]))
        | AssetSelection.assets(AssetKey(["silver", "stg_cross_references"]))
        | AssetSelection.assets(AssetKey(["silver", "stg_shipping_addresses"]))
        | AssetSelection.assets(AssetKey(["silver", "stg_ship_to_links"]))
        # Bronze dbt layer (optional - for completeness)
        | AssetSelection.assets(AssetKey(["bronze", "raw_products"]))
        | AssetSelection.assets(AssetKey(["bronze", "raw_customers"]))
        | AssetSelection.assets(AssetKey(["bronze", "raw_cross_references"]))
        | AssetSelection.assets(AssetKey(["bronze", "raw_shipping_addresses"]))
        | AssetSelection.assets(AssetKey(["bronze", "raw_ship_to_links"]))
    ),
    description="Run dbt transformations (bronze → silver → gold)",
    tags={"kind": "transform"},
)

Alternative (if Dagster supports key prefix matching):

# If AssetSelection.keys() with prefix exists
selection=AssetSelection.keys("gold", "silver", "bronze").downstream()

🔴 Issue 2: `full_sync_job` - Invalid Asset Selection

Location: apps/dagster/erp_pipeline/jobs.py:22

Problem:

full_sync_job = define_asset_job(
    name="full_sync_pipeline",
    selection=AssetSelection.groups("bronze", "typesense")
    | AssetSelection.tag("dbt_project", "erp_pipeline"),  # ← Won't match dbt assets!
    ...
)

Root Cause:

Same issue as Issue 1: AssetSelection.tag("dbt_project", "erp_pipeline") doesn’t match dbt assets
Full sync job won’t include dbt transformations

Impact: ⚠️ HIGH - Full syncs won’t run dbt transformations

Fix Required: Replace tag selection with explicit asset keys (same as Issue 1):

full_sync_job = define_asset_job(
    name="full_sync_pipeline",
    selection=(
        AssetSelection.groups("bronze", "typesense")
        | AssetSelection.assets(AssetKey(["gold", "dim_products"]))
        | AssetSelection.assets(AssetKey(["gold", "dim_customers"]))
        | AssetSelection.assets(AssetKey(["gold", "dim_shipping_addresses"]))
        | AssetSelection.assets(AssetKey(["gold", "xref_products"]))
        | AssetSelection.assets(AssetKey(["silver", "stg_products"]))
        | AssetSelection.assets(AssetKey(["silver", "stg_customers"]))
        | AssetSelection.assets(AssetKey(["silver", "stg_cross_references"]))
        | AssetSelection.assets(AssetKey(["silver", "stg_shipping_addresses"]))
        | AssetSelection.assets(AssetKey(["silver", "stg_ship_to_links"]))
        # ... (all dbt assets)
    ),
    ...
)

🟡 Issue 3: `dbt_to_typesense_sensor` - Incorrect Partition Logic

Location: apps/dagster/erp_pipeline/sensors.py:169

Problem:

# dbt assets are not partitioned, so trigger for all partitions
# that have data. We'll get the current partitions.
partitions_to_sync.add("ALL")

Root Cause:

Comment says “dbt assets are not partitioned” but they ARE partitioned (partitions_def=erp_connection_partitions)
Sensor should extract partition key from materialization events, not trigger all partitions
This causes unnecessary Typesense syncs for all partitions even when only one changed

Impact: ⚠️ MEDIUM - Performance issue (unnecessary syncs), but functionality works

Fix Required: Extract partition key from materialization events (similar to bronze_to_dbt_sensor):

for record in records:
    if record.storage_id > new_last_event_id:
        new_last_event_id = record.storage_id

        # Get partition key from the run that materialized this asset
        run = context.instance.get_run_by_id(record.run_id)
        if run and "dagster/partition" in run.tags:
            partition_key = run.tags["dagster/partition"]
            partitions_to_sync.add(partition_key)
            context.log.info(
                f"Gold materialization for partition {partition_key}: "
                f"{asset_key.to_user_string()}"
            )

Note: This requires verifying that run.tags["dagster/partition"] is the correct way to extract partition from event records. May need to use record.dagster_event.partition if available.

🟡 Issue 4: `bronze_to_dbt_sensor` - Potential Partition Extraction Issue

Location: apps/dagster/erp_pipeline/sensors.py:96

Current Code:

run = context.instance.get_run_by_id(record.run_id)
if run and "dagster/partition" in run.tags:
    partition_key = run.tags["dagster/partition"]
    partitions_to_transform.add(partition_key)

Potential Issue:

Using run.tags["dagster/partition"] may not be the most reliable way
Should verify if record.dagster_event.partition or record.partition exists
May need to check Dagster API for proper partition extraction from event records

Impact: ⚠️ LOW - May work but could be fragile

Investigation Needed:

Check Dagster API for EventLogEntry.partition or DagsterEvent.partition
Verify if run.tags["dagster/partition"] is the recommended approach
Test with actual materialization events

Asset Key Reference

Bronze Python Assets (group_name=“bronze”)

["bronze", "products"]
["bronze", "customers"]
["bronze", "cross_references"]
["bronze", "shipping_addresses"]
["bronze", "ship_to_links"]

dbt Assets (multi-asset, no group_name)

Gold Layer:

["gold", "dim_products"]
["gold", "dim_customers"]
["gold", "dim_shipping_addresses"]
["gold", "xref_products"]

Silver Layer:

["silver", "stg_products"]
["silver", "stg_customers"]
["silver", "stg_cross_references"]
["silver", "stg_shipping_addresses"]
["silver", "stg_ship_to_links"]

Bronze dbt Layer:

["bronze", "raw_products"]
["bronze", "raw_customers"]
["bronze", "raw_cross_references"]
["bronze", "raw_shipping_addresses"]
["bronze", "raw_ship_to_links"]

Typesense Assets (group_name=“typesense”)

["typesense", "products"]
["typesense", "customers"]
["typesense", "cross_references"]
["typesense", "shipping_addresses"]

Sensor Asset Key References

`bronze_to_dbt_sensor` (sensors.py:40-46)

BRONZE_ASSET_KEYS = [
    AssetKey(["bronze", "products"]),
    AssetKey(["bronze", "customers"]),
    AssetKey(["bronze", "cross_references"]),
    AssetKey(["bronze", "shipping_addresses"]),
    AssetKey(["bronze", "ship_to_links"]),
]

✅ CORRECT - Matches Python bronze assets

`dbt_to_typesense_sensor` (sensors.py:49-54)

GOLD_ASSET_KEYS = [
    AssetKey(["gold", "dim_products"]),
    AssetKey(["gold", "dim_customers"]),
    AssetKey(["gold", "xref_products"]),
    AssetKey(["gold", "dim_shipping_addresses"]),
]

✅ CORRECT - Matches dbt gold assets

Summary of Required Fixes

Issue	Priority	File	Line	Fix Type
1. `dbt_transform_job` selection	🔴 CRITICAL	`jobs.py`	82	Replace tag selection with asset keys
2. `full_sync_job` selection	🔴 HIGH	`jobs.py`	22	Replace tag selection with asset keys
3. `dbt_to_typesense_sensor` partition logic	🟡 MEDIUM	`sensors.py`	169	Extract partition from events
4. `bronze_to_dbt_sensor` partition extraction	🟡 LOW	`sensors.py`	96	Verify/improve partition extraction

Testing Plan

After fixes:

Verify Job Asset Selection:

# Should show > 0 assets
len(dbt_transform_job.selection.resolve(all_assets))

Verify Partition Inference:

# Should have inferred partitions
dbt_transform_job.partitions_def  # Should not be None

Test Sensor Triggering:
- Trigger bronze sync → Should trigger dbt transform for correct partition
- Trigger dbt transform → Should trigger Typesense sync for correct partition only
Test Full Sync:
- Run full sync job → Should include all assets (bronze, dbt, typesense)

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

Pipeline Chain Mislabeling Audit - Complete Report

Executive Summary

Pipeline Chain Overview

Issues Identified

🔴 Issue 1: dbt_transform_job - Invalid Asset Selection

🔴 Issue 2: full_sync_job - Invalid Asset Selection

🟡 Issue 3: dbt_to_typesense_sensor - Incorrect Partition Logic

🟡 Issue 4: bronze_to_dbt_sensor - Potential Partition Extraction Issue

Asset Key Reference

Bronze Python Assets (group_name=“bronze”)

dbt Assets (multi-asset, no group_name)

Typesense Assets (group_name=“typesense”)

Sensor Asset Key References

bronze_to_dbt_sensor (sensors.py:40-46)

dbt_to_typesense_sensor (sensors.py:49-54)

Summary of Required Fixes

Testing Plan

🔴 Issue 1: `dbt_transform_job` - Invalid Asset Selection

🔴 Issue 2: `full_sync_job` - Invalid Asset Selection

🟡 Issue 3: `dbt_to_typesense_sensor` - Incorrect Partition Logic

🟡 Issue 4: `bronze_to_dbt_sensor` - Potential Partition Extraction Issue

`bronze_to_dbt_sensor` (sensors.py:40-46)

`dbt_to_typesense_sensor` (sensors.py:49-54)