Repository: GridTools/gt4py
Pull request #1922 (2025-03-19 15:44:02)

fix[next]: change ordering of dims so vertical dims go at the end
This PR modifies the default ordering of dimensions in a domain defined in #1847 to follow the most common ordering used by users, where horizontal dimensions go first and the vertical dimension goes at the end. See #1847 for more details.


Patch:
diff --git a/.python-version b/.default-python-version
similarity index 100%
rename from .python-version
rename to .default-python-version
diff --git a/.github/actions/get-python-versions/action.yml b/.github/actions/get-python-versions/action.yml
new file mode 100644
index 00000000..82d23d82
--- /dev/null
+++ b/.github/actions/get-python-versions/action.yml
@@ -0,0 +1,17 @@
+name: "Get Python Versions"
+description: "Reads Python versions from .python-versions file and outputs them as a JSON array"
+
+outputs:
+  python-versions:
+    description: "JSON array of Python versions"
+    value: ${{ steps.set-versions.outputs.python-versions }}
+
+runs:
+  using: "composite"
+  steps:
+  - uses: actions/checkout@v4
+  - id: set-versions
+    shell: bash
+    run: |
+      PYTHON_VERSIONS=$(cat .python-versions | jq -R -s -c 'split("\n") | map(select(length > 0))')
+      echo "python-versions=${PYTHON_VERSIONS}" >> $GITHUB_OUTPUT
diff --git a/.github/workflows/_disabled/documentation.yml b/.github/workflows/_disabled/documentation.yml
index 71361549..f0bcbecc 100644
--- a/.github/workflows/_disabled/documentation.yml
+++ b/.github/workflows/_disabled/documentation.yml
@@ -9,18 +9,31 @@ on:
     - functional
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.set-matrix.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: set-matrix
+      run: |
+        PYTHON_VERSIONS=$(cat .python-versions | jq -R -s -c 'split("\n") | map(select(length > 0))')
+        echo "python-versions=${PYTHON_VERSIONS}" >> $GITHUB_OUTPUT
+
   quickstart-guide:
+    needs: get-python-versions
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
 
     steps:
     - uses: actions/checkout@v3
-    - name: Set up Python ${{ matrix.python-version }}
+    - name: Set up Python
       uses: actions/setup-python@v4
       with:
-        python-version: ${{ matrix.python-version }}
+        python-version-file: ".default-python-version"
     - name: Install python dependencies
       run: |
         python -m pip install --upgrade pip setuptools wheel
diff --git a/.github/workflows/_disabled/gt4py-sphinx.yml b/.github/workflows/_disabled/gt4py-sphinx.yml
index 2533b2a4..2b071582 100644
--- a/.github/workflows/_disabled/gt4py-sphinx.yml
+++ b/.github/workflows/_disabled/gt4py-sphinx.yml
@@ -20,7 +20,7 @@ jobs:
     - name: Set up Python
       uses: actions/setup-python@v1
       with:
-        python-version: 3.10
+        python-version-file: ".default-python-version"
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
diff --git a/.github/workflows/_disabled/template_coverage.yml b/.github/workflows/_disabled/template_coverage.yml
index e592ca32..ec0aa436 100644
--- a/.github/workflows/_disabled/template_coverage.yml
+++ b/.github/workflows/_disabled/template_coverage.yml
@@ -16,14 +16,12 @@ on:
 jobs:
   upload:
     runs-on: ubuntu-latest
-    env:
-      PY_VERSION: "3.10"
     steps:
     - uses: actions/checkout@v3
-    - name: Set up Python ${{ env.PY_VERSION }}
+    - name: Set up Python
       uses: actions/setup-python@v4
       with:
-        python-version: ${{ env.PY_VERSION }}
+        python-version-file: ".default-python-version"
 
     - name: Install python dependencies
       shell: bash
diff --git a/.github/workflows/code-quality.yml b/.github/workflows/code-quality.yml
index 10bb537e..09e1ea0a 100644
--- a/.github/workflows/code-quality.yml
+++ b/.github/workflows/code-quality.yml
@@ -17,7 +17,7 @@ jobs:
     - name: Set up Python
       uses: actions/setup-python@v5
       with:
-        python-version-file: ".python-version"
+        python-version-file: ".default-python-version"
 
     - name: Install uv
       uses: astral-sh/setup-uv@v5
diff --git a/.github/workflows/daily-ci.yml b/.github/workflows/daily-ci.yml
index a2a52ce1..1497853e 100644
--- a/.github/workflows/daily-ci.yml
+++ b/.github/workflows/daily-ci.yml
@@ -13,7 +13,18 @@ on:
   ## END
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
   daily-ci:
+    needs: get-python-versions
     strategy:
       matrix:
         # dependencies-strategy -> The strategy that `uv lock` should use to select
@@ -22,7 +33,7 @@ jobs:
         dependencies-strategy: ["lowest-direct", "highest"]
         gt4py-module: ["cartesian", "eve", "next", "storage"]
         os: ["ubuntu-latest"] #, "macos-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
       fail-fast: false
 
     runs-on: ${{ matrix.os }}
diff --git a/.github/workflows/deploy-release.yml b/.github/workflows/deploy-release.yml
index 7a7505ca..ab7d0343 100644
--- a/.github/workflows/deploy-release.yml
+++ b/.github/workflows/deploy-release.yml
@@ -15,10 +15,10 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v4
-    - name: Set up Python 3.10
+    - name: Set up Python
       uses: actions/setup-python@v5
       with:
-        python-version: "3.10"
+        python-version-file: ".default-python-version"
     - name: Install pypa/build
       run: |
         python -m pip install build --user
diff --git a/.python-versions b/.python-versions
new file mode 100644
index 00000000..6db9dc12
--- /dev/null
+++ b/.python-versions
@@ -0,0 +1,2 @@
+3.10
+3.11
diff --git a/noxfile.py b/noxfile.py
index 3aad5658..5fad105c 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -111,7 +111,6 @@ def test_examples(session: nox.Session) -> None:
     session.run(*"jupytext docs/user/next/QuickstartGuide.md --to .ipynb".split())
     session.run(*"jupytext docs/user/next/advanced/*.md --to .ipynb".split())
 
-    num_processes = os.environ.get("NUM_PROCESSES", "auto")
     for notebook, extra_args in [
         ("docs/user/next/workshop/slides", None),
         ("docs/user/next/workshop/exercises", ["-k", "solutions"]),
@@ -120,7 +119,7 @@ def test_examples(session: nox.Session) -> None:
         ("examples", (None)),
     ]:
         session.run(
-            *f"pytest --nbmake {notebook} -sv -n {num_processes}".split(),
+            *f"pytest --nbmake {notebook} -sv -n 1 --benchmark-disable".split(),
             *(extra_args or []),
         )
 
diff --git a/src/gt4py/cartesian/backend/gtc_common.py b/src/gt4py/cartesian/backend/gtc_common.py
index 348e85de..91a56bff 100644
--- a/src/gt4py/cartesian/backend/gtc_common.py
+++ b/src/gt4py/cartesian/backend/gtc_common.py
@@ -296,7 +296,6 @@ class BaseGTBackend(gt_backend.BasePyExtBackend, gt_backend.CLIBackendMixin):
                 ),
                 add_profile_info=self.builder.options.backend_opts.get("add_profile_info", False),
                 uses_cuda=uses_cuda,
-                gt_version=2,
             ),
         )
 
diff --git a/src/gt4py/cartesian/backend/pyext_builder.py b/src/gt4py/cartesian/backend/pyext_builder.py
index 8875e3e3..937ed1dd 100644
--- a/src/gt4py/cartesian/backend/pyext_builder.py
+++ b/src/gt4py/cartesian/backend/pyext_builder.py
@@ -48,7 +48,6 @@ def get_gt_pyext_build_opts(
     add_profile_info: bool = False,
     uses_openmp: bool = True,
     uses_cuda: bool = False,
-    gt_version: int = 1,
 ) -> Dict[str, Union[str, List[str], Dict[str, Any]]]:
     include_dirs = [gt_config.build_settings["boost_include_path"]]
     extra_compile_args_from_config = gt_config.build_settings["extra_compile_args"]
diff --git a/src/gt4py/eve/visitors.py b/src/gt4py/eve/visitors.py
index 59b4ef08..6e559a26 100644
--- a/src/gt4py/eve/visitors.py
+++ b/src/gt4py/eve/visitors.py
@@ -121,6 +121,20 @@ class NodeVisitor:
         return None
 
 
