Refactor group_nodes_by_folder hierarchical naming to be opt-in (#2862)

#2824 fixed `group_nodes_by_folder` collapsing folders that share a leaf
name under different parents (e.g. `staging/aurora` and `marts/aurora`)
— thanks @anor4k for catching and fixing this!

My only concern is that it changes the names of the generated Cosmos
task groups, which is a breaking change for users who reference those
task groups from other tasks, e.g.:

```
non_cosmos_task = ...
cosmos_dbt_task_group = task_group.children["tg_a.models"].children["tg_a.models.marts"]

non_cosmos_task >> cosmos_dbt_task_group
```

To avoid this breaking change in a 1.x release, this PR makes the fix
configurable, defaulting it to `False` and following the same config
pattern as the existing Cosmos settings:
https://astronomer.github.io/astronomer-cosmos/reference/configs/cosmos-conf.html

When `enable_hierarchical_naming_for_group_nodes_by_folder` is disabled
(the default), Cosmos keeps the previous task-group names. When it is
enabled — through the setting or the
`AIRFLOW__COSMOS__ENABLE_HIERARCHICAL_NAMING_FOR_GROUP_NODES_BY_FOLDER`
environment variable — Cosmos applies the #2824 fix, and folders that
share a leaf name render as distinct task groups.

We intend to make hierarchical naming the only behaviour, and drop this
setting altogether, in Cosmos 2.0: #2863

Author: tatiana

cosmos/airflow/graph.py

@@ -3,6 +3,7 @@
 from collections import OrderedDict, defaultdict
 from collections.abc import Callable
 from copy import deepcopy
+from pathlib import Path
 from typing import Any
 
 try:  # Airflow 3
@@ -948,26 +949,35 @@ def identify_detached_nodes(
 
 
 def create_task_groups_based_on_folder(
-    dag: DAG, node: DbtNode, parent_task_group: TaskGroup | None, task_groups: dict[str, TaskGroup]
+    dag: DAG,
+    node: DbtNode,
+    parent_task_group: TaskGroup | None,
+    task_groups: dict[str, TaskGroup],
+    use_hierarchical_naming: bool = False,
 ) -> TaskGroup | None:
     """
     Generate the parent task group for the given node based on the node's file path. If a TaskGroup is given, it will
     be used as the parent group.
+
+    When ``use_hierarchical_naming`` is True, groups are cached by their cumulative folder path, so that folders
+    sharing a leaf name under different parents (e.g. ``staging/aurora`` and ``marts/aurora``) become distinct task
+    groups instead of collapsing into the first one created. When False (default), groups are cached by the bare
+    folder name, preserving the legacy behaviour — and the existing task-group ids that users may reference — at the
+    cost of collapsing same-named folders. See https://github.com/astronomer/astronomer-cosmos/pull/2824.
     """
     task_group = None
-    resource_file_path_parts = str(node.original_file_path).split("/")[:-1]
-    # Cache groups by their cumulative path rather than the bare folder name, so
-    # that folders sharing a leaf name under different parents (e.g.
-    # ``staging/aurora`` and ``marts/aurora``) become distinct task groups instead
-    # of collapsing into the first one created.
+    # Use ``Path.parts`` (OS-agnostic) rather than ``str(path).split("/")`` so folder grouping also
+    # works on Windows, where ``str(Path(...))`` uses backslash separators.
+    resource_file_path_parts = Path(node.original_file_path).parts[:-1]
     cumulative_path = ""
     for resource_file_path_part in resource_file_path_parts:
         cumulative_path = f"{cumulative_path}/{resource_file_path_part}" if cumulative_path else resource_file_path_part
-        if cumulative_path in task_groups:
-            task_group = task_groups[cumulative_path]
+        cache_key = cumulative_path if use_hierarchical_naming else resource_file_path_part
+        if cache_key in task_groups:
+            task_group = task_groups[cache_key]
         else:
             task_group = TaskGroup(dag=dag, group_id=resource_file_path_part, parent_group=parent_task_group)
-            task_groups[cumulative_path] = task_group
+            task_groups[cache_key] = task_group
         parent_task_group = task_group
     return task_group
 
@@ -1145,6 +1155,7 @@ def build_airflow_graph(
     :return: Dictionary mapping dbt nodes (node.unique_id to Airflow task)
     """
     group_nodes_by_folder = render_config.group_nodes_by_folder
+    use_hierarchical_naming = settings.enable_hierarchical_naming_for_group_nodes_by_folder
     tasks_map: dict[str, TaskGroup | BaseOperator] = {}
     task_groups: dict[str, TaskGroup] = {}
     task_or_group: TaskGroup | BaseOperator | None
@@ -1180,7 +1191,7 @@ def build_airflow_graph(
 
     for node_id, node in nodes.items():
         task_group = (
-            create_task_groups_based_on_folder(dag, node, parent_task_group, task_groups)
+            create_task_groups_based_on_folder(dag, node, parent_task_group, task_groups, use_hierarchical_naming)
             if group_nodes_by_folder
             else task_group
         )

cosmos/settings.py

@@ -33,6 +33,13 @@
 enable_cache_dbt_ls = conf.getboolean("cosmos", "enable_cache_dbt_ls", fallback=True)
 enable_cache_dbt_yaml_selectors = conf.getboolean("cosmos", "enable_cache_dbt_yaml_selectors", fallback=True)
 enable_lax_selector_parsing = conf.getboolean("cosmos", "enable_lax_selector_parsing", fallback=False)
+# When RenderConfig.group_nodes_by_folder is enabled, key folder task groups by their full path so
+# that folders sharing a leaf name under different parents render as distinct task groups. Defaults
+# to False to preserve existing task-group ids (enabling it is a breaking change for DAGs that
+# reference Cosmos task groups by id); expected to become the default in Cosmos 2.0. See #2824.
+enable_hierarchical_naming_for_group_nodes_by_folder = conf.getboolean(
+    "cosmos", "enable_hierarchical_naming_for_group_nodes_by_folder", fallback=False
+)
 rich_logging = conf.getboolean("cosmos", "rich_logging", fallback=False)
 dbt_docs_dir = conf.get("cosmos", "dbt_docs_dir", fallback=None)
 dbt_docs_conn_id = conf.get("cosmos", "dbt_docs_conn_id", fallback=None)

docs/guides/translate_dbt_to_airflow/render-config.rst

@@ -33,7 +33,7 @@ The ``RenderConfig`` class takes the following arguments:
 - ``should_detach_multiple_parents_tests``: A boolean to control if tests that depend on multiple parents should be run as standalone tasks. See :doc:`Testing Behavior </guides/translate_dbt_to_airflow/testing-behavior>` for more information.
 - ``enable_owner_inheritance``: (introduced in 1.10.2) A boolean to control if dbt owners should be imported as part of the airflow DAG owners. Defaults to True.
 - ``ephemeral_models_as_empty_operator``: (new in v1.15.0) A boolean to control how ephemeral models are rendered. Ephemeral models are inlined as CTEs into downstream models and never written to the warehouse, so running them via a dbt operator is effectively a no-op. When ``True`` (default), they are rendered as ``EmptyOperator`` tasks, which preserves the dependency chain that passes through them while avoiding wasted dbt invocations and decluttering the DAG. Because the ``EmptyOperator`` does not run dbt, behaviour tied to the ephemeral model's run task no longer occurs: no Airflow Dataset/Asset is emitted for it (so dataset-scheduled DAGs keyed on it are not triggered), task callbacks are not invoked, no OpenLineage events are produced, and per-node operator arguments (e.g. ``profile_args``) do not apply. Set to ``False`` to render them as regular dbt run tasks.
-- ``group_nodes_by_folder``: When enabled, groups nodes by folder structure, creating a ``TaskGroup`` per resource type and folder. Disabled by default.
+- ``group_nodes_by_folder``: When enabled, groups nodes by folder structure, creating a ``TaskGroup`` per resource type and folder. Disabled by default. By default, folders that share a leaf name under different parents (e.g. ``staging/aurora`` and ``marts/aurora``) collapse into a single task group; set the :ref:`enable_hierarchical_naming_for_group_nodes_by_folder` Cosmos config (or the ``AIRFLOW__COSMOS__ENABLE_HIERARCHICAL_NAMING_FOR_GROUP_NODES_BY_FOLDER`` environment variable) to ``True`` to render them as distinct task groups. It is disabled by default because enabling it changes task-group ids, which is a breaking change for DAGs that reference Cosmos task groups by id, and is expected to become the default in Cosmos 2.0.
 
 How to run dbt ls (invocation mode)
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

docs/reference/configs/cosmos-conf.rst

@@ -73,6 +73,22 @@ This page lists all available `Apache Airflow® <https://airflow.apache.org/>`_
     - Default: ``False``
     - Environment Variable: ``AIRFLOW__COSMOS__ENABLE_LAX_SELECTOR_PARSING``
 
+.. _enable_hierarchical_naming_for_group_nodes_by_folder:
+
+`enable_hierarchical_naming_for_group_nodes_by_folder`_:
+    Only relevant when ``RenderConfig.group_nodes_by_folder`` is enabled. When enabled, folder task groups
+    are keyed by their full folder path, so folders that share a leaf name under different parents
+    (e.g. ``staging/aurora`` and ``marts/aurora``) render as distinct task groups instead of collapsing into
+    the first one created. Disabled by default to preserve the existing task-group ids — enabling it changes
+    those ids, which is a breaking change for DAGs that reference Cosmos task groups by id. This is expected
+    to become the default in Cosmos 2.0.
+
+    Being an Airflow configuration setting, it applies **globally to every Cosmos DAG** in the deployment
+    and cannot be configured per DAG (unlike the ``RenderConfig.group_nodes_by_folder`` option itself).
+
+    - Default: ``False``
+    - Environment Variable: ``AIRFLOW__COSMOS__ENABLE_HIERARCHICAL_NAMING_FOR_GROUP_NODES_BY_FOLDER``
+
 .. _enable_cache_partial_parse:
 
 `enable_cache_partial_parse`_:

tests/airflow/test_graph.py

@@ -146,8 +146,16 @@ def test_calculate_datached_node_name_under_is_under_250():
     assert calculate_detached_node_name(node) == "detached_1_test"
 
 
+@pytest.fixture
+def hierarchical_naming_disabled(monkeypatch):
+    """Pin the global hierarchical-naming setting to False so the folder-grouping tests stay hermetic
+    regardless of any AIRFLOW__COSMOS__ENABLE_HIERARCHICAL_NAMING_FOR_GROUP_NODES_BY_FOLDER set in the
+    developer's environment. See https://github.com/astronomer/astronomer-cosmos/issues/1763."""
+    monkeypatch.setattr("cosmos.settings.enable_hierarchical_naming_for_group_nodes_by_folder", False)
+
+
 @pytest.mark.integration
-def test_build_airflow_graph_with_after_each():
+def test_build_airflow_graph_with_after_each(hierarchical_naming_disabled):
     with DAG("test-id", start_date=datetime(2022, 1, 1)) as dag:
         task_args = {
             "project_dir": SAMPLE_PROJ_PATH,
@@ -179,13 +187,13 @@ def test_build_airflow_graph_with_after_each():
         "seed_parent_seed",
         "gen2.models.parent.run",
         "gen2.models.parent.test",
-        "gen3.models.child_run",
-        "gen3.models.child2_v2_run",
+        "gen2.models.child_run",
+        "gen2.models.child2_v2_run",
     ]
 
     assert topological_sort == expected_sort
     task_groups = dag.task_group_dict
-    assert len(task_groups) == 5
+    assert len(task_groups) == 4
 
     assert task_groups["gen2.models.parent"].upstream_task_ids == {"seed_parent_seed"}
     assert list(task_groups["gen2.models.parent"].children.keys()) == [
@@ -194,8 +202,8 @@ def test_build_airflow_graph_with_after_each():
     ]
 
     assert len(dag.leaves) == 2
-    assert dag.leaves[0].task_id == "gen3.models.child_run"
-    assert dag.leaves[1].task_id == "gen3.models.child2_v2_run"
+    assert dag.leaves[0].task_id == "gen2.models.child_run"
+    assert dag.leaves[1].task_id == "gen2.models.child2_v2_run"
 
     task_seed_parent_seed = dag.tasks[0]
     task_parent_run = dag.tasks[1]
@@ -341,7 +349,48 @@ def test_generate_task_or_group_with_dynamic_node_type_no_converter_returns_none
 
 
 @pytest.mark.integration
-def test_build_airflow_graph_with_after_all():
+def test_build_airflow_graph_hierarchical_naming_enabled(monkeypatch):
+    """With the global setting enabled, folders that share a leaf name under different parents render
+    as distinct task groups (``gen3.models`` instead of collapsing into ``gen2.models``) — the opt-in
+    #2824 behaviour, wired through build_airflow_graph. See
+    https://github.com/astronomer/astronomer-cosmos/issues/1763."""
+    monkeypatch.setattr("cosmos.settings.enable_hierarchical_naming_for_group_nodes_by_folder", True)
+    with DAG("test-id", start_date=datetime(2022, 1, 1)) as dag:
+        task_args = {
+            "project_dir": SAMPLE_PROJ_PATH,
+            "conn_id": "fake_conn",
+            "profile_config": ProfileConfig(
+                profile_name="default",
+                target_name="default",
+                profile_mapping=PostgresUserPasswordProfileMapping(
+                    conn_id="fake_conn",
+                    profile_args={"schema": "public"},
+                ),
+            ),
+        }
+        build_airflow_graph(
+            nodes=sample_nodes,
+            dag=dag,
+            execution_mode=ExecutionMode.LOCAL,
+            test_indirect_selection=TestIndirectSelection.EAGER,
+            task_args=task_args,
+            render_config=RenderConfig(
+                group_nodes_by_folder=True,
+                test_behavior=TestBehavior.AFTER_EACH,
+                source_rendering_behavior=SOURCE_RENDERING_BEHAVIOR,
+            ),
+            dbt_project_name="astro_shop",
+        )
+    task_groups = dag.task_group_dict
+    # gen3/models becomes its own task group instead of collapsing into gen2.models
+    assert "gen3.models" in task_groups
+    assert len(task_groups) == 5
+    leaf_ids = {leaf.task_id for leaf in dag.leaves}
+    assert leaf_ids == {"gen3.models.child_run", "gen3.models.child2_v2_run"}
+
+
+@pytest.mark.integration
+def test_build_airflow_graph_with_after_all(hierarchical_naming_disabled):
     with DAG("test-id", start_date=datetime(2022, 1, 1)) as dag:
         task_args = {
             "project_dir": SAMPLE_PROJ_PATH,
@@ -374,14 +423,14 @@ def test_build_airflow_graph_with_after_all():
     expected_sort = [
         "seed_parent_seed",
         "gen2.models.parent_run",
-        "gen3.models.child_run",
-        "gen3.models.child2_v2_run",
+        "gen2.models.child_run",
+        "gen2.models.child2_v2_run",
         "astro_shop_test",
     ]
     assert topological_sort == expected_sort
 
     task_groups = dag.task_group_dict
-    assert len(task_groups) == 4
+    assert len(task_groups) == 3
 
     assert len(dag.leaves) == 1
     assert dag.leaves[0].task_id == "astro_shop_test"
@@ -426,7 +475,7 @@ def test_build_airflow_graph_with_after_all_and_empty_nodes():
 
 
 @pytest.mark.integration
-def test_build_airflow_graph_with_build():
+def test_build_airflow_graph_with_build(hierarchical_naming_disabled):
     with DAG("test-id", start_date=datetime(2022, 1, 1)) as dag:
         task_args = {
             "project_dir": SAMPLE_PROJ_PATH,
@@ -457,17 +506,17 @@ def test_build_airflow_graph_with_build():
     expected_sort = [
         "seed_parent_seed_build",
         "gen2.models.parent_model_build",
-        "gen3.models.child_model_build",
-        "gen3.models.child2_v2_model_build",
+        "gen2.models.child_model_build",
+        "gen2.models.child2_v2_model_build",
     ]
     assert topological_sort == expected_sort
 
     task_groups = dag.task_group_dict
-    assert len(task_groups) == 4
+    assert len(task_groups) == 3
 
     assert len(dag.leaves) == 2
-    assert dag.leaves[0].task_id in ("gen3.models.child_model_build", "gen3.models.child2_v2_model_build")
-    assert dag.leaves[1].task_id in ("gen3.models.child_model_build", "gen3.models.child2_v2_model_build")
+    assert dag.leaves[0].task_id in ("gen2.models.child_model_build", "gen2.models.child2_v2_model_build")
+    assert dag.leaves[1].task_id in ("gen2.models.child_model_build", "gen2.models.child2_v2_model_build")
 
 
 @pytest.mark.integration
@@ -506,10 +555,7 @@ def test_build_airflow_graph_with_override_profile_config():
     assert generated_parent_profile_config.profile_mapping.profile_args["schema"] == "public"
 
 
-def test_create_task_groups_based_on_folder_distinguishes_same_leaf_name():
-    """Folders sharing a leaf name under different parents (e.g. ``staging/aurora``
-    and ``marts/aurora``) must map to distinct TaskGroups instead of collapsing
-    into the first one created."""
+def _same_leaf_name_nodes() -> tuple[DbtNode, DbtNode]:
     staging_node = DbtNode(
         unique_id=f"{DbtResourceType.MODEL.value}.{SAMPLE_PROJ_PATH.stem}.staging_aurora_orders",
         resource_type=DbtResourceType.MODEL,
@@ -524,14 +570,22 @@ def test_create_task_groups_based_on_folder_distinguishes_same_leaf_name():
         path_base=SAMPLE_PROJ_PATH,
         original_file_path=Path("models/marts/aurora/orders.sql"),
     )
+    return staging_node, marts_node
+
+
+def test_create_task_groups_based_on_folder_distinguishes_same_leaf_name():
+    """With ``use_hierarchical_naming=True``, folders sharing a leaf name under different parents
+    (e.g. ``staging/aurora`` and ``marts/aurora``) must map to distinct TaskGroups instead of
+    collapsing into the first one created."""
+    staging_node, marts_node = _same_leaf_name_nodes()
 
     task_groups: dict = {}
     with DAG("test-folder-collision", start_date=datetime(2022, 1, 1)) as dag:
         staging_group = create_task_groups_based_on_folder(
-            dag=dag, node=staging_node, parent_task_group=None, task_groups=task_groups
+            dag=dag, node=staging_node, parent_task_group=None, task_groups=task_groups, use_hierarchical_naming=True
         )
         marts_group = create_task_groups_based_on_folder(
-            dag=dag, node=marts_node, parent_task_group=None, task_groups=task_groups
+            dag=dag, node=marts_node, parent_task_group=None, task_groups=task_groups, use_hierarchical_naming=True
         )
 
     assert staging_group is not None
@@ -550,6 +604,27 @@ def test_create_task_groups_based_on_folder_distinguishes_same_leaf_name():
     }
 
 
+def test_create_task_groups_based_on_folder_collapses_same_leaf_name_by_default():
+    """Default (``use_hierarchical_naming=False``) preserves the legacy behaviour: folders sharing a
+    leaf name are keyed by the bare folder name, so the second ``aurora`` reuses the first group.
+    This keeps existing task-group ids stable (avoiding a breaking change) until the behaviour is
+    flipped in a future major release."""
+    staging_node, marts_node = _same_leaf_name_nodes()
+
+    task_groups: dict = {}
+    with DAG("test-folder-collision-legacy", start_date=datetime(2022, 1, 1)) as dag:
+        staging_group = create_task_groups_based_on_folder(
+            dag=dag, node=staging_node, parent_task_group=None, task_groups=task_groups
+        )
+        marts_group = create_task_groups_based_on_folder(
+            dag=dag, node=marts_node, parent_task_group=None, task_groups=task_groups
+        )
+
+    # Both folders collapse onto the same cached ``aurora`` group (legacy behaviour).
+    assert staging_group is marts_group
+    assert set(task_groups) == {"models", "staging", "marts", "aurora"}
+
+
 def test_calculate_operator_class():
     class_module_import_path = calculate_operator_class(execution_mode=ExecutionMode.KUBERNETES, dbt_class="DbtSeed")
     assert class_module_import_path == "cosmos.operators.kubernetes.DbtSeedKubernetesOperator"
@@ -1640,7 +1715,7 @@ def test_watcher_producer_preserves_existing_dbt_cmd_flags(test_behavior):
     assert "--resource-type" in producer_task.dbt_cmd_flags
 
 
-def test_custom_meta():
+def test_custom_meta(hierarchical_naming_disabled):
     with DAG("test-id", start_date=datetime(2022, 1, 1)) as dag:
         task_args = {
             "project_dir": SAMPLE_PROJ_PATH,
@@ -1669,12 +1744,12 @@ def test_custom_meta():
         )
         # test custom meta (queue, pool)
         for task in dag.tasks:
-            if task.task_id == "gen3.models.child2_v2_run":
+            if task.task_id == "gen2.models.child2_v2_run":
                 assert task.pool == "custom_pool"
             else:
                 assert task.pool == "default_pool"
 
-            if task.task_id == "gen3.models.child_run":
+            if task.task_id == "gen2.models.child_run":
                 assert task.queue == "custom_queue"
             else:
                 assert task.queue == "default"

tests/test_settings.py

@@ -11,6 +11,22 @@ def test_enable_cache_env_var():
     assert settings.enable_cache is False
 
 
+@patch.dict(
+    os.environ,
+    {"AIRFLOW__COSMOS__ENABLE_HIERARCHICAL_NAMING_FOR_GROUP_NODES_BY_FOLDER": "True"},
+    clear=True,
+)
+def test_enable_hierarchical_naming_for_group_nodes_by_folder_env_var():
+    reload(settings)
+    assert settings.enable_hierarchical_naming_for_group_nodes_by_folder is True
+
+
+@patch.dict(os.environ, {}, clear=True)
+def test_enable_hierarchical_naming_for_group_nodes_by_folder_defaults_to_false():
+    reload(settings)
+    assert settings.enable_hierarchical_naming_for_group_nodes_by_folder is False
+
+
 @patch.dict(os.environ, {"AIRFLOW__COSMOS__ENABLE_DEBUG_MODE": "True"}, clear=True)
 def test_enable_debug_mode_env_var():
     reload(settings)