[Feature] Select the nodes to run or skip within a DbtDAG at runtime

[SoheilSalmani]

Description

I would like to be able to select the tasks to run and to skip within a DbtDAG at runtime.

Use case/motivation

Suppose I have 4 massive tables built using incremental materialization in dbt, and they are all in the same DbtDag. If I introduce breaking changes to one of them, I would like to be able to full_refresh it without running the other nodes. I think it would be great to have the possibility to select the nodes to run at runtime if I only want to full refresh one of them (and their dependencies), typically by using the dbt selection syntax in an Airflow Param when manually triggering the DAG.

I was thinking about adding a select and exclude parameters in operator_args. We would be able to select the nodes to run in the DAG at runtime (among the nodes that were rendered using the render_config), and skip the others:

dag = DbtDag(
    dag_id=dag_id,
    project_config=ProjectConfig(dbt_project_path=dbt_project_path),
    render_config=RenderConfig(
        # Render every node in the `marketing` folder
        select=["marketing"],
        emit_datasets=False,
    ),
    profile_config=profile_config,
    execution_config=venv_execution_config,
    schedule=None,
    operator_args={"select": "{{ params.select }}", "exclude": "{{ params.exclude }}"},
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["dbt"],
    params={
        "full_refresh": Param(
            False, type="boolean", description="Full-refresh models?"
        ),
        # Passing `source:google_analytics+` at runtime will only run the tasks that
        # depend on the `google_analytics` source, and skip the rest.
        "select": Param(
            None,
            type=["string", "null"],
            description="dbt selection syntax to select tasks to run within the DAG.",
        ),
        "exclude": Param(
            None,
            type=["string", "null"],
            description="dbt selection syntax to select tasks to skip within the DAG.",
        ),
    },
    render_template_as_native_obj=True,
)

I assume one of the main challenge would be to only select and exclude a subset of the nodes that were rendered in the DAG, so that we do not run a dbt command that interferes with models outside of the DAG scope.

Related issues

No response

Are you willing to submit a PR?

• Yes, I am willing to submit a PR!