+def _preserve_annex(
+    node: concepts.Node, new_node: concepts.Node, preserved_annex_attrs: tuple[str, ...]
+) -> None:
+    # access to `new_node.annex` implicitly creates the `__node_annex__` attribute in the property getter
+    new_annex_dict = new_node.annex.__dict__
+    old_annex_dict = node.annex.__dict__
+    for key in (old_annex_dict.keys() & preserved_annex_attrs) - new_annex_dict.keys():
+        # Note: The annex item's value of the new node might not be equal
+        # (in the sense that an equality comparison returns false), to
+        # the old one, but in the context of the pass, they are
+        # equivalent. Therefore, we don't assert equality here.
+        new_annex_dict[key] = old_annex_dict[key]
+
+
 class NodeTranslator(NodeVisitor):
     """Special `NodeVisitor` to translate nodes and trees.
 
@@ -158,13 +172,8 @@ class NodeTranslator(NodeVisitor):
                     if (new_child := self.visit(child, **kwargs)) is not NOTHING
                 }
             )
-            if self.PRESERVED_ANNEX_ATTRS and (old_annex := getattr(node, "__node_annex__", None)):
-                # access to `new_node.annex` implicitly creates the `__node_annex__` attribute in the property getter
-                new_annex_dict = new_node.annex.__dict__
-                for key in self.PRESERVED_ANNEX_ATTRS:
-                    if (value := getattr(old_annex, key, NOTHING)) is not NOTHING:
-                        assert key not in new_annex_dict
-                        new_annex_dict[key] = value
+            if self.PRESERVED_ANNEX_ATTRS and getattr(node, "__node_annex__", None):
+                _preserve_annex(node, new_node, self.PRESERVED_ANNEX_ATTRS)
 
             return new_node
 
@@ -189,3 +198,16 @@ class NodeTranslator(NodeVisitor):
             )
 
         return copy.deepcopy(node, memo=memo)
+
+    def visit(self, node: concepts.RootNode, **kwargs: Any) -> Any:
+        new_node = super().visit(node, **kwargs)
+
+        if (
+            isinstance(node, concepts.Node)
+            and isinstance(new_node, concepts.Node)
+            and self.PRESERVED_ANNEX_ATTRS
+            and getattr(node, "__node_annex__", None)
+        ):
+            _preserve_annex(node, new_node, self.PRESERVED_ANNEX_ATTRS)
+
+        return new_node
diff --git a/src/gt4py/next/common.py b/src/gt4py/next/common.py
index f6158330..d2bc090c 100644
--- a/src/gt4py/next/common.py
+++ b/src/gt4py/next/common.py
@@ -69,7 +69,7 @@ class DimensionKind(StrEnum):
         return self.value
 
 
-_DIM_KIND_ORDER = {DimensionKind.HORIZONTAL: 1, DimensionKind.VERTICAL: 2, DimensionKind.LOCAL: 3}
+_DIM_KIND_ORDER = {DimensionKind.HORIZONTAL: 0, DimensionKind.LOCAL: 1, DimensionKind.VERTICAL: 2}
 
 
 def dimension_to_implicit_offset(dim: str) -> str:
@@ -1015,7 +1015,21 @@ OffsetProvider: TypeAlias = Mapping[Tag, OffsetProviderElem]
 OffsetProviderType: TypeAlias = Mapping[Tag, OffsetProviderTypeElem]
 
 
-def offset_provider_to_type(offset_provider: OffsetProvider) -> OffsetProviderType:
+def is_offset_provider(obj: Any) -> TypeGuard[OffsetProvider]:
+    if not isinstance(obj, Mapping):
+        return False
+    return all(isinstance(el, OffsetProviderElem) for el in obj.values())
+
+
+def is_offset_provider_type(obj: Any) -> TypeGuard[OffsetProviderType]:
+    if not isinstance(obj, Mapping):
+        return False
+    return all(isinstance(el, OffsetProviderTypeElem) for el in obj.values())
+
+
+def offset_provider_to_type(
+    offset_provider: OffsetProvider | OffsetProviderType,
+) -> OffsetProviderType:
     return {
         k: v.__gt_type__() if isinstance(v, Connectivity) else v for k, v in offset_provider.items()
     }
@@ -1145,7 +1159,7 @@ def promote_dims(*dims_list: Sequence[Dimension]) -> list[Dimension]:
     Find an ordering of multiple lists of dimensions.
 
     The resulting list contains all unique dimensions from the input lists,
-    sorted first by dims_kind_order, i.e., `Dimension.kind` (`HORIZONTAL` < `VERTICAL` < `LOCAL`) and then
+    sorted first by dims_kind_order, i.e., `Dimension.kind` (`HORIZONTAL` < `LOCAL` < `VERTICAL`) and then
     lexicographically by `Dimension.value`.
 
     Examples:
@@ -1161,9 +1175,9 @@ def promote_dims(*dims_list: Sequence[Dimension]) -> list[Dimension]:
         Traceback (most recent call last):
         ...
         ValueError: Dimensions 'K[vertical], J[horizontal]' are not ordered correctly, expected 'J[horizontal], K[vertical]'.
-        >>> promote_dims([I, K], [J, E2V]) == [I, J, K, E2V]
+        >>> promote_dims([I, K], [J, E2V]) == [I, J, E2V, K]
         True
-        >>> promote_dims([I, E2C], [K, E2V])
+        >>> promote_dims([I, E2C], [E2V, K])
         Traceback (most recent call last):
         ...
         ValueError: There are more than one dimension with DimensionKind 'LOCAL'.
diff --git a/src/gt4py/next/ffront/foast_to_gtir.py b/src/gt4py/next/ffront/foast_to_gtir.py
index f884ec55..d76bc250 100644
--- a/src/gt4py/next/ffront/foast_to_gtir.py
+++ b/src/gt4py/next/ffront/foast_to_gtir.py
@@ -224,9 +224,18 @@ class FieldOperatorLowering(eve.PreserveLocationVisitor, eve.NodeTranslator):
     def visit_Symbol(self, node: foast.Symbol, **kwargs: Any) -> itir.Sym:
         return im.sym(node.id)
 
-    def visit_Name(self, node: foast.Name, **kwargs: Any) -> itir.SymRef:
+    def visit_Name(self, node: foast.Name, **kwargs: Any) -> itir.SymRef | itir.AxisLiteral:
+        if isinstance(node.type, ts.DimensionType):
+            return itir.AxisLiteral(value=node.type.dim.value, kind=node.type.dim.kind)
         return im.ref(node.id)
 
+    def visit_Attribute(self, node: foast.Attribute, **kwargs: Any) -> itir.AxisLiteral:
+        if isinstance(node.type, ts.DimensionType):
+            return itir.AxisLiteral(value=node.type.dim.value, kind=node.type.dim.kind)
+        raise AssertionError(
+            "Unexpected attribute access. At this point all accesses should have been removed by `ClosureVarFolding`."
+        )
+
     def visit_Subscript(self, node: foast.Subscript, **kwargs: Any) -> itir.Expr:
         return im.tuple_get(node.index, self.visit(node.value, **kwargs))
 
@@ -397,8 +406,7 @@ class FieldOperatorLowering(eve.PreserveLocationVisitor, eve.NodeTranslator):
     _visit_concat_where = _visit_where  # TODO(havogt): upgrade concat_where
 
     def _visit_broadcast(self, node: foast.Call, **kwargs: Any) -> itir.FunCall:
-        expr = self.visit(node.args[0], **kwargs)
-        return im.as_fieldop(im.ref("deref"))(expr)
+        return im.call("broadcast")(*self.visit(node.args, **kwargs))
 
     def _visit_math_built_in(self, node: foast.Call, **kwargs: Any) -> itir.FunCall:
         return self._lower_and_map(self.visit(node.func, **kwargs), *node.args)
diff --git a/src/gt4py/next/iterator/builtins.py b/src/gt4py/next/iterator/builtins.py
index 8e5f7add..9b065de6 100644
--- a/src/gt4py/next/iterator/builtins.py
+++ b/src/gt4py/next/iterator/builtins.py
@@ -27,6 +27,11 @@ def index(*args):
     raise BackendNotSelectedError()
 
 
+@builtin_dispatch
+def broadcast(*args):
+    raise BackendNotSelectedError()
+
+
 @builtin_dispatch
 def deref(*args):
     raise BackendNotSelectedError()
@@ -470,6 +475,7 @@ ARITHMETIC_BUILTINS = {
 
 BUILTINS = {
     "as_fieldop",  # `as_fieldop(stencil, domain)` creates field_operator from stencil (domain is optional, but for now required for embedded execution)
+    "broadcast",
     "can_deref",
     "cartesian_domain",
     "cast_",
diff --git a/src/gt4py/next/iterator/ir_utils/domain_utils.py b/src/gt4py/next/iterator/ir_utils/domain_utils.py
index 17df4f2e..5973715d 100644
--- a/src/gt4py/next/iterator/ir_utils/domain_utils.py
+++ b/src/gt4py/next/iterator/ir_utils/domain_utils.py
@@ -91,11 +91,13 @@ class SymbolicDomain:
             | Literal[trace_shifts.Sentinel.VALUE, trace_shifts.Sentinel.ALL_NEIGHBORS],
             ...,
         ],
-        offset_provider: common.OffsetProvider,
+        offset_provider: common.OffsetProvider | common.OffsetProviderType,
         #: A dictionary mapping axes names to their length. See
         #: func:`gt4py.next.iterator.transforms.infer_domain.infer_expr` for more details.
         symbolic_domain_sizes: Optional[dict[str, str]] = None,
     ) -> SymbolicDomain:
+        offset_provider_type = common.offset_provider_to_type(offset_provider)
+
         dims = list(self.ranges.keys())
         new_ranges = {dim: self.ranges[dim] for dim in dims}
         if len(shift) == 0:
@@ -103,17 +105,18 @@ class SymbolicDomain:
         if len(shift) == 2:
             off, val = shift
             assert isinstance(off, itir.OffsetLiteral) and isinstance(off.value, str)
-            nbt_provider = offset_provider[off.value]
-            if isinstance(nbt_provider, common.Dimension):
+            connectivity_type = offset_provider_type[off.value]
+
+            if isinstance(connectivity_type, common.Dimension):
                 if val is trace_shifts.Sentinel.VALUE:
                     raise NotImplementedError("Dynamic offsets not supported.")
                 assert isinstance(val, itir.OffsetLiteral) and isinstance(val.value, int)
-                current_dim = nbt_provider
+                current_dim = connectivity_type
                 # cartesian offset
                 new_ranges[current_dim] = SymbolicRange.translate(
                     self.ranges[current_dim], val.value
                 )
-            elif common.is_neighbor_connectivity(nbt_provider):
+            elif isinstance(connectivity_type, common.NeighborConnectivityType):
                 # unstructured shift
                 assert (
                     isinstance(val, itir.OffsetLiteral) and isinstance(val.value, int)
@@ -123,16 +126,19 @@ class SymbolicDomain:
                 ]
                 horizontal_sizes: dict[str, itir.Expr]
                 if symbolic_domain_sizes is not None:
-                    horizontal_sizes = {k: im.ref(v) for k, v in symbolic_domain_sizes.items()}
+                    horizontal_sizes = {
+                        k: im.ensure_expr(v) for k, v in symbolic_domain_sizes.items()
+                    }
                 else:
                     # note: ugly but cheap re-computation, but should disappear
+                    assert common.is_offset_provider(offset_provider)
                     horizontal_sizes = {
                         k: im.literal(str(v), builtins.INTEGER_INDEX_BUILTIN)
                         for k, v in _max_domain_sizes_by_location_type(offset_provider).items()
                     }
 
-                old_dim = nbt_provider.__gt_type__().source_dim
-                new_dim = nbt_provider.__gt_type__().codomain
+                old_dim = connectivity_type.source_dim
+                new_dim = connectivity_type.codomain
 
                 assert new_dim not in new_ranges or old_dim == new_dim
 
diff --git a/src/gt4py/next/iterator/ir_utils/misc.py b/src/gt4py/next/iterator/ir_utils/misc.py
index 03652cdf..d604d976 100644
--- a/src/gt4py/next/iterator/ir_utils/misc.py
+++ b/src/gt4py/next/iterator/ir_utils/misc.py
@@ -71,3 +71,10 @@ def is_equal(a: itir.Expr, b: itir.Expr):
     return a == b or (
         CannonicalizeBoundSymbolNames.apply(a) == CannonicalizeBoundSymbolNames.apply(b)
     )
+
+
+def with_altered_arg(node: itir.FunCall, arg_idx: int, new_arg: itir.Expr | str) -> itir.FunCall:
+    """Given a itir.FunCall return a new call with one of its argument replaced."""
+    return im.call(node.fun)(
+        *(arg if i != arg_idx else im.ensure_expr(new_arg) for i, arg in enumerate(node.args))
+    )
diff --git a/src/gt4py/next/iterator/transforms/collapse_tuple.py b/src/gt4py/next/iterator/transforms/collapse_tuple.py
index 462f87b6..7a10587d 100644
--- a/src/gt4py/next/iterator/transforms/collapse_tuple.py
+++ b/src/gt4py/next/iterator/transforms/collapse_tuple.py
@@ -29,14 +29,6 @@ from gt4py.next.iterator.type_system import inference as itir_type_inference
 from gt4py.next.type_system import type_info, type_specifications as ts
 
 
-def _with_altered_arg(node: ir.FunCall, arg_idx: int, new_arg: ir.Expr | str):
-    """Given a itir.FunCall return a new call with one of its argument replaced."""
-    return ir.FunCall(
-        fun=node.fun,
-        args=[arg if i != arg_idx else im.ensure_expr(new_arg) for i, arg in enumerate(node.args)],
-    )
-
-
 def _is_trivial_make_tuple_call(node: ir.Expr):
     """Return if node is a `make_tuple` call with all elements `SymRef`s, `Literal`s or tuples thereof."""
     if not cpm.is_call_to(node, "make_tuple"):
@@ -334,10 +326,10 @@ class CollapseTuple(
                 if cpm.is_call_to(arg, "if_"):
                     cond, true_branch, false_branch = arg.args
                     new_true_branch = self.fp_transform(
-                        _with_altered_arg(node, i, true_branch), **kwargs
+                        ir_misc.with_altered_arg(node, i, true_branch), **kwargs
                     )
                     new_false_branch = self.fp_transform(
-                        _with_altered_arg(node, i, false_branch), **kwargs
+                        ir_misc.with_altered_arg(node, i, false_branch), **kwargs
                     )
                     return im.if_(cond, new_true_branch, new_false_branch)
         return None
@@ -397,7 +389,7 @@ class CollapseTuple(
                     for type_ in tuple_type.types
                 ]
                 f_args = [im.ref(param.id, param.type) for param in f_params]
-                f_body = _with_altered_arg(node, i, im.make_tuple(*f_args))
+                f_body = ir_misc.with_altered_arg(node, i, im.make_tuple(*f_args))
                 # simplify, e.g., inline trivial make_tuple args
                 new_f_body = self.fp_transform(f_body, **kwargs)
                 # if the continuation did not simplify there is nothing to gain. Skip
diff --git a/src/gt4py/next/iterator/transforms/global_tmps.py b/src/gt4py/next/iterator/transforms/global_tmps.py
index ac7fcb8f..b7b36305 100644
--- a/src/gt4py/next/iterator/transforms/global_tmps.py
+++ b/src/gt4py/next/iterator/transforms/global_tmps.py
@@ -176,7 +176,10 @@ def _transform_stmt(
 
 def create_global_tmps(
     program: itir.Program,
-    offset_provider: common.OffsetProvider,
+    offset_provider: common.OffsetProvider | common.OffsetProviderType,
+    #: A dictionary mapping axes names to their length. See :func:`infer_domain.infer_expr` for
+    #: more details.
+    symbolic_domain_sizes: Optional[dict[str, str]] = None,
     *,
     uids: Optional[eve_utils.UIDGenerator] = None,
 ) -> itir.Program:
@@ -186,10 +189,11 @@ def create_global_tmps(
     This pass looks at all `as_fieldop` calls and transforms field-typed subexpressions of its
     arguments into temporaries.
     """
