# Dagster-dbt Integration Guide

This guide covers how to build and configure dbt models that integrate with Dagster at Suno. Our setup uses Dagster to orchestrate dbt materialization with sophisticated partitioning, automation conditions, and dependency management.

## Table of Contents

- [Project Structure](#project-structure)
- [Basic Model Configuration](#basic-model-configuration)
- [Referencing Sources and Models](#referencing-sources-and-models)
- [Asset Keys and Dagster Integration](#asset-keys-and-dagster-integration)
  - [Asset Kinds and Metadata Synchronization](#asset-kinds-and-metadata-synchronization)
- [Automation Conditions (Recommended)](#automation-conditions-recommended)
- [Legacy Scheduling (Deprecated)](#legacy-scheduling-deprecated---do-not-use)
- [Partitioning with partitions_def](#partitioning-with-partitions_def)
- [Backfill Policies](#backfill-policies)
- [Groups for Organization](#groups-for-organization)
- [Warehouse Configuration](#warehouse-configuration)
- [Custom Macros](#custom-macros)
- [Complete Model Example](#complete-model-example)
- [Model YAML Documentation](#model-yaml-documentation)
- [Environment-Specific Configuration](#environment-specific-configuration)
- [Best Practices](#best-practices)

## Project Structure

```
src/assets/dbt/analytics/
├── models/           # dbt models organized by layer
│   ├── raw/         # Raw data sources and staging
│   ├── staging/     # Cleaned and standardized data
│   ├── intermediate/# Business logic transformations (ephemeral)
│   └── marts/       # Final analytical tables
├── macros/          # Custom dbt macros
├── profiles.yml     # Snowflake connection configuration
├── dbt_project.yml  # Project configuration and scheduling tags
└── SCHEDULING_README.md  # Comprehensive scheduling guide
```

In `profiles.yml` we define three different dbt targets:

- `dev`: Intended for local development. Reads in a **path** to a snowflake private key, and outputs to each user's individual schema (e.g. `<user>_marts`).
- `stg`: Used in branch deployments. Reads **private key** from environment variables, and outputs to `stg_marts`.
- `prod`: Used in prod. Reads **private key** from environment variables, and outputs to `prod_marts`.

## Setup and Local Development

### dagster-dbt setup

1. In the Dagster setup section above, you should have set up your Snowflake account, assigned a keypair to your user, and granted the `DBT_DEV_ROLE` to your user.
1. Create or add to an .env file containing `DBT_USER` and `DBT_SNOWFLAKE_PRIVATE_KEY_PATH`. Example:

   ```sh
   DBT_USER=<snowflake_username>
   DBT_SNOWFLAKE_PRIVATE_KEY_PATH=<absolute/path/to/private/key>
   DBT_TARGET=dev
   ```

### Local development

There are two ways to run dbt locally: either via Dagster (UI) or directly through dbt (CLI).

- **Dagster (UI)**:
   ```sh
   cd suno-dagster
   uv run dagster dev
   ```

   The desired assets can then be materialized via the UI on the Dagster localhost.

- **dbt (CLI)**:
   ```sh
   cd suno-dagster/src/assets/dbt/analytics
   uv run --env-file <path/to/env/file> dbt run <config>
   ```

   `<config>` can be customized depending on the DBT asset you are materializing. For example, the following command:
   ```sh
   uv run --env-file <path/to/env/file> dbt run --select dim_anonymous_user_mapping+ --full-refresh
   ```
   would materialize `dim_anonymous_user_mapping` and all upstream assets (`+`) with a full refresh. See [dbt documentation](https://docs.getdbt.com/reference/commands/run) for more details.

### Branch/prod deployment

When a PR is submitted/updated, a branch deployment is kicked off via Github workflows. You can see these on the Suno Dagster [deployments page](https://suno.dagster.plus/prod/org-settings/deployments).

## Basic Model Configuration

### Model Header Configuration

Every dbt model starts with a `{{ config() }}` block that defines how it should be materialized:

```sql
{{
    config(
        materialized='incremental',
        incremental_strategy='delete+insert',
        unique_key=['user_id', 'platform', 'p_date'],
        snowflake_warehouse='FACT_HOOK_PLAY_LARGE',
        cluster_by=['p_date']
    )
}}

SELECT
    user_id,
    platform,
    SUM(play_count) AS total_plays,
    p_date
FROM {{ ref('upstream_model') }}
WHERE {{ partition_filter_daily('p_date') }}
GROUP BY user_id, platform, p_date
```

### Key Configuration Options

| Option | Values | Description |
|--------|--------|-------------|
| **materialized** | `view`, `table`, `incremental`, `ephemeral` | How the model should be materialized |
| **incremental_strategy** | `delete+insert`, `merge`, `insert_overwrite` | Strategy for incremental models |
| **unique_key** | Array of column names | Columns that define uniqueness for incremental models |
| **snowflake_warehouse** | Warehouse name | Snowflake warehouse to use for materialization |
| **cluster_by** | Array of column names | Columns to cluster the table on for performance |
| **on_schema_change** | `ignore`, `fail`, `append_new_columns`, `sync_all_columns` | How to handle schema changes |

## Referencing Sources and Models

### Source References

Define sources in `sources.yml` files:

```yaml
sources:
  - name: suno_prod
    database: suno_prod
    schema: prod
    tables:
      - name: fact_hook_play
        description: "Hook play events from production"
      - name: dim_user
        description: "User dimension table"
```

Reference sources in models:

```sql
SELECT *
FROM {{ source('suno_prod', 'fact_hook_play') }}
WHERE p_date >= '2023-01-01'
```

### Model References

Reference other dbt models using `ref()`:

```sql
SELECT
    u.user_id,
    u.subscription_tier,
    p.play_count
FROM {{ ref('stg_cleaned__dim_user') }} u
JOIN {{ ref('agg_user_hook_platform_daily') }} p
  ON u.user_id = p.user_id
```

## Asset Keys and Dagster Integration

### Setting Asset Keys for Sources

For external sources created in dbt that need custom asset keys in Dagster, use metadata:

```yaml
sources:
  - name: web
    database: "{{ 'suno_prod' if target.name == 'prod' else 'suno_staging' }}"
    schema: "{{ 'prod' if target.name == 'prod' else 'staging' }}"
    tables:
      - name: web_user_event
        description: "Web user events from web"
        tags: ['web', 'raw_event', 'frontend']
        meta:
          dagster:
            asset_key: ["web", "user_events"]
```

NOTE: Always use asset_key if the table is managed by Dagster directly! If not, an external source will be tried to be created with the same name and it will fail.

### External Source Configuration (Without asset_key)

For dbt sources that don't have a `meta.dagster.asset_key` defined, Dagster automatically creates external source assets. These assets are monitored by the `external_source_partition_availability_sensor` which checks for partition data availability in Snowflake.

NOTE: Use for assets materialized thru snowflake tasks.

#### Basic Configuration

External sources are automatically partitioned and monitored. By default, they use daily partitions unless specified otherwise:

```yaml
sources:
  - name: snowflake
    database: suno_prod
    schema: prod
    tables:
      - name: fact_subscription_period
        description: "Cleaned subscription period data"
        tags: ["verified", "subscription"]
        # No partition_type tag means default to daily partitions
```

#### Non-Partitioned Configuration

For tables that don't have partition columns (`p_date` or `p_hour`), use `partition_type=none`. These tables will not be monitored by the partition availability sensor and will not have partition definitions:

```yaml
sources:
  - name: snowflake
    database: suno_prod
    schema: prod
    tables:
      - name: dim_country
        description: "Cleaned country data"
        tags: ["verified", "partition_type=none"]

      - name: dim_subscription_plan
        description: "Cleaned subscription plan dimension table"
        tags: ["verified", "subscription", "partition_type=none"]
```

> **Note**: Non-partitioned sources (with `partition_type=none`) are excluded from the `external_source_partition_availability_sensor` and will not have partition definitions. Use this for dimension tables or tables that don't have date-based partitioning.

#### Daily Partition Configuration

```yaml
sources:
  - name: snowflake
    database: suno_prod
    schema: prod
    tables:
      - name: fact_subscription_period
        description: "Cleaned subscription period data"
        tags: ["verified", "subscription", "partition_type=daily"]
        meta:
          dagster:
            partition_start_date: "2023-03-26"
```

#### Hourly Partition Configuration

```yaml
sources:
  - name: snowflake
    database: suno_prod
    schema: prod
    tables:
      - name: bot_hourly
        description: "Bot hourly features"
        tags: ["bot", "partition_type=hourly"]
        meta:
          dagster:
            partition_start_date: "2023-03-26"
            partition_start_hour: 0  # Optional, defaults to 0
```

#### Custom Partition Column Names

If your table uses different column names for partitions (default is `p_date` and `p_hour`):

```yaml
sources:
  - name: snowflake
    database: suno_prod
    schema: prod
    tables:
      - name: custom_table
        description: "Table with custom partition columns"
        tags: ["partition_type=daily"]
        meta:
          dagster:
            partition_start_date: "2023-03-26"
            partition_date_column: "date_partition"  # Override default p_date
            partition_hour_column: "hour_partition"   # Override default p_hour (only used for hourly)
```

#### External Source Configuration Options

| Option | Type | Required | Default | Description |
|--------|------|----------|---------|-------------|
| `partition_type` | Tag | No | `daily` | Partition cadence: `daily`, `hourly`, or `none` |
| `partition_start_date` | Meta | No | DBT_MIN_PARTITION_DATE | First partition date (YYYY-MM-DD format) - only used when `partition_type` is `daily` or `hourly` |
| `partition_start_hour` | Meta | No | `0` | Starting hour for hourly partitions (0-23) - only used when `partition_type` is `hourly` |
| `partition_date_column` | Meta | No | `p_date` | Column name for date partitioning - only used when `partition_type` is `daily` or `hourly` |
| `partition_hour_column` | Meta | No | `p_hour` | Column name for hour partitioning - only used when `partition_type` is `hourly` |

> **Note**: When `partition_type=none`, partition-related metadata options are ignored and the source will not be monitored by the partition availability sensor.


#### How It Works

1. **Sensor Monitoring**: The `external_source_partition_availability_sensor` runs every 5 minutes
2. **Source Filtering**: Sources with `partition_type=none` are skipped by the sensor
3. **Partition Checking**: For each partitioned external source:
   - **Hourly**: Checks if data exists for the previous hour
   - **Daily**: Checks if data exists for the previous day
4. **Materialization**: When data is found for a partition, the sensor emits an `AssetMaterialization` event, marking that partition as available
5. **Downstream Dependencies**: This enables downstream assets with `automation_condition=daily_cron_with_eager_historical_backfill_condition` to detect when their external source dependencies are ready

#### Example: Complete External Source Setup

```yaml
sources:
  - name: snowflake
    database: suno_prod
    schema: prod
    config:
      group: snowflake
    tables:
      - name: fact_subscription_period
        description: "Cleaned subscription period data"
        tags: ["verified", "subscription", "partition_type=daily"]
        meta:
          dagster:
            partition_start_date: "2023-03-26"

      - name: bot_hourly
        description: "Bot hourly features, not all users appear"
        tags: ["bot", "partition_type=hourly"]
        meta:
          dagster:
            partition_start_date: "2023-03-26"
            partition_start_hour: 0

      - name: dim_country
        description: "Cleaned country data"
        tags: ["verified", "partition_type=none"]

      - name: dim_subscription_plan
        description: "Cleaned subscription plan dimension table"
        tags: ["verified", "subscription", "partition_type=none"]
```

### Asset Kinds and Metadata Synchronization

Dagster assets support **kinds** which are used for categorization and filtering. For dbt sources and exposures, kinds are automatically extracted from tags.

#### KIND Configuration

Kinds are extracted from tags using the format `kind=<value>`. Multiple kinds can be specified by adding multiple `kind=` tags.

**For External Sources:**
```yaml
sources:
  - name: snowflake
    database: suno_prod
    schema: prod
    tables:
      - name: fact_subscription_period
        description: "Cleaned subscription period data"
        tags: ["verified", "subscription", "partition_type=daily", "kind=fact_table"]
        # This will create an asset with kinds: {"dbt_source", "dbt", "fact_table"}
```

**For Exposures:**
```yaml
exposures:
  - name: orpheus_dashboard
    type: dashboard
    description: "A dashboard for viewing Orpheus metrics."
    tags:
      - "kind=hex"
      - "kind=dashboard"
    # This will create an asset with kinds: {"dbt", "hex", "dashboard"}
```

**Default Kinds:**
- **External Sources**: Always include `"dbt_source"` and `"dbt"` by default
- **Exposures**: Always include `"dbt"` by default
- Additional kinds are added from `kind=` tags

#### Group Name Synchronization

Group names organize assets in the Dagster UI. For external sources, group names are determined in this order:

1. `meta.dagster.group` (highest priority)
2. Source-level `config.group`
3. Default: `"dbt_sources"`

```yaml
sources:
  - name: snowflake
    database: suno_prod
    schema: prod
    config:
      group: snowflake  # Source-level group (applies to all tables unless overridden)
    tables:
      - name: fact_subscription_period
        description: "Cleaned subscription period data"
        tags: ["partition_type=daily"]
        # This will use group: "snowflake"

      - name: custom_table
        description: "Custom table with specific group"
        tags: ["partition_type=daily"]
        meta:
          dagster:
            group: custom_group  # Override source-level group
        # This will use group: "custom_group"
```

**For Exposures:**
- Exposures always use the group name: `"dbt_exposures"` (not configurable)

#### Dependency Mapping for Exposures

Exposures can declare dependencies on dbt models using `depends_on`. These are automatically mapped to Dagster asset keys:

- **Models starting with `stg_`**: Mapped to `['stg', model_name]`
- **Models starting with `int_`**: Mapped to `['int', model_name]`
- **Models starting with `raw_`**: Mapped to `['raw', model_name]`
- **Other models**: Mapped to `model_name` (as-is)

```yaml
exposures:
  - name: orpheus_dashboard
    depends_on:
      - ref("stg_orpheus__user_sessions")  # Maps to ['stg', 'stg_orpheus__user_sessions']
      - ref("int_user_metrics")            # Maps to ['int', 'int_user_metrics']
      - ref("raw_events")                  # Maps to ['raw', 'raw_events']
      - ref("dim_user")                    # Maps to 'dim_user'
```

#### Metadata Synchronization Summary

The following metadata is automatically synchronized from dbt to Dagster:

| Metadata Field | Source | External Sources | Exposures | Notes |
|----------------|--------|------------------|-----------|-------|
| **Description** | `description` | ✅ Yes | ✅ Yes | Direct mapping |
| **Kinds** | `tags` with `kind=` | ✅ Yes | ✅ Yes | Extracted from tags, defaults included |
| **Group Name** | `meta.dagster.group` or `config.group` | ✅ Yes | ❌ No | Always `dbt_exposures` for exposures |
| **Dependencies** | `depends_on` | ❌ No | ✅ Yes | Mapped to Dagster asset keys via `ref()` |
| **Partition Type** | `tags` with `partition_type=` | ✅ Yes | ❌ No | Only for external sources |
| **Asset Key** | `meta.dagster.asset_key` | ✅ Yes | ✅ Yes | Custom asset key override |
| **Partition Config** | `meta.dagster.*` | ✅ Yes | ❌ No | Only for partitioned sources |
| **Exposure Type** | `type` | ❌ No | ✅ Yes | Stored in metadata (dashboard, notebook, etc.) |
| **Owner Info** | `owner.name`, `owner.email` | ❌ No | ✅ Yes | Stored in metadata as `owner_name`, `owner_email` |
| **URL** | `url` | ❌ No | ✅ Yes | Stored in metadata if provided |
| **Database** | `database` | ✅ Yes | ❌ No | Stored in metadata for external sources |
| **Schema** | `schema` | ✅ Yes | ❌ No | Stored in metadata for external sources |
| **Source Name** | `source_name` | ✅ Yes | ❌ No | Automatically stored in metadata |
| **Source Table Name** | `name` | ✅ Yes | ❌ No | Automatically stored in metadata |
| **dbt Source Unique ID** | Auto-generated | ✅ Yes | ❌ No | Internal dbt identifier stored in metadata |

#### Complete Example: External Source with All Metadata

```yaml
sources:
  - name: snowflake
    database: suno_prod
    schema: prod
    config:
      group: snowflake  # Source-level default group
    tables:
      - name: fact_subscription_period
        description: "Cleaned subscription period data"
        tags:
          - "verified"
          - "subscription"
          - "partition_type=daily"
          - "kind=fact_table"
          - "kind=subscription_data"
        meta:
          dagster:
            partition_start_date: "2023-03-26"
            group: billing  # Override source-level group
        # Result in Dagster:
        # - Asset Key: ["snowflake", "fact_subscription_period"]
        # - Group: "billing"
        # - Kinds: {"dbt_source", "dbt", "fact_table", "subscription_data"}
        # - Description: "Cleaned subscription period data"
        # - Metadata:
        #     - source_name: "snowflake"
        #     - source_table_name: "fact_subscription_period"
        #     - database: "suno_prod"
        #     - schema: "prod"
        #     - partition_type: "daily"
        #     - partition_start_date: "2023-03-26"
        #     - partition_date_column: "p_date"
        #     - dbt_source_unique_id: "source.suno_prod.snowflake.fact_subscription_period"
```

#### Complete Example: Exposure with All Metadata

```yaml
exposures:
  - name: orpheus_dashboard
    type: dashboard
    owner:
      name: "Omar Sanchez"
      email: "omar@suno.com"
    description: "A dashboard for viewing Orpheus metrics."
    url: "https://app.hex.tech/883a3868-44c8-45f6-aef6-baa783bac922/app/Orpheus-the-myth-the-legend-031KzJbZyX2AZ7RRplqbPi/latest"
    tags:
      - "kind=hex"
      - "kind=dashboard"
    depends_on:
      - ref("stg_orpheus__user_sessions")
      - ref("stg_orpheus__tool_calls")
    # Result in Dagster:
    # - Asset Key: "orpheus_dashboard"
    # - Group: "dbt_exposures" (fixed)
    # - Kinds: {"dbt", "hex", "dashboard"}
    # - Description: "A dashboard for viewing Orpheus metrics."
    # - Dependencies:
    #     - ['stg', 'stg_orpheus__user_sessions']
    #     - ['stg', 'stg_orpheus__tool_calls']
    # - Metadata:
    #     - exposure_type: "dashboard"
    #     - owner_name: "Omar Sanchez"
    #     - owner_email: "omar@suno.com"
    #     - url: "https://app.hex.tech/..."
```

### Model Asset Keys

dbt models automatically get asset keys based on their location and name. To customize:

```sql
{{
    config(
        materialized='table',
        meta={
            'dagster': {
                'asset_key': ['custom', 'namespace', 'table_name']
            }
        }
    )
}}
```

## Automation Conditions (Recommended)

> **⚠️ IMPORTANT: Legacy cron-based scheduling is DISCOURAGED. Use automation conditions instead.**
>
> Automation conditions provide superior control, dependency management, and backfill capabilities compared to traditional scheduling approaches.

### Why Automation Conditions?

**Benefits over legacy scheduling:**
- **Smart dependency handling**: Automatically waits for upstream assets
- **Built-in backfill logic**: Eager historical backfill when partitions are missing
- **Better resource management**: Runs only when needed, not on rigid schedules
- **Easier debugging**: Clear visibility into why/when assets run
- **Flexible conditions**: Beyond just time-based triggers

### Configuring Automation Conditions

> **⚠️ IMPORTANT: Tag-based automation configuration is DEPRECATED.**
>
> Use `meta.dagster.automation_condition` instead of tags. Tag-based configuration will be removed in a future version.

Configure automation through `meta.dagster` in your model YAML files. All models **must** have automation conditions defined - models without conditions will not run.

### Model-Level Automation (Recommended)

```yaml
models:
  - name: table_name
    meta:
      dagster:
        automation_condition: daily_cron_with_eager_historical_backfill_condition
```

### Project-Level Automation (dbt_project.yml)

> **⚠️ DEPRECATED**: Project-level automation via tags is deprecated. Configure automation conditions at the model level using `meta.dagster.automation_condition` instead.

For project-wide defaults, you can still use tags in `dbt_project.yml`, but this approach is deprecated:

```yaml
# ⚠️ DEPRECATED - Use meta.dagster.automation_condition in model YAML instead
models:
  analytics:
    staging:
      backend_events:
        +tags: ["automation_condition=hourly_cron_with_eager_historical_backfill_condition"]
```

### Available Automation Conditions

Some, but not all. Refer to [automation_conditions.py](../src/utils/automation_conditions.py) for all custom automation conditions, or Dagster documentation for more customization options.

| Condition | Timing | Description | Use Cases |
|-----------|--------|-------------|-----------|
| `daily_cron_with_eager_historical_backfill_condition` | Daily at 2 AM | **Most common choice** - Daily processing with smart backfill | Daily aggregations, user metrics, reporting tables |
| `hourly_cron_with_eager_historical_backfill_condition` | Every hour | Real-time processing with dependency awareness | Event processing, operational metrics, live dashboards |
| `weekly_cron_with_eager_historical_backfill_condition` | Weekly (Sundays) | Heavy computations with weekly cadence | Weekly reports, expensive aggregations, ML feature tables |

> **✅ All automation conditions include eager historical backfill** - Dagster automatically identifies and backfills missing historical partitions when upstream data becomes available.

---

## Legacy Scheduling

> **⚠️ WARNING: The following legacy scheduling approach is discouraged unless necessary.**
>
> **Legacy schedules lack:**
> - Dependency awareness
> - Smart backfill capabilities
> - Resource optimization
> - Clear failure modes
>
> **Use automation conditions instead.**

### Legacy Schedule Reference (For Migration Only)

If you encounter legacy tags in existing models, migrate them to automation conditions:

| Legacy Tag | ❌ Don't Use | ✅ Migrate To |
|------------|-------------|---------------|
| `cadence=15min` | ❌ | meta.dagster.automation_condition or use tag `automation_condition=hourly_cron_with_eager_historical_backfill_condition` |
| `cadence=hourly` | ❌ | meta.dagster.automation_condition or use tag `automation_condition=hourly_cron_with_eager_historical_backfill_condition` |
| `cadence=daily` | ❌ | meta.dagster.automation_condition or use tag `automation_condition=daily_cron_with_eager_historical_backfill_condition` |
| `cadence=weekly` | ❌ | meta.dagster.automation_condition or use tag `automation_condition=weekly_cron_with_eager_historical_backfill_condition` |

### Legacy Jobs (Being Phased Out):
- `dbt_15min_models_job` - Every 15 minutes ❌
- `dbt_hourly_models_job` - Every hour ❌
- `dbt_daily_models_job` - Daily at 2 AM ❌
- `dbt_weekly_models_job` - Sunday at 3 AM ❌

## Partitioning with partitions_def

Our Dagster setup uses a custom `CustomDagsterDbtTranslator` ([dagster_dbt_translator.py:50-81](../src/assets/dbt/analytics/dagster_dbt_translator.py#L50-L81)) that extracts partition definitions directly from dbt model configs. This allows you to define partitioning behavior in your dbt YAML files rather than in Dagster Python code.

### How It Works

The translator:
1. Reads `partitions_def` from your dbt model's config
2. Creates the appropriate Dagster `PartitionsDefinition` (daily or hourly)
3. Applies `partition_mapping` for upstream dependencies
4. Falls back to legacy `meta.dagster` approach for backwards compatibility

### Partition Definition Configuration

> **⚠️ IMPORTANT: Tag-based partition configuration is DEPRECATED.**
>
> Use `meta.dagster.partitions_def` instead of `partition_type=*` tags. Tag-based configuration will be removed in a future version.

Define partitions directly in your model config using the `partitions_def` key in `meta.dagster`:

```yaml
models:
  - name: agg_play_info_hourly_v2
    meta:
      dagster:
        automation_condition: hourly_cron_with_eager_historical_backfill_condition
        partitions_def:
          type: hourly
          start_date: 2024-06-01
          end_offset: -1
```

**Do NOT use `partition_type=hourly` or `partition_type=daily` tags.** All partition configuration should be done via `meta.dagster.partitions_def`.

### Supported Partition Types

#### Daily Partitions

```yaml
partitions_def:
  type: daily
  start_date: 2024-01-01  # First partition date
  end_offset: 0            # 0 = include today, -1 = exclude today
```

#### Hourly Partitions

```yaml
partitions_def:
  type: hourly
  start_date: 2024-01-01  # First partition date
  end_offset: -1          # -1 = exclude current hour
```

## Partition Mapping for Dependencies

Control how partitions map between models and their upstream dependencies using `partition_mappings`. As partitions may have multiple dependencies, these are configured as a list.

```yaml
models:
  - name: agg_play_info_hourly_v2
    meta:
      dagster:
        partition_mappings:
          - asset_key: fact_play
            type: time_window
            start_offset: 0
            end_offset: 0
            allow_nonexistent_upstream_partitions: false
```

The different types of partition mappings can be found in [utils/dbt.py](../src/utils/dbt.py).

**Examples of partition mapping types:**
- **Exact partition match**: Parent and child use identical partition definitions, and child requires 1:1 mapping. Use IdentityPartitionMapping.
- **Lookback window** (start_offset=-7, end_offset=0): Current partition needs last 7 partitions from upstream. Use TimeWindowPartitionMapping.
- **Future window** (start_offset=0, end_offset=7): Current partition needs next 7 partitions from upstream. Use TimeWindowPartitionMapping.
- **Allow missing partitions**: If multiple upstream data sources have different partition start dates. Use TimeWindowPartitionMapping and set `allow_nonexistent_upstream_partitions: true` when upstream data may not always exist

### Legacy Configuration (Deprecated)

> **⚠️ DEPRECATED**: The following approaches are deprecated and will be removed in a future version:
> - Tag-based configuration: `tags: ["partition_type=daily"]` or `tags: ["automation_condition=..."]`
> - Legacy meta: `meta.dagster.partition_start_date` (without `partitions_def`)

**Deprecated approaches (do not use):**
```yaml
# ❌ DEPRECATED - Tag-based partition configuration
config:
  tags: ["partition_type=daily", "automation_condition=daily_cron_with_eager_historical_backfill_condition"]

# ❌ DEPRECATED - Legacy partition_start_date without partitions_def
meta:
  dagster:
    partition_start_date: "2025-07-31"
```

**Current approach (use this):**
```yaml
# ✅ CORRECT - Use meta.dagster for all configuration
meta:
  dagster:
    automation_condition: daily_cron_with_eager_historical_backfill_condition
    partitions_def:
      type: daily
      start_date: 2025-07-31
      end_offset: 0
```

## Backfill Policies

Control how backfills are executed:

### Model-Level Backfill Configuration

```yaml
models:
  - name: agg_user_hook_platform_daily_spine
    meta:
      dagster:
        partition_start_date: "2025-07-31"
        backfill_policy:
          max_partitions_per_run: 7  # Process 7 partitions per run
```

### Common Backfill Patterns

| Model Type | Recommended `max_partitions_per_run` | Reasoning |
|------------|-------------------------------------|-----------|
| **Heavy aggregations** | 1-7 | Complex calculations require more resources per partition |
| **Light transformations** | 14-30 | Simple transformations can process more partitions efficiently |
| **Simple views** | No limit needed | Views are typically fast and don't need throttling |
| **Large fact tables** | 1-3 | Very large tables may need even more conservative limits |

> **Best Practice**: Start with conservative limits and increase based on performance monitoring. Monitor warehouse usage and query execution times during backfills.

## Groups for Organization

Organize models into logical groups using tags or metadata:

### In dbt_project.yml

```yaml
models:
  analytics:
    staging:
      hooks:
        +tags: ["group=hooks"]
    marts:
      hooks:
        +tags: ["group=hooks"]
      user_analytics:
        +tags: ["group=user_analytics"]
```

### In Model YAML

```yaml
models:
  - name: agg_user_hook_daily
    config:
      group: hooks
    meta:
      dagster:
        automation_condition: daily_cron_with_eager_historical_backfill_condition
        partitions_def:
          type: daily
          start_date: 2025-07-31
          end_offset: 0
```

## Warehouse Configuration

### Model-Level Warehouse Selection

```sql
{{
    config(
        materialized='incremental',
        snowflake_warehouse='FACT_HOOK_PLAY_LARGE'  -- For heavy processing
    )
}}
```

### Available Warehouses

| Warehouse | Use Case | Performance Level |
|-----------|----------|-------------------|
| `DBT_DEV_MEDIUM` | Development and light processing | Medium |
| `FACT_HOOK_PLAY_LARGE` | Heavy hook-related processing | Large |
| `HOOK_SESSION_X_SMALL` | Hook session analysis | Large |
| `DIM_HOOK_HOURLY_LARGE` | Dimension table processing | Large |

More are available - refer to Snowflake to see more options.

> **Tip**: Choose warehouses based on your data volume and processing requirements. Use `DBT_DEV_MEDIUM` for development and testing, and the `LARGE` warehouses for production workloads with significant data volumes.

## Custom Macros

### Partition Filtering Macros

**We use custom macros instead of dbt's built-in `is_incremental` flag** because Dagster passes partition start and end dates that we can leverage:

```sql
-- Instead of this:
{% if is_incremental() %}
  WHERE p_date > (SELECT MAX(p_date) FROM {{ this }})
{% endif %}

-- We use this:
WHERE {{ partition_filter_daily('p_date') }}
```

### Daily Partition Filter

```sql
{% macro partition_filter_daily(p_date_column) %}
{{ p_date_column }} >= '{{ var('partition_start_date', '1900-01-01') }}' AND {{ p_date_column }} < '{{ var('partition_end_date', '9999-12-31') }}'
{% endmacro %}
```

### Hourly Partition Filter

```sql
{% macro partition_filter_hourly(p_date_column, p_hour_column) %}
IFF(
    '{{ var('partition_start_date', '1900-01-01') }}' = '{{ var('partition_end_date', '9999-12-31') }}',
    {{ p_date_column }} = '{{ var('partition_start_date', '1900-01-01') }}' AND {{ p_hour_column }} >= {{ var('partition_start_hour', 0) }} AND {{ p_hour_column }} < {{ var('partition_end_hour', 23) }},
    (
        ({{ p_date_column }} = '{{ var('partition_start_date', '1900-01-01') }}' AND {{ p_hour_column }} >= {{ var('partition_start_hour', 0) }}) OR
        ({{ p_date_column }} > '{{ var('partition_start_date', '1900-01-01') }}' AND {{ p_date_column }} < '{{ var('partition_end_date', '9999-12-31') }}') OR
        ({{ p_date_column }} = '{{ var('partition_end_date', '9999-12-31') }}' AND {{ p_hour_column }} < {{ var('partition_end_hour', 23) }})
    )
)
{% endmacro %}
```

## Complete Model Example

### Daily Partitioned Model

```sql
{{
    config(
        materialized='incremental',
        incremental_strategy='delete+insert',
        unique_key=['user_id', 'p_date'],
        snowflake_warehouse='FACT_HOOK_PLAY_LARGE',
        cluster_by=['p_date']
    )
}}

WITH user_daily_activity AS (
    SELECT
        user_id,
        user_uid,
        SUM(play_count) AS total_plays,
        SUM(play_duration_seconds) AS total_duration,
        MAX(is_active) AS is_active,
        p_date
    FROM {{ ref('agg_user_hook_platform_daily_spine') }}
    WHERE {{ partition_filter_daily('p_date') }}
    GROUP BY user_id, user_uid, p_date
)

SELECT
    u.user_id,
    u.user_uid,
    du.subscription_tier,
    u.total_plays,
    u.total_duration,
    u.is_active,
    u.p_date
FROM user_daily_activity u
LEFT JOIN {{ ref('stg_cleaned__dim_user') }} du
    ON u.user_id = du.user_id
WHERE u.is_active = TRUE
```

### Hourly Partitioned Model with Partition Mapping

```sql
{{
    config(
        materialized='incremental',
        incremental_strategy='delete+insert',
        unique_key=['user_id', 'clip_id', 'p_date', 'p_hour', 'play_duration_threshold'],
        cluster_by=['p_date', 'p_hour']
    )
}}

SELECT
    user_id,
    clip_id,
    SUM(play_duration_seconds) AS play_duration_seconds,
    COUNT(*) AS play_cnt,
    p_date,
    p_hour
FROM {{ source('snowflake', 'fact_play_v2') }}
WHERE {{ partition_filter_hourly('p_date', 'p_hour') }}
GROUP BY user_id, clip_id, p_date, p_hour
```

## Model YAML Documentation

### Daily Partitioned Model YAML

```yaml
models:
  - name: agg_user_hook_daily
    description: "Daily user hook activity aggregated across platforms"
    config:
      database: "{{ env_var('SNOWFLAKE_DB') }}"
      schema: "{{ env_var('SNOWFLAKE_SCHEMA') }}"
      warehouse: FACT_HOOK_PLAY_LARGE
      group: hooks
    meta:
      dagster:
        automation_condition: daily_cron_with_eager_historical_backfill_condition
        partitions_def:
          type: daily
          start_date: 2025-07-31
          end_offset: 0
        backfill_policy:
          max_partitions_per_run: 14
    columns:
      - name: user_id
        description: "User identifier"
        tests:
          - not_null
      - name: total_plays
        description: "Total hook plays across all platforms"
      - name: p_date
        description: "Date partition"
        tests:
          - not_null
```

### Hourly Partitioned Model YAML with Partition Mapping

```yaml
models:
  - name: agg_play_info_hourly_v2
    description: "Hourly aggregated play information on clips"
    config:
      database: "{{ env_var('SNOWFLAKE_DB') }}"
      schema: "{{ env_var('SNOWFLAKE_SCHEMA') }}"
      warehouse: SUNO_PROD_AGG_HOURLY_LARGE
      group: clips
    meta:
      dagster:
        automation_condition: hourly_cron_with_eager_historical_backfill_condition
        partitions_def:
          type: hourly
          start_date: 2024-06-01
          end_offset: -1
        partition_mappings:
          - asset_key: fact_play
            type: identity
        backfill_policy:
          max_partitions_per_run: 336  # 2 weeks
    columns:
      - name: user_id
        description: "User ID (integer)"
        tests:
          - not_null
      - name: clip_id
        description: "Clip ID"
        tests:
          - not_null
      - name: p_date
        description: "Partition date"
      - name: p_hour
        description: "Partition hour (0-23)"
```

## Environment-Specific Configuration

### Dynamic Database References

```yaml
sources:
  - name: suno_data
    database: "{{ 'suno_prod' if target.name == 'prod' else 'suno_staging' }}"
    schema: "{{ 'prod' if target.name == 'prod' else 'staging' }}"
```

### Environment Variables

```yaml
config:
  database: "{{ env_var('SNOWFLAKE_DB') }}"
  schema: "{{ env_var('SNOWFLAKE_SCHEMA') }}"
```

## Best Practices

### 1. Model Organization
- **Raw**: Direct source references with minimal transformation
- **Staging**: Cleaned, typed, and renamed columns
- **Intermediate**: Business logic, ephemeral models
- **Marts**: Final tables for consumption

### 2. Incremental Strategy Selection
- **delete+insert**: Most reliable, good for most use cases
- **merge**: For complex upsert logic
- **insert_overwrite**: For partition-based overwrites

### 3. Performance Optimization
- Use appropriate warehouses for different workloads
- Cluster tables on frequently filtered columns
- Use ephemeral models for intermediate transformations
- Consider using `on_schema_change: 'append_new_columns'` for evolving schemas

### 4. Partition Management
- **Use `meta.dagster.partitions_def`** for all partition configuration (do NOT use `partition_type=*` tags)
- **Use `meta.dagster.automation_condition`** for automation configuration (do NOT use `automation_condition=*` tags)
- Always use partition filtering macros (`partition_filter_daily` or `partition_filter_hourly`)
- Set appropriate backfill limits based on processing requirements
- Use `partition_mapping` for complex upstream dependencies (lookback windows, etc)
- Set `allow_nonexistent_upstream_partitions: true` only when upstream partitions might legitimately not exist

### 5. Testing and Quality
- Add `not_null` tests for key columns
- Use `unique` tests for unique keys
- Test relationships between models
- Document all columns with descriptions

### 6. Scheduling and Orchestration
- **Always use automation conditions** - Never use legacy `cadence=*` tags
- **Configure via `meta.dagster.automation_condition`** - Do NOT use `automation_condition=*` tags
- **Configure partitions via `meta.dagster.partitions_def`** - Do NOT use `partition_type=*` tags
- Choose appropriate automation condition based on data freshness needs
- Leverage built-in backfill capabilities instead of manual processes
- Monitor automation condition performance in Dagster UI

### 7. Development Workflow
- Test models in development environment first
- Use descriptive model and column names
- Keep models focused on single business concepts
- Document complex business logic in comments

### 8. Error Handling
- Use `on_schema_change: 'fail'` for critical models
- Implement proper error handling in macros
- Monitor model execution times and resource usage

## Troubleshooting

### Common Issues

#### Model Materialization Failures
- **Issue**: Models fail to materialize with warehouse errors
- **Solution**: Check warehouse configuration and ensure appropriate warehouse is selected for the workload

#### Partition Filter Issues
- **Issue**: Models process too much data or miss partitions
- **Solution**: Verify partition filtering macros are correctly implemented and partition variables are set

#### Backfill Performance
- **Issue**: Backfills are too slow or fail due to resource limits
- **Solution**: Reduce `max_partitions_per_run` or use a larger warehouse

#### Schema Change Errors
- **Issue**: Models fail when source schemas change
- **Solution**: Use appropriate `on_schema_change` configuration or update model logic

### Debugging Tips

1. **Check Dagster logs** for detailed error messages
2. **Use dbt debug** to validate project configuration
3. **Test models locally** before deploying to production
4. **Monitor warehouse usage** during materialization
5. **Use dbt compile** to check SQL generation

## Additional Resources

- [dbt Documentation](https://docs.getdbt.com/)
- [Dagster Documentation](https://docs.dagster.io/)
- [Snowflake Documentation](https://docs.snowflake.com/)
- Internal Slack channels: `#data-engineering`, `#dbt-help`

---

This guide provides the foundation for building efficient, well-orchestrated dbt models in our Dagster-managed environment. For questions or improvements to this guide, please reach out to the Data Engineering team.