Dataproduct Dbt

Build and run dbt transformations for a data product

dataproduct-implement generates the ends of the pipeline: input-port sources (from access agreements) and output-port models (from the data contract). This skill fills in the middle, the staging/ and intermediate/ layers, and runs dbt against the result. The conventions below are adapted from dbt's structure and materialization best practices, and are meant to be edited by the organization adopting this plugin.

When to use this vs. other skills

• No dbt project yet → run dataproduct-bootstrap first.
• No output-port models or input-port sources yet → run dataproduct-implement first.
• Auditing or reviewing an existing dbt project → use /review or /simplify, not this skill.
• Authoring or refactoring the middle of the DAG, or running dbt → this skill.

Conventions (the editable part)

These rules are the contract between this skill and the rest of the plugin. Organizations forking the plugin should treat this section as the place to encode their own style.

Layer responsibilities

Deviations from upstream dbt conventions, called out so the lineage stays visible:

• We use input_ports/ and output_ports/v1/ instead of dbt's sources / marts, to match the data-product lifecycle terminology.
• Output-port models are not prefixed with fct_ / dim_. The contract dictates the published name, and consumers see that name directly.
• We keep staging/ and intermediate/ flat for single-data-product repos. Use subfolders by output port (e.g. staging/<output-port-id>/) only if a data product has more than two or three output ports.

Staging model shape (CTE pattern)

Every staging model uses this three-CTE skeleton so the structure is uniform and a reader knows where to find the rename block.

-- Staging model for source <provider-dp-id>_<provider-op-id>.<table>
with

source as (
    select * from {{ source('<provider-dp-id>_<provider-op-id>', '<table>') }}
),

renamed as (
    select
        cast(<raw_col> as <warehouse_type>) as <canonical_name>,
        ...
    from source
),

final as (
    select * from renamed
)

select * from final

Allowed transformations in renamed: column renames, casts, simple case rewrites, unit conversions (e.g. cents to dollars), trim/lower. Disallowed: joins, aggregations, any cross-source logic. Those belong in intermediate/.

Intermediate model shape

• One file per purpose. If you find yourself naming it int_foo_and_bar, split it.
• All inputs come through ref(), never source(). A source() call in intermediate/ means a staging model is missing.
• Pick view unless the model is referenced once and is cheap, in which case ephemeral keeps it out of the warehouse.

Tests

Tests are declared in the _models.yml next to the file.

Custom singular tests live in tests/<purpose>.sql and assert cross-model invariants the schema-tests cannot express.

Materializations

dbt's tiered rule applies: start with view, promote to table when query latency hurts, promote to incremental when the build window hurts.

• Default per layer (set in dbt_project.yml): staging → view, intermediate → view, output_ports → table.
• Switch an output port to incremental when one of these is true:
  • The model takes longer than 5 minutes to build full-refresh.
  • The source data grows by more than ~10M rows per scheduled run.
  • A CI run cannot rebuild the model within the deploy budget.

When switching to incremental, configure all three keys; defaults vary by warehouse:

{{ config(
    materialized='incremental',
    unique_key='<natural_key>',
    on_schema_change='append_new_columns',
    incremental_strategy='merge'   # databricks/snowflake/bigquery; use 'delete+insert' on postgres
) }}

Inside the model body, gate new-row logic with {% if is_incremental() %} where updated_at > (select max(updated_at) from {{ this }}) {% endif %}.

Schemas

Each layer lands in its own warehouse schema so consumers see only the published output, not the scaffolding.

Why this split:

• The output-port schema embeds the output port version so v1 and v2 of the same port can coexist while one is being rolled out. Consumers are granted access to op_<output-port-id>_v<N> only.
• Staging and intermediate share a single internal schema, scoped by data product id so two data products on the same warehouse do not collide. That schema is not granted to consumers.

To wire this up in dbt, the project needs two pieces:

1. Override generate_schema_name so +schema: is taken literally instead of being suffixed onto the target's default schema. Place this in macros/get_custom_schema.sql:

   {% macro generate_schema_name(custom_schema_name, node) -%}
       {%- if custom_schema_name is none -%}
           {{ target.schema }}
       {%- else -%}
           {{ custom_schema_name | trim }}
       {%- endif -%}
   {%- endmacro %}

2. Per-model schema for output ports (one schema per port, so a directory-level rule does not work when v<N>/ contains more than one port):

   -- top of models/output_ports/v1/<table>.sql
   {{ config(schema='op_<output_port_id>_v1') }}

3. Directory-level schema for internal models in dbt_project.yml:

   models:
     <data_product_id>:
       staging:
         +schema: internal_<data_product_id>
       intermediate:
         +schema: internal_<data_product_id>

With these in place, dbt run creates exactly op_<op-id>_v1, op_<op-id>_v2, and internal_<dp-id> in the warehouse, no extra prefixes.

Documentation

• Every model has a description: in _models.yml. Single sentence: what it represents, not how it is built.
• Every column with a non-obvious meaning has a description:. Obvious ones (customer_id, created_at) can skip it.
• Output-port descriptions come from the contract; do not edit them locally — edit the contract via datacontract-edit.

How to run this skill

${PLUGIN_ROOT} below refers to the root of this plugin (the directory that contains skills/). On Claude Code it is set automatically as ${CLAUDE_PLUGIN_ROOT}; use that. On any other agent it is unset; resolve it as ../.. relative to this SKILL.md file's directory.

Plan announcement (before Step 0)

Before running Step 0, print this plan to the user verbatim:

Running dataproduct-dbt. I'll:

1. Pre-checks: confirm this is a dbt project with input_ports/, staging/, intermediate/, output_ports/v1/.
2. Identify which operation you want (build staging, add intermediate, refactor an output port, run dbt, switch to incremental).
3. Apply the operation following the conventions in this skill (which you can edit to match your org's style).
4. Verify with dbt parse and, when you ask, dbt build --select <scope>.
5. Summarize what changed and what's open.

Then proceed.

Step 0 — Pre-checks

• Confirm uv run --quiet dbt --version succeeds from the project root. If it fails, run uv sync and retry; if still missing, stop and tell the user to add the dbt adapter for their warehouse to pyproject.toml's [dependency-groups].dev (e.g. dbt-snowflake, dbt-databricks) and re-run uv sync. Use uv run dbt … for every dbt CLI invocation in this skill.
• Confirm dbt_project.yml exists at the working directory root. If not, route to dataproduct-bootstrap.
• Confirm the four model directories exist (input_ports/, staging/, intermediate/, output_ports/v1/). If any are missing, route to dataproduct-bootstrap or entropy-data-sync (whichever the user prefers) and stop.

Step 1 — Identify the operation

Match the user's ask to one of the operations below. If two fit, ask which one. The operations are deliberately small so the skill stays focused; run it again for the next operation.

Step 2 — Apply the operation