-    program = infer_domain.infer_program(program, offset_provider=offset_provider)
-    program = type_inference.infer(
-        program, offset_provider_type=common.offset_provider_to_type(offset_provider)
+    offset_provider_type = common.offset_provider_to_type(offset_provider)
+    program = infer_domain.infer_program(
+        program, offset_provider=offset_provider, symbolic_domain_sizes=symbolic_domain_sizes
     )
+    program = type_inference.infer(program, offset_provider_type=offset_provider_type)
 
     if not uids:
         uids = eve_utils.UIDGenerator(prefix="__tmp")
diff --git a/src/gt4py/next/iterator/transforms/infer_domain.py b/src/gt4py/next/iterator/transforms/infer_domain.py
index f3c31852..1f73d315 100644
--- a/src/gt4py/next/iterator/transforms/infer_domain.py
+++ b/src/gt4py/next/iterator/transforms/infer_domain.py
@@ -20,8 +20,9 @@ from gt4py.next.iterator.ir_utils import (
     common_pattern_matcher as cpm,
     domain_utils,
     ir_makers as im,
+    misc as ir_misc,
 )
-from gt4py.next.iterator.transforms import trace_shifts
+from gt4py.next.iterator.transforms import constant_folding, trace_shifts
 from gt4py.next.utils import flatten_nested_tuple, tree_map
 
 
@@ -51,7 +52,7 @@ AccessedDomains: TypeAlias = dict[str, DomainAccess]
 
 
 class InferenceOptions(typing.TypedDict):
-    offset_provider: common.OffsetProvider
+    offset_provider: common.OffsetProvider | common.OffsetProviderType
     symbolic_domain_sizes: Optional[dict[str, str]]
     allow_uninferred: bool
 
@@ -161,7 +162,7 @@ def _extract_accessed_domains(
     stencil: itir.Expr,
     input_ids: list[str],
     target_domain: NonTupleDomainAccess,
-    offset_provider: common.OffsetProvider,
+    offset_provider: common.OffsetProvider | common.OffsetProviderType,
     symbolic_domain_sizes: Optional[dict[str, str]],
 ) -> dict[str, NonTupleDomainAccess]:
     accessed_domains: dict[str, NonTupleDomainAccess] = {}
@@ -194,7 +195,7 @@ def _infer_as_fieldop(
     applied_fieldop: itir.FunCall,
     target_domain: DomainAccess,
     *,
-    offset_provider: common.OffsetProvider,
+    offset_provider: common.OffsetProvider | common.OffsetProviderType,
     symbolic_domain_sizes: Optional[dict[str, str]],
     allow_uninferred: bool,
 ) -> tuple[itir.FunCall, AccessedDomains]:
@@ -363,6 +364,19 @@ def _infer_if(
     return result_expr, actual_domains
 
 
+def _infer_broadcast(
+    expr: itir.Expr,
+    domain: DomainAccess,
+    **kwargs: Unpack[InferenceOptions],
+) -> tuple[itir.Expr, AccessedDomains]:
+    assert cpm.is_call_to(expr, "broadcast")
+    # We just propagate the domain to the first argument. Restriction of the domain is based
+    # on the type and occurs in a general setting (not yet merged #1853).
+    infered_expr, actual_domains = infer_expr(expr.args[0], domain, **kwargs)
+
+    return ir_misc.with_altered_arg(expr, 0, infered_expr), actual_domains
+
+
 def _infer_expr(
     expr: itir.Expr,
     domain: DomainAccess,
@@ -382,6 +396,8 @@ def _infer_expr(
         return _infer_tuple_get(expr, domain, **kwargs)
     elif cpm.is_call_to(expr, "if_"):
         return _infer_if(expr, domain, **kwargs)
+    elif cpm.is_call_to(expr, "broadcast"):
+        return _infer_broadcast(expr, domain, **kwargs)
     elif (
         cpm.is_call_to(expr, builtins.ARITHMETIC_BUILTINS)
         or cpm.is_call_to(expr, builtins.TYPE_BUILTINS)
@@ -396,7 +412,7 @@ def infer_expr(
     expr: itir.Expr,
     domain: DomainAccess,
     *,
-    offset_provider: common.OffsetProvider,
+    offset_provider: common.OffsetProvider | common.OffsetProviderType,
     symbolic_domain_sizes: Optional[dict[str, str]] = None,
     allow_uninferred: bool = False,
 ) -> tuple[itir.Expr, AccessedDomains]:
@@ -436,8 +452,12 @@ def _infer_stmt(
     **kwargs: Unpack[InferenceOptions],
 ):
     if isinstance(stmt, itir.SetAt):
+        # constant fold once otherwise constant folding after domain inference might create (syntactic) differences
+        # between the domain stored in IR and in the annex
+        domain = constant_folding.ConstantFolding.apply(stmt.domain)
+
         transformed_call, _ = infer_expr(
-            stmt.expr, domain_utils.SymbolicDomain.from_expr(stmt.domain), **kwargs
+            stmt.expr, domain_utils.SymbolicDomain.from_expr(domain), **kwargs
         )
 
         return itir.SetAt(
@@ -457,7 +477,7 @@ def _infer_stmt(
 def infer_program(
     program: itir.Program,
     *,
-    offset_provider: common.OffsetProvider,
+    offset_provider: common.OffsetProvider | common.OffsetProviderType,
     symbolic_domain_sizes: Optional[dict[str, str]] = None,
     allow_uninferred: bool = False,
 ) -> itir.Program:
diff --git a/src/gt4py/next/iterator/transforms/pass_manager.py b/src/gt4py/next/iterator/transforms/pass_manager.py
index 4023950d..809fbe9c 100644
--- a/src/gt4py/next/iterator/transforms/pass_manager.py
+++ b/src/gt4py/next/iterator/transforms/pass_manager.py
@@ -18,6 +18,7 @@ from gt4py.next.iterator.transforms import (
     inline_dynamic_shifts,
     inline_fundefs,
     inline_lifts,
+    prune_broadcast,
 )
 from gt4py.next.iterator.transforms.collapse_list_get import CollapseListGet
 from gt4py.next.iterator.transforms.collapse_tuple import CollapseTuple
@@ -43,7 +44,9 @@ class GTIRTransform(Protocol):
 def apply_common_transforms(
     ir: itir.Program,
     *,
-    offset_provider=None,  # TODO(havogt): should be replaced by offset_provider_type, but global_tmps currently relies on runtime info
+    # TODO(havogt): should be replaced by `common.OffsetProviderType`, but global_tmps currently
+    #  relies on runtime info or `symbolic_domain_sizes`.
+    offset_provider: common.OffsetProvider | common.OffsetProviderType,
     extract_temporaries=False,
     unroll_reduce=False,
     common_subexpression_elimination=True,
@@ -52,14 +55,11 @@ def apply_common_transforms(
     #: A dictionary mapping axes names to their length. See :func:`infer_domain.infer_expr` for
     #: more details.
     symbolic_domain_sizes: Optional[dict[str, str]] = None,
-    offset_provider_type: Optional[common.OffsetProviderType] = None,
 ) -> itir.Program:
-    # TODO(havogt): if the runtime `offset_provider` is not passed, we cannot run global_tmps
-    if offset_provider_type is None:
-        offset_provider_type = common.offset_provider_to_type(offset_provider)
-
     assert isinstance(ir, itir.Program)
 
+    offset_provider_type = common.offset_provider_to_type(offset_provider)
+
     tmp_uids = eve_utils.UIDGenerator(prefix="__tmp")
     mergeasfop_uids = eve_utils.UIDGenerator()
     collapse_tuple_uids = eve_utils.UIDGenerator()
@@ -92,6 +92,7 @@ def apply_common_transforms(
         offset_provider=offset_provider,
         symbolic_domain_sizes=symbolic_domain_sizes,
     )
+    ir = prune_broadcast.PruneBroadcast.apply(ir)
 
     for _ in range(10):
         inlined = ir
@@ -130,7 +131,12 @@ def apply_common_transforms(
 
     if extract_temporaries:
         ir = infer(ir, inplace=True, offset_provider_type=offset_provider_type)
-        ir = global_tmps.create_global_tmps(ir, offset_provider=offset_provider, uids=tmp_uids)
+        ir = global_tmps.create_global_tmps(
+            ir,
+            offset_provider=offset_provider,
+            symbolic_domain_sizes=symbolic_domain_sizes,
+            uids=tmp_uids,
+        )
 
     # Since `CollapseTuple` relies on the type inference which does not support returning tuples
     # larger than the number of closure outputs as given by the unconditional collapse, we can
@@ -187,4 +193,5 @@ def apply_fieldview_transforms(
         ir
     )  # domain inference does not support dynamic offsets yet
     ir = infer_domain.infer_program(ir, offset_provider=offset_provider)
+    ir = prune_broadcast.PruneBroadcast.apply(ir)
     return ir
diff --git a/src/gt4py/next/iterator/transforms/prune_broadcast.py b/src/gt4py/next/iterator/transforms/prune_broadcast.py
new file mode 100644
index 00000000..432e9d2d
--- /dev/null
+++ b/src/gt4py/next/iterator/transforms/prune_broadcast.py
@@ -0,0 +1,35 @@
+# GT4Py - GridTools Framework
+#
+# Copyright (c) 2014-2024, ETH Zurich
+# All rights reserved.
+#
+# Please, refer to the LICENSE file in the root directory.
+# SPDX-License-Identifier: BSD-3-Clause
+import dataclasses
+
+from gt4py.eve import NodeTranslator, PreserveLocationVisitor
+from gt4py.next.iterator import ir as itir
+from gt4py.next.iterator.ir_utils import (
+    common_pattern_matcher as cpm,
+    domain_utils,
+    ir_makers as im,
+)
+
+
+@dataclasses.dataclass
+class PruneBroadcast(PreserveLocationVisitor, NodeTranslator):
+    PRESERVED_ANNEX_ATTRS = ("domain",)
+
+    @classmethod
+    def apply(cls, node: itir.Node):
+        return cls().visit(node)
+
+    def visit_FunCall(self, node: itir.FunCall) -> itir.FunCall:
+        node = self.generic_visit(node)
+
+        if cpm.is_call_to(node, "broadcast"):
+            expr = self.visit(node.args[0])
+            node = im.as_fieldop("deref", domain_utils.SymbolicDomain.as_expr(node.annex.domain))(
+                expr
+            )
+        return node
diff --git a/src/gt4py/next/iterator/type_system/type_synthesizer.py b/src/gt4py/next/iterator/type_system/type_synthesizer.py
index 131b773d..df36189b 100644
--- a/src/gt4py/next/iterator/type_system/type_synthesizer.py
+++ b/src/gt4py/next/iterator/type_system/type_synthesizer.py
@@ -205,6 +205,25 @@ def index(arg: ts.DimensionType) -> ts.FieldType:
     )
 
 
+@_register_builtin_type_synthesizer
+def broadcast(
+    arg: ts.FieldType | ts.ScalarType | ts.DeferredType, dims: tuple[ts.DimensionType]
+) -> ts.FieldType | ts.DeferredType:
+    if isinstance(arg, ts.DeferredType):
+        return arg
+
+    dims_ = [dim.dim for dim in dims]
+
+    if isinstance(arg, ts.FieldType):
+        dtype = arg.dtype
+    elif isinstance(arg, ts.ScalarType):
+        dtype = arg
+    else:
+        raise AssertionError("`args` need to be either `ts.FieldType` or `ts.ScalarType`.")
+
+    return ts.FieldType(dims=dims_, dtype=dtype)
+
+
 @_register_builtin_type_synthesizer
 def neighbors(offset_literal: it_ts.OffsetLiteralType, it: it_ts.IteratorType) -> ts.ListType:
     assert (
diff --git a/src/gt4py/next/program_processors/codegens/gtfn/gtfn_module.py b/src/gt4py/next/program_processors/codegens/gtfn/gtfn_module.py
index 48f15acf..db0c1237 100644
--- a/src/gt4py/next/program_processors/codegens/gtfn/gtfn_module.py
+++ b/src/gt4py/next/program_processors/codegens/gtfn/gtfn_module.py
@@ -156,7 +156,7 @@ class GTFNTranslationStep(
     def _preprocess_program(
         self,
         program: itir.Program,
-        offset_provider: common.OffsetProvider,
+        offset_provider: common.OffsetProvider | common.OffsetProviderType,
     ) -> itir.Program:
         apply_common_transforms = functools.partial(
             pass_manager.apply_common_transforms,
@@ -184,7 +184,7 @@ class GTFNTranslationStep(
     def generate_stencil_source(
         self,
         program: itir.Program,
-        offset_provider: common.OffsetProvider,
+        offset_provider: common.OffsetProvider | common.OffsetProviderType,
         column_axis: Optional[common.Dimension],
     ) -> str:
         if self.enable_itir_transforms:
diff --git a/src/gt4py/next/program_processors/runners/dace/transformations/__init__.py b/src/gt4py/next/program_processors/runners/dace/transformations/__init__.py
index 61577048..57056a1f 100644
--- a/src/gt4py/next/program_processors/runners/dace/transformations/__init__.py
+++ b/src/gt4py/next/program_processors/runners/dace/transformations/__init__.py
@@ -27,6 +27,7 @@ from .map_promoter import SerialMapPromoter
 from .redundant_array_removers import (
     CopyChainRemover,
     MultiStateGlobalSelfCopyElimination,
+    SingleStateGlobalDirectSelfCopyElimination,
     SingleStateGlobalSelfCopyElimination,
     gt_multi_state_global_self_copy_elimination,
     gt_remove_copy_chain,
@@ -64,6 +65,7 @@ __all__ = [
     "MultiStateGlobalSelfCopyElimination",
     "SerialMapPromoter",
     "SerialMapPromoterGPU",
+    "SingleStateGlobalDirectSelfCopyElimination",
     "SingleStateGlobalSelfCopyElimination",
     "gt_auto_optimize",
     "gt_change_transient_strides",
diff --git a/src/gt4py/next/program_processors/runners/dace/transformations/redundant_array_removers.py b/src/gt4py/next/program_processors/runners/dace/transformations/redundant_array_removers.py
index 8919d2bc..00c0c3c8 100644
--- a/src/gt4py/next/program_processors/runners/dace/transformations/redundant_array_removers.py
+++ b/src/gt4py/next/program_processors/runners/dace/transformations/redundant_array_removers.py
@@ -477,14 +477,23 @@ class SingleStateGlobalSelfCopyElimination(dace_transformation.SingleStateTransf
     This transformation matches the following case `(G) -> (T) -> (G)`, i.e. `G`
     is read from and written too at the same time, however, in between is `T`
     used as a buffer. In the example above `G` is a global memory and `T` is a
-    temporary. This situation is generated by the lowering if the data node is
-    not needed (because the computation on it is only conditional).
-
-    In case `G` refers to global memory rule 3 of ADR-18 guarantees that we can
-    only have a point wise dependency of the output on the input.
-    This transformation will remove the write into `G`, i.e. we thus only have
-    `(G) -> (T)`. The read of `G` and the definition of `T`, will only be removed
-    if `T` is not used downstream. If it is used `T` will be maintained.
+    temporary. The transformation applies if the only incoming edge of the second
+    `G` AccessNode comes from the `T` node.
+    In case there are direct connections between the two `G` nodes, they will
+    be removed as well.
+
+    The transformation will then remove the second `G` node, all reads from the
+    second `G` node will redirected such that they use the first `G` node.
+    Furthermore, if `T` is not used downstream then also the `T` node will be
+    removed as well.
+
+    This transformation assumes that the SDFG follows rule 3 of ADR-18, which
+    guarantees that there is only a point wise dependency of the output on the
+    input.
+
+    Todo:
+        If `T` is only ready in this state, then `T` should not be created, instead
+        it should be read directly from `G`.
     """
 
     node_read_g = dace_transformation.PatternNode(dace_nodes.AccessNode)
@@ -522,17 +531,21 @@ class SingleStateGlobalSelfCopyElimination(dace_transformation.SingleStateTransf
             return False
         if not tmp_desc.transient:
             return False
-        if graph.in_degree(read_g) != 0:
-            return False
-        if graph.out_degree(read_g) != 1:
-            return False
-        if graph.degree(tmp_node) != 2:
-            return False
-        if graph.in_degree(write_g) != 1:
+        if graph.scope_dict()[read_g] is not None:
             return False
-        if graph.out_degree(write_g) != 0:
+
+        # For ease of implementation we require that only `G` defines `T`. This is
+        #  needed to ensure that `G` only depends on `G`.
+        if graph.in_degree(tmp_node) != 1:
             return False
-        if graph.scope_dict()[read_g] is not None:
+
+        # Now we look at all incoming connections of the second `G` nodes, they must
+        #  either come from `T` or from the first `G` node directly.
+        for iedge in graph.in_edges(write_g):
+            if iedge.src is read_g:
+                continue
+            if iedge.src is tmp_node:
+                continue
             return False
 
         return True
@@ -582,21 +595,44 @@ class SingleStateGlobalSelfCopyElimination(dace_transformation.SingleStateTransf
         write_g: dace_nodes.AccessNode = self.node_write_g
         tmp_node: dace_nodes.AccessNode = self.node_tmp
 
-        # We first check if `T`, the intermediate is not used downstream. In this
-        #  case we can remove the read to `G` and `T` itself from the SDFG.
-        #  We have to do this check before, because the matching is not fully stable.
-        is_tmp_used_downstream = self._is_read_downstream(
-            start_state=graph, sdfg=sdfg, data_to_look=tmp_node.data
-        )
+        # First check if the `T` node is still needed. The node is needed if the data
+        #  is referenced downstream or if there are edges other than to the second
+        #  `G` node.
+        if not all(oedge.dst is write_g for oedge in graph.out_edges(tmp_node)):
+            tmp_node_is_still_needed = True
+        elif self._is_read_downstream(start_state=graph, sdfg=sdfg, data_to_look=tmp_node.data):
+            tmp_node_is_still_needed = True
+        else:
+            tmp_node_is_still_needed = False
+
+        # Redirect the reads from the second `G` node such that they now go through
+        #  the first `G` node. Because the names are the same there is no special
+        #  handling needed.
+        # NOTE: There are no writes to the second `G` nodes beside directly the one
+        #  coming from the first `G` node or the `T` node. Thus nothing to do.
+        for wg_oedge in list(graph.out_edges(write_g)):
+            graph.add_edge(
+                read_g,
+                wg_oedge.src_conn,
+                wg_oedge.dst,
+                wg_oedge.dst_conn,
+                wg_oedge.data,
+            )
+            graph.remove_edge(wg_oedge)
 
-        # The write to `G` can always be removed.
+        # Now we remove the second `G` node as it is no longer needed.
         graph.remove_node(write_g)
 
-        # Also remove the read to `G` and `T` from the SDFG if possible.
-        if not is_tmp_used_downstream:
-            graph.remove_node(read_g)
+        # If the `T` is no longer needed then remove it. If the first node has become
+        #  isolated, also remove it.
+        if not tmp_node_is_still_needed:
             graph.remove_node(tmp_node)
-            # It could still be used in a parallel branch.
+
+            if graph.degree(read_g) == 0:
+                graph.remove_node(read_g)
+
+            # If it fails then `T` is still used in a parallel controlflow path, for
+            #  example in `if` branches.
             try:
                 sdfg.remove_data(tmp_node.data, validate=True)
             except ValueError as e:
@@ -604,6 +640,156 @@ class SingleStateGlobalSelfCopyElimination(dace_transformation.SingleStateTransf
                     raise
 
 
+@dace_properties.make_properties
+class SingleStateGlobalDirectSelfCopyElimination(dace_transformation.SingleStateTransformation):
+    """Removes an unbuffered self copy in a state.
+
+    This transformation is extremely similar to `SingleStateGlobalSelfCopyElimination`,
+    however, this transformation does not need a buffer between the two global
+    AccessNodes. Therefore it matches the pattern `(G) -> (G)`, where `G` refers to
+    global data. Note that the transformation only considers copies on global scope.
+    The transformation has two modes in how this is achieved, which one is chosen,
+    depends on if cycles are created or not. The first mode, which is selected if
+    cycles would be created, is to merge the two AccessNodes together. For this
+    the edges of the first AccessNode are reconnected to the second AccessNode.
+
+    In the second mode, which is selected if cycles would be created, for example in
+    case of `if` branches, then the edges are removed, which will create two
+    independent `G` AccessNodes.
+
+    Todo:
+        Merge this transformation with `SingleStateGlobalSelfCopyElimination`.
+    """
+
+    node_read_g = dace_transformation.PatternNode(dace_nodes.AccessNode)
+    node_write_g = dace_transformation.PatternNode(dace_nodes.AccessNode)
+
+    def __init__(
+        self,
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
+
+    @classmethod
+    def expressions(cls) -> Any:
+        return [dace.sdfg.utils.node_path_graph(cls.node_read_g, cls.node_write_g)]
+
+    def can_be_applied(
+        self,
+        graph: dace.SDFGState | dace.SDFG,
+        expr_index: int,
+        sdfg: dace.SDFG,
+        permissive: bool = False,
+    ) -> bool:
+        read_g = self.node_read_g
+        write_g = self.node_write_g
+        g_desc = read_g.desc(sdfg)
+
+        # NOTE: We do not check if `G` is read downstream.
+        if read_g.data != write_g.data:
+            return False
+        if g_desc.transient:
+            return False
+        # Restrict to global scope.
+        if graph.scope_dict()[read_g] is not None:
+            return False
+
+        return True
+
+    def apply(
+        self,
+        graph: dace.SDFGState | dace.SDFG,
+        sdfg: dace.SDFG,
+    ) -> None:
+        read_g: dace_nodes.AccessNode = self.node_read_g
+        write_g: dace_nodes.AccessNode = self.node_write_g
+
+        # Select in which mode the transformation should run, depending on
+        #  if this would lead to cycles.
+        merge_g_nodes = True
+        for oedge in graph.out_edges(read_g):
+            if oedge.dst is write_g:
+                continue
+            if gtx_transformations.utils.is_reachable(
+                start=oedge.dst,
+                target=write_g,
+                state=graph,
+            ):
+                merge_g_nodes = False
+                break
+
+        if merge_g_nodes:
+            self._merge_g_nodes(state=graph, sdfg=sdfg)
+        else:
+            self._split_g_nodes(state=graph)
+
+    def _split_g_nodes(
+        self,
+        state: dace.SDFGState,
+    ) -> None:
+        """Second mode of the transformation, i.e. the splitting.
+
+        All edges between the two nodes, will be removed. There is no longer a
+        direct connection between the two nodes.
+        """
+        read_g: dace_nodes.AccessNode = self.node_read_g
+        write_g: dace_nodes.AccessNode = self.node_write_g
+
+        for oedge in list(state.out_edges(read_g)):
+            if oedge.dst is write_g:
+                state.remove_edge(oedge)
+
+        assert state.degree(read_g) != 0
+        assert state.degree(write_g) != 0
+
+    def _merge_g_nodes(
+        self,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> None:
+        """The first mode of the transformation, i.e. the merging of the nodes.
+
+        All incoming edges of the first node will be reconnected to the second node.
+        The outgoing edges will be reconnected to the second node, if they connect
+        the two nodes, they will be removed.
+        """
+        read_g: dace_nodes.AccessNode = self.node_read_g
+        write_g: dace_nodes.AccessNode = self.node_write_g
+
+        # Reconnect any incoming edges of the read AccessNode, aka. the first one.
+        for iedge in list(state.in_edges(read_g)):
+            # Since the names are the same, just creating a new edge is enough.
+            state.add_edge(
+                iedge.src,
+                iedge.src_conn,
+                write_g,
+                iedge.dst_conn,
+                iedge.data,
+            )
+            state.remove_edge(iedge)
+
+        # Now reconnect all edges that leave the first AccessNodes, with the
+        #  exception of the edges that connecting the two, they are just removed.
+        for oedge in list(state.out_edges(read_g)):
+            if oedge.dst is not write_g:
+                state.add_edge(
+                    write_g,
+                    oedge.src_conn,
+                    oedge.dst,
+                    oedge.dst_conn,
+                    oedge.data,
+                )
+            state.remove_edge(oedge)
+
+        # Now remove the first node to `G` and if the second has become isolated
+        #  then also remove it.
+        state.remove_node(read_g)
+
+        if state.degree(write_g) == 0:
+            state.remove_node(write_g)
+
+
 @dace_properties.make_properties
 class CopyChainRemover(dace_transformation.SingleStateTransformation):
     """Removes chain of redundant copies, mostly related to `concat_where`.
@@ -613,16 +799,23 @@ class CopyChainRemover(dace_transformation.SingleStateTransformation):
     fix point is reached and should be seen as an addition to the array removal passes
     that ship with DaCe.
     The transformation will look for the pattern `(A1) -> (A2)`, i.e. a data container
-    is copied into another one. The transformation will then remove `A1` and rewire
-    the edges such that they now refer to `A2`. Another, and probably better way, is to
-    consider the transformation as fusion transformation for AccessNodes.
+    is copied into another one, at the global scope. The transformation will then
+    remove `A1` and rewire the edges such that they now refer to `A2`. Another, and
+    probably better way, is to consider the transformation as fusion transformation
+    for AccessNodes.
 
     The transformation builds on ADR-18 and imposes the following additional
     requirements before it can be applied:
     - Through the merging of `A1` and `A2` no cycles are created.
     - `A1` can not be used anywhere else.
-    - `A1` is fully read by `A2`.
     - `A1` is a transient and must have the same dimensionality than `A2`.
+    - `A1` is fully read by `A2`.
+
+    In certain cases the last rule can not be verified, an alternative formulation,
+    which is a consequence of the lowering and domain interference, will be checked:
+    - `A1` is read from the beginning, i.e. all subsets starts at literal `0`.
+    - `A1` has only one going edge.
+    - `A2` is global memory.
 
     Notes:
         - The transformation assumes that the domain inference adjusted the ranges of
@@ -637,7 +830,6 @@ class CopyChainRemover(dace_transformation.SingleStateTransformation):
     Todo:
         - Extend such that not the full array must be read.
         - Try to allow more than one connection between `A1` and `A2`.
-        - Modify it such that also `A2` can be removed.
     """
 
     node_a1 = dace_transformation.PatternNode(dace_nodes.AccessNode)
@@ -676,6 +868,10 @@ class CopyChainRemover(dace_transformation.SingleStateTransformation):
         a1: dace_nodes.AccessNode = self.node_a1
         a2: dace_nodes.AccessNode = self.node_a2
 
+        # We only allow that we operate on the top level scope.
+        if graph.scope_dict()[a1] is not None:
+            return False
+
         a1_desc = a1.desc(sdfg)
         a2_desc = a2.desc(sdfg)
 
@@ -696,10 +892,6 @@ class CopyChainRemover(dace_transformation.SingleStateTransformation):
         if gtx_transformations.utils.is_view(a2_desc, None):
             return False
 
-        # We only allow that we operate on the top level scope.
-        if graph.scope_dict()[a1] is not None:
-            return False
-
         # TODO(phimuell): Relax this to only prevent host-device copies.
         if a1_desc.storage != a2_desc.storage:
             return False
@@ -731,10 +923,25 @@ class CopyChainRemover(dace_transformation.SingleStateTransformation):
         if dst_subset is None:
             return False
 
+        # Checking if the whole array is read.
+        #  As described in the description of the class there are two different
+        #  formulation of this rule, either the simple one or one that proves this
+        #  indirectly.
         # NOTE: The main benefit of requiring that the whole array is read is
         #  that we do not have to adjust maps.
         a1_range = dace_sbs.Range.from_array(a1_desc)
-        if not src_subset.covers(a1_range):
+        if src_subset.covers(a1_range):
+            pass
+        elif all(ss.is_constant() for ss in src_subset.size()):
+            # If the subset is fully known then we require that it is covered.
+            return False
+        elif (
+            (not a2_desc.transient)
+            and (graph.out_degree(a1) == 1)
+            and all(ss_start == 0 for ss_start in src_subset.min_element())
+        ):
+            pass
+        else:
             return False
 
         # We have to ensure that no cycle is created through the removal of `a1`.
diff --git a/src/gt4py/next/program_processors/runners/dace/transformations/simplify.py b/src/gt4py/next/program_processors/runners/dace/transformations/simplify.py
index c2c5acf0..c9608010 100644
--- a/src/gt4py/next/program_processors/runners/dace/transformations/simplify.py
+++ b/src/gt4py/next/program_processors/runners/dace/transformations/simplify.py
@@ -59,6 +59,8 @@ def gt_simplify(
     Further, the function will run the following passes in addition to DaCe simplify:
     - `SingleStateGlobalSelfCopyElimination`: Special copy pattern that in the context
         of GT4Py based SDFG behaves as a no op, i.e. `(G) -> (T) -> (G)`.
+    - `SingleStateGlobalDirectSelfCopyElimination`: Special copy pattern of the form
+        `(G) -> (G)` which can always be eliminated.
     - `MultiStateGlobalSelfCopyElimination`: Very similar to
         `SingleStateGlobalSelfCopyElimination`, with the exception that the write to
         `T`, i.e. `(G) -> (T)` and the write back to `G`, i.e. `(T) -> (G)` might be
@@ -149,6 +151,21 @@ def gt_simplify(
                     result["CopyChainRemover"] = 0
                 result["CopyChainRemover"] += copy_chain_remover_result
 
+        if "SingleStateGlobalDirectSelfCopyElimination" not in skip:
+            direct_self_copy_removal_result = sdfg.apply_transformations_repeated(
+                gtx_transformations.SingleStateGlobalDirectSelfCopyElimination(),
+                validate=validate,
+                validate_all=validate_all,
+            )
+            if direct_self_copy_removal_result > 0:
+                at_least_one_xtrans_run = True
+                result = result or {}
+                if "SingleStateGlobalDirectSelfCopyElimination" not in result:
+                    result["SingleStateGlobalDirectSelfCopyElimination"] = 0
+                result["SingleStateGlobalDirectSelfCopyElimination"] += (
+                    direct_self_copy_removal_result
+                )
+
         if "SingleStateGlobalSelfCopyElimination" not in skip:
             self_copy_removal_result = sdfg.apply_transformations_repeated(
                 gtx_transformations.SingleStateGlobalSelfCopyElimination(),


Test patch:
diff --git a/.github/workflows/test-cartesian-fallback.yml b/.github/workflows/test-cartesian-fallback.yml
index a846af2e..1cce8831 100644
--- a/.github/workflows/test-cartesian-fallback.yml
+++ b/.github/workflows/test-cartesian-fallback.yml
@@ -12,12 +12,23 @@ on:
     - "*.rst"
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
   test-cartesian:
+    needs: get-python-versions
     strategy:
       matrix:
         codegen-factor: [internal, dace]
         os: ["ubuntu-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
     runs-on: ${{ matrix.os }}
     steps:
     - run: 'echo "No build required"'
diff --git a/.github/workflows/test-cartesian.yml b/.github/workflows/test-cartesian.yml
index ea6b7940..5c7e2e90 100644
--- a/.github/workflows/test-cartesian.yml
+++ b/.github/workflows/test-cartesian.yml
@@ -19,12 +19,24 @@ concurrency:
   cancel-in-progress: true
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
+  # Main job that runs the tests
   test-cartesian:
+    needs: get-python-versions
     strategy:
       matrix:
         codegen-factor: [internal, dace]
         os: ["ubuntu-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
       fail-fast: false
 
     runs-on: ${{ matrix.os }}
diff --git a/.github/workflows/test-eve-fallback.yml b/.github/workflows/test-eve-fallback.yml
index f3dbb58a..94e870dc 100644
--- a/.github/workflows/test-eve-fallback.yml
+++ b/.github/workflows/test-eve-fallback.yml
@@ -16,11 +16,22 @@ on:
     - "*.yml"
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
   test-eve:
+    needs: get-python-versions
     strategy:
       matrix:
         os: ["ubuntu-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
 
     runs-on: ${{ matrix.os }}
     steps:
diff --git a/.github/workflows/test-eve.yml b/.github/workflows/test-eve.yml
index aad3971a..76bcc972 100644
--- a/.github/workflows/test-eve.yml
+++ b/.github/workflows/test-eve.yml
@@ -16,11 +16,22 @@ on:
     - "*.yml"
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
   test-eve:
+    needs: get-python-versions
     strategy:
       matrix:
         os: ["ubuntu-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
       fail-fast: false
 
     runs-on: ${{ matrix.os }}
diff --git a/.github/workflows/test-examples.yml b/.github/workflows/test-examples.yml
index 836af45d..93c703c2 100644
--- a/.github/workflows/test-examples.yml
+++ b/.github/workflows/test-examples.yml
@@ -9,11 +9,22 @@ on:
     - main
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
   test-notebooks:
+    needs: get-python-versions
     strategy:
       matrix:
         os: ["ubuntu-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
       fail-fast: false
 
     runs-on: ${{ matrix.os }}
diff --git a/.github/workflows/test-next-fallback.yml b/.github/workflows/test-next-fallback.yml
index ef8be3df..68b3e052 100644
--- a/.github/workflows/test-next-fallback.yml
+++ b/.github/workflows/test-next-fallback.yml
@@ -12,13 +12,24 @@ on:
     - "*.rst"
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
   test-next:
+    needs: get-python-versions
     strategy:
       matrix:
         codegen-factor: [internal, dace]
         mesh-factor: [nomesh, atlas]
         os: ["ubuntu-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
 
     runs-on: ${{ matrix.os }}
     steps:
diff --git a/.github/workflows/test-next.yml b/.github/workflows/test-next.yml
index 068377c6..68d86e5a 100644
--- a/.github/workflows/test-next.yml
+++ b/.github/workflows/test-next.yml
@@ -15,13 +15,24 @@ on:
     - "*.rst"
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
   test-next:
+    needs: get-python-versions
     strategy:
       matrix:
         codegen-factor: [internal, dace]
         mesh-factor: [nomesh, atlas]
         os: ["ubuntu-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
       fail-fast: false
 
     runs-on: ${{ matrix.os }}
diff --git a/.github/workflows/test-storage-fallback.yml b/.github/workflows/test-storage-fallback.yml
index c913529a..11739104 100644
--- a/.github/workflows/test-storage-fallback.yml
+++ b/.github/workflows/test-storage-fallback.yml
@@ -17,11 +17,22 @@ on:
     - "*.yml"
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
   test-storage:
+    needs: get-python-versions
     strategy:
       matrix:
         os: ["ubuntu-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
 
     runs-on: ${{ matrix.os }}
     steps:
diff --git a/.github/workflows/test-storage.yml b/.github/workflows/test-storage.yml
index b2bb09df..4abedbfb 100644
--- a/.github/workflows/test-storage.yml
+++ b/.github/workflows/test-storage.yml
@@ -17,11 +17,22 @@ on:
     - "*.yml"
 
 jobs:
+  # First job to read Python versions from .python-versions file
+  get-python-versions:
+    runs-on: ubuntu-latest
+    outputs:
+      python-versions: ${{ steps.get-versions.outputs.python-versions }}
+    steps:
+    - uses: actions/checkout@v4
+    - id: get-versions
+      uses: ./.github/actions/get-python-versions
+
   test-storage:
+    needs: get-python-versions
     strategy:
       matrix:
         os: ["ubuntu-latest"]
-        python-version: ["3.10", "3.11"]
+        python-version: ${{ fromJson(needs.get-python-versions.outputs.python-versions) }}
       fail-fast: false
 
     runs-on: ${{ matrix.os }}
diff --git a/tests/eve_tests/unit_tests/test_visitors.py b/tests/eve_tests/unit_tests/test_visitors.py
index 2eff41d7..c59fa167 100644
--- a/tests/eve_tests/unit_tests/test_visitors.py
+++ b/tests/eve_tests/unit_tests/test_visitors.py
@@ -24,3 +24,38 @@ def test_annex_preservation(compound_node: eve.Node):
     assert translated_node.annex.foo == 1
     assert translated_node.annex.bar is None
     assert not hasattr(translated_node.annex, "baz")
+
+
+def test_annex_preservation_translated_node(compound_node: eve.Node):
+    compound_node.annex.foo = 1
+    compound_node.annex.baz = 2
+
+    class SampleTranslator(eve.NodeTranslator):
+        PRESERVED_ANNEX_ATTRS = ("foo",)
+
+        def visit_Node(self, node: eve.Node):
+            # just return an empty node, we care about the annex only anyway
+            return eve.Node()
+
+    translated_node = SampleTranslator().visit(compound_node)
+
+    assert translated_node.annex.foo == 1
+    assert not hasattr(translated_node.annex, "baz")
+
+
+def test_annex_preservation_translated_node_overwritten(compound_node: eve.Node):
+    compound_node.annex.foo = "1+1"
+
+    class SampleTranslator(eve.NodeTranslator):
+        PRESERVED_ANNEX_ATTRS = ("foo",)
+
+        def visit_Node(self, node: eve.Node):
+            # just return an empty node, we care about the annex only anyway
+            new_node = eve.Node()
+            # the annex value is different, but considered equivalent by this pass
+            new_node.annex.foo = "2"
+            return new_node
+
+    translated_node = SampleTranslator().visit(compound_node)
+
+    assert translated_node.annex.foo == "2"
diff --git a/tests/next_tests/unit_tests/ffront_tests/test_foast_to_gtir.py b/tests/next_tests/unit_tests/ffront_tests/test_foast_to_gtir.py
index 776cd4e1..53661427 100644
--- a/tests/next_tests/unit_tests/ffront_tests/test_foast_to_gtir.py
+++ b/tests/next_tests/unit_tests/ffront_tests/test_foast_to_gtir.py
@@ -38,7 +38,7 @@ from gt4py.next.ffront.func_to_foast import FieldOperatorParser
 from gt4py.next.iterator.ir_utils import ir_makers as im
 from gt4py.next.iterator.transforms import inline_lambdas
 from gt4py.next.type_system import type_specifications as ts, type_translation
-
+from gt4py.next.iterator import ir as itir
 
 Edge = gtx.Dimension("Edge")
 Vertex = gtx.Dimension("Vertex")
@@ -907,7 +907,10 @@ def test_broadcast():
     lowered = FieldOperatorLowering.apply(parsed)
 
     assert lowered.id == "foo"
-    assert lowered.expr == im.as_fieldop("deref")(im.ref("inp"))
+    assert lowered.expr == im.call("broadcast")(
+        im.ref("inp"),
+        im.make_tuple(*(itir.AxisLiteral(value=dim.value, kind=dim.kind) for dim in (TDim, UDim))),
+    )
 
 
 def test_scalar_broadcast():
@@ -918,4 +921,7 @@ def test_scalar_broadcast():
     lowered = FieldOperatorLowering.apply(parsed)
 
     assert lowered.id == "foo"
-    assert lowered.expr == im.as_fieldop("deref")(1)
+    assert lowered.expr == im.call("broadcast")(
+        1,
+        im.make_tuple(*(itir.AxisLiteral(value=dim.value, kind=dim.kind) for dim in (TDim, UDim))),
+    )
diff --git a/tests/next_tests/unit_tests/ffront_tests/test_type_deduction.py b/tests/next_tests/unit_tests/ffront_tests/test_type_deduction.py
index f9393bd9..d3396a54 100644
--- a/tests/next_tests/unit_tests/ffront_tests/test_type_deduction.py
+++ b/tests/next_tests/unit_tests/ffront_tests/test_type_deduction.py
@@ -163,6 +163,20 @@ def test_premap_nbfield(premap_setup):
     )
 
 
+def test_premap_nbfield_with_vertical(premap_setup):
+    X, Y, Y2XDim, Y2X = premap_setup
+    K = Dimension("K", kind=DimensionKind.VERTICAL)
+
+    def premap_fo(bar: Field[[X, K], int64]) -> Field[[Y, Y2XDim, K], int64]:
+        return bar(Y2X)
+
+    parsed = FieldOperatorParser.apply_to_function(premap_fo)
+
+    assert parsed.body.stmts[0].value.type == ts.FieldType(
+        dims=[Y, Y2XDim, K], dtype=ts.ScalarType(kind=ts.ScalarKind.INT64)
+    )
+
+
 def test_premap_reduce(premap_setup):
     X, Y, Y2XDim, Y2X = premap_setup
 
diff --git a/tests/next_tests/unit_tests/iterator_tests/transforms_tests/test_domain_inference.py b/tests/next_tests/unit_tests/iterator_tests/transforms_tests/test_domain_inference.py
index 86cc8a67..4385cbe7 100644
--- a/tests/next_tests/unit_tests/iterator_tests/transforms_tests/test_domain_inference.py
+++ b/tests/next_tests/unit_tests/iterator_tests/transforms_tests/test_domain_inference.py
@@ -1093,3 +1093,12 @@ def test_never_accessed_domain_tuple(offset_provider):
         "in_field2": infer_domain.DomainAccessDescriptor.NEVER,
     }
     run_test_expr(testee, testee, domain, expected_domains, offset_provider)
+
+
+def test_broadcast(offset_provider):
+    testee = im.call("broadcast")("in_field", im.make_tuple(itir.AxisLiteral(value="IDim")))
+    domain = im.domain(common.GridType.CARTESIAN, {IDim: (0, 10)})
+    expected_domains = {
+        "in_field": {IDim: (0, 10)},
+    }
+    run_test_expr(testee, testee, domain, expected_domains, offset_provider)
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/transformation_tests/test_copy_chain_remover.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/transformation_tests/test_copy_chain_remover.py
index 8251352e..da0ced4e 100644
--- a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/transformation_tests/test_copy_chain_remover.py
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/transformation_tests/test_copy_chain_remover.py
@@ -430,6 +430,21 @@ def test_multi_stage_reduction():
 def test_not_fully_copied():
     sdfg = _make_not_fully_copied()
 
+    # NOTE: In the unprocessed SDFG, especially in the `(d) -> (e)` Memlet.
+    #  we read uninitialized memory from `d`, because we only write into `[0:8]`
+    #  but read from it in the range `[0:10]`, thus the range `[8:10]` in `e`
+    #  has undefined value. We have to take that into account.
+    ref = {
+        "a": np.array(np.random.rand(10), dtype=np.float64, copy=True),
+        "e": np.array(np.random.rand(10), dtype=np.float64, copy=True),
+    }
+    res = copy.deepcopy(ref)
+    org = copy.deepcopy(ref)
+
+    # Compile and run the original SDFG
+    csdfg_ref = sdfg.compile()
+    csdfg_ref(**ref)
+
     # Apply the transformation.
     #  It will only remove `d` all the others are retained, because they are not read
     #  correctly, i.e. fully.
@@ -443,6 +458,15 @@ def test_not_fully_copied():
     assert nb_applies == 1
     assert "d" not in acnodes
 
+    # Now run the test and compare the results, see above for the ranges.
+    csdfg_res = sdfg.compile()
+    csdfg_res(**res)
+
+    assert np.all(ref["a"] == res["a"])
+    assert np.all(org["a"] == res["a"])
+    assert np.all(res["e"][0:8] == ref["e"][0:8])
+    assert np.all(res["e"][8:10] == org["e"][8:10])
+
 
 def test_possible_cyclic_sdfg():
     sdfg = _make_possible_cyclic_sdfg()
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/transformation_tests/test_single_state_global_self_copy_elimination.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/transformation_tests/test_single_state_global_self_copy_elimination.py
index 22641021..043a9b9e 100644
--- a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/transformation_tests/test_single_state_global_self_copy_elimination.py
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/transformation_tests/test_single_state_global_self_copy_elimination.py
@@ -11,6 +11,8 @@ import pytest
 dace = pytest.importorskip("dace")
 from dace.sdfg import nodes as dace_nodes
 
+import numpy as np
+
 from gt4py.next.program_processors.runners.dace import (
     transformations as gtx_transformations,
 )
@@ -40,6 +42,75 @@ def _make_self_copy_sdfg() -> tuple[dace.SDFG, dace.SDFGState]:
     return sdfg, state
 
 
+def _make_direct_self_copy_elimination_used_sdfg() -> dace.SDFG:
+    sdfg = dace.SDFG(util.unique_name("direct_self_copy_elimination_used"))
+    state = sdfg.add_state(is_start_block=True)
+
+    for name in "ABCG":
+        sdfg.add_array(
+            name,
+            shape=(20,),
+            dtype=dace.float64,
+            transient=False,
+        )
+
+    g_read = state.add_access("G")
+    g_write = state.add_access("G")
+
+    state.add_nedge(state.add_access("A"), g_read, dace.Memlet("A[0:5] -> [0:5]"))
+    state.add_nedge(g_read, state.add_access("B"), dace.Memlet("G[5:10] -> [5:10]"))
+    state.add_nedge(g_write, state.add_access("C"), dace.Memlet("G[10:15] -> [10:15]"))
+
+    # It might look a bit strange (and might not be correct in general), but in
+    #  GT4Py there is no requirement that the whole array is self copied.
+    state.add_nedge(g_read, g_write, dace.Memlet("G[1:19] -> [1:19]"))
+    sdfg.validate()
+
+    return sdfg
+
+
+def _make_self_copy_sdfg_with_multiple_paths() -> (
+    tuple[
+        dace.SDFG,
+        dace.SDFGState,
+        dace_nodes.AccessNode,
+        dace_nodes.AccessNode,
+        dace_nodes.AccessNode,
+    ]
+):
+    """There are multiple paths between the two global nodes.
+
+    There are to global nodes and two paths between them. The first one is direct,
+    i.e. `(G) -> (G)`. The second one involves an intermediate buffer, i.e. the
+    pattern `(G) -> (T) -> (G)`.
+    The merge mode, which is supposed to be the normal mode, of the
+    `SingleStateGlobalDirectSelfCopyElimination` transformation can not handle
+    this case, but its split node can handle it.
+    """
+    sdfg = dace.SDFG(util.unique_name("self_copy_sdfg_with_multiple_paths"))
+    state = sdfg.add_state(is_start_block=True)
+
+    for name in "GT":
+        sdfg.add_array(
+            name,
+            shape=(20,),
+            dtype=dace.float64,
+            transient=False,
+        )
+    sdfg.arrays["T"].transient = True
+
+    g_read = state.add_access("G")
+    g_write = state.add_access("G")
+    tmp_node = state.add_access("T")
+
+    state.add_nedge(g_read, g_write, dace.Memlet("G[0:20] -> [0:20]"))
+    state.add_nedge(g_read, tmp_node, dace.Memlet("G[5:15] -> [5:15]"))
+    state.add_nedge(tmp_node, g_write, dace.Memlet("T[5:15] -> [5:15]"))
+    sdfg.validate()
+
+    return sdfg, state, g_read, tmp_node, g_write
+
+
 def test_global_self_copy_elimination_only_pattern():
     """Contains only the pattern -> Total elimination."""
     sdfg, state = _make_self_copy_sdfg()
@@ -144,3 +215,108 @@ def test_global_self_copy_elimination_tmp_downstream():
     assert state2.number_of_nodes() == 5
     assert util.count_nodes(state2, dace_nodes.AccessNode) == 2
     assert util.count_nodes(state2, dace_nodes.MapEntry) == 1
+
+
+def test_direct_global_self_copy_simple():
+    sdfg = dace.SDFG(util.unique_name("simple_direct_self_copy"))
+    state = sdfg.add_state(is_start_block=True)
+
+    sdfg.add_array(
+        name="G",
+        shape=(20, 20),
+        dtype=dace.float64,
+        transient=False,
+    )
+
+    state.add_nedge(
+        state.add_access("G"),
+        state.add_access("G"),
+        dace.Memlet("G[0:20, 0:20] -> [0:20, 0:20]"),
+    )
+
+    sdfg.validate()
+    assert util.count_nodes(sdfg, dace_nodes.AccessNode) == 2
+
+    count = sdfg.apply_transformations_repeated(
+        gtx_transformations.SingleStateGlobalDirectSelfCopyElimination,
+        validate=True,
+        validate_all=True,
+    )
+
+    assert count == 1
+    assert util.count_nodes(sdfg, dace_nodes.AccessNode) == 0
+
+
+def test_direct_global_self_copy_used():
+    """The SDFG has a direct self copy pattern, but there are other involved nodes."""
+    sdfg = _make_direct_self_copy_elimination_used_sdfg()
+    assert util.count_nodes(sdfg, dace_nodes.AccessNode) == 5
+
+    ref = {
+        "A": np.array(np.random.rand(20), dtype=np.float64, copy=True),
+        "B": np.array(np.random.rand(20), dtype=np.float64, copy=True),
+        "C": np.array(np.random.rand(20), dtype=np.float64, copy=True),
+        "G": np.array(np.random.rand(20), dtype=np.float64, copy=True),
+    }
+    res = {k: np.copy(v, order="K") for k, v in ref.items()}
+
+    csdfg_ref = sdfg.compile()
+    csdfg_ref(**ref)
+
+    count = sdfg.apply_transformations_repeated(
+        gtx_transformations.SingleStateGlobalDirectSelfCopyElimination,
+        validate=True,
+        validate_all=True,
+    )
+
+    ac_nodes_after = util.count_nodes(sdfg, dace_nodes.AccessNode, return_nodes=True)
+    assert count == 1
+    assert len(ac_nodes_after) == 4
+    assert len(set(ac.data for ac in ac_nodes_after)) == 4
+
+    csdfg_res = sdfg.compile()
+    csdfg_res(**res)
+
+    assert all(np.all(ref[k] == res[k]) for k in ref.keys())
+
+
+def test_direct_self_copy_elimination_split_mode():
+    sdfg, state, node_read_g, node_tmp, node_write_g = _make_self_copy_sdfg_with_multiple_paths()
+    assert state.number_of_nodes() == 3
+    assert state.number_of_edges() == 3
+    assert state.out_degree(node_read_g) == 2
+    assert state.in_degree(node_write_g) == 2
+    assert state.degree(node_tmp) == 2
+
+    # The `SingleStateGlobalSelfCopyElimination` transformation will use its "split"
+    #  mode, thus only removing the direct connection between the two g nodes.
+    count = sdfg.apply_transformations_repeated(
+        gtx_transformations.SingleStateGlobalDirectSelfCopyElimination,
+        validate=True,
+        validate_all=True,
+    )
+
+    assert count == 1
+    assert state.number_of_nodes() == 3
+    assert state.number_of_edges() == 2
+    assert state.out_degree(node_read_g) == 1
+    assert state.in_degree(node_write_g) == 1
+    assert state.degree(node_tmp) == 2
+
+
+def test_global_self_copy_elimination_multi_path():
+    sdfg, _, node_read_g, node_tmp, node_write_g = _make_self_copy_sdfg_with_multiple_paths()
+    assert util.count_nodes(sdfg, dace_nodes.AccessNode) == 3
+
+    # For some reason calling `sdfg.apply_transformations_repeated()` does not work.
+    #  Probably a problem in the matcher, for that reason we will call it directly.
+    gtx_transformations.SingleStateGlobalSelfCopyElimination.apply_to(
+        sdfg=sdfg,
+        verify=True,
+        node_read_g=node_read_g,
+        node_tmp=node_tmp,
+        node_write_g=node_write_g,
+    )
+
+    sdfg.validate()
+    assert util.count_nodes(sdfg, dace_nodes.AccessNode) == 0
diff --git a/tests/next_tests/unit_tests/test_common.py b/tests/next_tests/unit_tests/test_common.py
index 09ca44aa..c38496eb 100644
--- a/tests/next_tests/unit_tests/test_common.py
+++ b/tests/next_tests/unit_tests/test_common.py
@@ -582,11 +582,11 @@ def dimension_promotion_cases() -> (
             "Dimensions 'KDim[vertical], JDim[horizontal]' are not ordered correctly, expected 'JDim[horizontal], KDim[vertical]'.",
         ),
         (
-            [[JDim, V2E], [IDim, KDim, E2C2V]],
+            [[JDim, V2E], [IDim, E2C2V, KDim]],
             None,
             "There are more than one dimension with DimensionKind 'LOCAL'.",
         ),
-        ([[JDim, V2E], [IDim, KDim]], [IDim, JDim, KDim, V2E], None),
+        ([[JDim, V2E], [IDim, KDim]], [IDim, JDim, V2E, KDim], None),
     ]
     return [
         ([[el for el in arg] for arg in args], [el for el in result] if result else result, msg)