Repository: GridTools/gt4py
Pull request #1594 (2024-07-29 07:59:45)

feat[next][dace]: Dace fieldview transformations
Initial PR for the Optimization pipeline of the DaCe backend in GT4Py.
The PR is very basic, but contains the basis for later advancement.
Essentially, it defines the `gt_auto_optimize()` function which performs optimization on a basic heuristic.

There are some parts missing, such as:
- There is no way to guide the optimization, i.e. a decision making process that determine if a map should be fused or not.
- There is also no transformation which fuses parallel maps (was not needed for this kernel).


philip-paul-mueller: @havogt I am on vacation, so no hurry.
philip-paul-mueller: @edopao Thanks for spotting it I have removed it.
philip-paul-mueller: cscs-ci run default
egparedes: It looks like the tests you were trying to fix are still failing...


Patch:
diff --git a/AUTHORS.md b/AUTHORS.md
index 1bcdd266..f6f8ff77 100644
--- a/AUTHORS.md
+++ b/AUTHORS.md
@@ -24,6 +24,7 @@
 - Madonna, Alberto. ETH Zurich - CSCS
 - Mariotti, Kean. ETH Zurich - CSCS
 - Müller, Christoph. MeteoSwiss
+- Müller, Philip. ETH Zurich - CSCS
 - Osuna, Carlos. MeteoSwiss
 - Paone, Edoardo. ETH Zurich - CSCS
 - Röthlin, Matthias. MeteoSwiss
diff --git a/docs/development/ADRs/0018-Canonical_SDFG_in_GT4Py_Transformations.md b/docs/development/ADRs/0018-Canonical_SDFG_in_GT4Py_Transformations.md
new file mode 100644
index 00000000..18b9c1f8
--- /dev/null
+++ b/docs/development/ADRs/0018-Canonical_SDFG_in_GT4Py_Transformations.md
@@ -0,0 +1,130 @@
+---
+tags: [backend, dace, optimization]
+---
+
+# Canonical Form of an SDFG in GT4Py (Especially for Optimizations)
+
+- **Status**: valid
+- **Authors**: Philip Müller (@philip-paul-mueller)
+- **Created**: 2024-08-27
+
+In the context of the implementation of the new DaCe fieldview we decided about a particular form of the SDFG.
+Their main intent is to reduce the complexity of the GT4Py specific transformations.
+
+## Context
+
+The canonical form outlined in this document was mainly designed from the perspective of the optimization pipeline.
+Thus it emphasizes a form that can be handled in a simple and efficient way by a transformation.
+In the pipeline we distinguish between:
+
+- Intrastate optimization: optimization of the data flow within states.
+- Interstate optimization: optimization between states, these are transformations that are _intended_ to _reduce_ the number of states.
+
+The current (GT4Py) pipeline mainly focus on intrastate optimization and relays on DaCe, especially its simplify pass, for interstate optimizations.
+
+## Decision
+
+The canonical form is defined by several rules that affect different aspects of an SDFG and what a transformation can assume.
+This allows simplifying the implementation of certain transformations.
+
+#### General Aspects
+
+The following rules especially affects transformations and how they operate:
+
+1. Intrastate transformation and interstate transformations must run separately and can not be mixed in the same (DaCe) pipeline.
+
+   - [Rationale]: As a consequence the number of "interstate transients" (transients that are used in multiple states) remains constant during intrastate transformations.
+   - [Note 1]: It is allowed to run them one after another, as long as they are strictly separated.
+   - [Note 2]: It is allowed for an _intrastate_ transformation to act in a way that allows state fusion by later intrastate transformations.
+   - [Note 3]: The DaCe simplification pass violates this rule, for that reason this pass must always be called on its own, see also rule 2.
+
+2. It is invalid to call the simplification pass directly, i.e. the usage of `SDFG.simplify()` is not allowed. The only valid way to call _simplify()_ is to call the `gt_simplify()` function provided by GT4Py.
+   - [Rationale]: It was observed that some sub-passes in _simplify()_ have a negative impact and that additional passes might be needed in the future.
+     By using a single function later modifications to _simplify()_ are easy.
+   - [Note]: One issue is that the remove redundant array transformation is not able to handle all cases.
+
+#### Global Memory
+
+The only restriction we impose on global memory is:
+
+3. The same global memory is allowed to be used as input and output at the same time, if and only if the output depends _elementwise_ on the input.
+   - [Rationale 1]: This allows the removal of double buffering, that DaCe may not remove. See also rule 2.
+   - [Rationale 2]: This formulation allows writing expressions such as `a += 1`, with only memory for `a`.
+     Phrased more technically, using global memory for input and output is allowed if and only if the two computations `tmp = computation(global_memory); global_memory = tmp;` and `global_memory = computation(global_memory);` are equivalent.
+   - [Note]: In the long term this rule will be changed to: Global memory (an array) is either used as input (only read from) or as output (only written to) but never for both.
+
+#### State Machine
+
+For the SDFG state machine we assume that:
+
+4. An interstate edge can only access scalars, i.e. use them in their assignment or condition expressions, but not arrays, even if they have shape `(1,)`.
+
+   - [Rationale]: If an array is also used in interstate edges it became very tedious to verify if the array could be removed or not.
+   - [Note]: Running _simplify()_ might actually result in the violation of this rule, see note of rule 9.
+
+5. The state graph does not contain any cycles, i.e. the implementation of a for/while loop using states is not allowed, the new loop construct or serial maps must be used in that case.
+   - [Rationale]: This is a simplification that makes it much simpler to define what "later in the computation" means, as we will never have a cycle.
+   - [Note]: Currently the code generator does not support the `LoopRegion` construct and it is transformed to a state machine.
+
+#### Transients
+
+The rules we impose on transients are a bit more complicated, however, while sounding restrictive, they are very permissive.
+It is important to note that these rules only have to be met after _simplify()_ was called once on the SDFG:
+
+6. Downstream of a write access, i.e., in all states that follow the state where the access node is located, there are no other access nodes that are used to write to the same array.
+
+   - [Rationale 1]: This rule, together with rule 7 and 8, essentially ensures that the assignment in the SDFG follows SSA style, while allowing for expressions such as:
+
+   ```python
+   if cond:
+       a = true_branch()
+   else:
+       a = false_branch()
+   ```
+
+   (**NOTE:** This could also be done with references, however, they are strongly discouraged.)
+
+   - [Rationale 2]: This still allows reductions with WCR as they write to the same access node and loops, whose body modifies a transient that outlives the loop body, as they use the same access node.
+
+7. It is _recommended_ that a write access node should only have one incoming edge.
+
+   - [Rationale]: This case is handled poorly by some DaCe transformations, thus we should avoid them as much as possible.
+
+8. No two access nodes in a state can refer to the same array.
+
+   - [Rationale]: Together with rule 5 this guarantees SSA style.
+   - [Note]: An SDFG can still be constructed using different access node for the same underlying data; _simplify()_ will combine them.
+
+9. Every access node that reads from an array (having an outgoing edge) that was not written to in the same state must be a source node.
+
+   - [Rationale]: Together with rule 1, 4, 5, 6, 7 and 8 this simplifies checking if a transient can be safely removed or if it is used somewhere else.
+     These rules guarantee that the number of "interstate transients" remains constant and this set is given by the _set of source nodes and all access nodes that have an outgoing degree larger than one_.
+   - [Note]: To prevent some issues caused by the violation of rule 4 by _simplify()_, this set is extended with the transient sink nodes and all scalars.
+     Excess interstate transients, that will be kept alive that way, will be removed by later calls to _simplify()_.
+
+10. Every AccessNode within a map scope must refer to a data descriptor whose lifetime must be `dace.dtypes.AllocationLifetime.Scope` and its storage class should either be `dace.dtypes.StorageType.Default` or _preferably_ `dace.dtypes.StorageType.Register`.
+    - [Rationale 1]: This makes optimizations operating inside maps/kernels simpler, as it guarantees that the AccessNode does not propagate outside.
+    - [Rationale 2]: The storage type avoids the need to dynamically allocate memory inside a kernel.
+
+#### Maps
+
+For maps we assume the following:
+
+11. The names of map variables (iteration variables) follow the following pattern.
+
+    - 11.1: All map variables iterating over the same dimension (disregarding the actual range) have the same deterministic name, that includes the `gtx.Dimension.value` string.
+    - 11.2: The name of horizontal dimensions (`kind` attribute) always end in `__gtx_horizontal`.
+    - 11.3: The name of vertical dimensions (`kind` attribute) always end in `__gtx_vertical`.
+    - 11.4: The name of local dimensions always ends in `__gtx_localdim`.
+    - 11.5: No transformation is allowed to modify the name of an iteration variable that follows rules 11.2, 11.3 or 11.4.
+    - [Rationale]: Without this rule it is very hard to tell which map variable does what, this way we can transmit information from GT4Py to DaCe, see also rule 12.
+
+12. Two map ranges, i.e. the pair map/iteration variable and range, can only be fused if they have the same name _and_ cover the same range.
+    - [Rationale 1]: Because of rule 11, we will only fuse maps that actually makes sense to fuse.
+    - [Rationale 2]: This allows fusing maps without renaming the map variables.
+    - [Note]: This rule might be dropped in the future.
+
+## Consequences
+
+The rules outlined above impose a certain form of an SDFG.
+Most of these rules are designed to ensure that the SDFG follows SSA style and to simplify transformations, especially making validation checks simple, while imposing a minimal number of restrictions.
diff --git a/docs/development/ADRs/Index.md b/docs/development/ADRs/Index.md
index 072e6dc2..ccf2e744 100644
--- a/docs/development/ADRs/Index.md
+++ b/docs/development/ADRs/Index.md
@@ -38,7 +38,7 @@ _None_
 
 ### Transformations
 
-_None_
+- [0018 - Canonical Form of an SDFG in GT4Py (Especially for Optimizations)](0018-Canonical_SDFG_in_GT4Py_Transformations.md)
 
 ### Backends and Code Generation
 
@@ -47,6 +47,7 @@ _None_
 - [0008 - Mapping Domain to Cpp Backend](0008-Mapping_Domain_to_Cpp-Backend.md)
 - [0016 - Multiple Backends and Build Systems](0016-Multiple-Backends-and-Build-Systems.md)
 - [0017 - Toolchain Configuration](0017-Toolchain-Configuration.md)
+- [0018 - Canonical Form of an SDFG in GT4Py (Especially for Optimizations)](0018-Canonical_SDFG_in_GT4Py_Transformations.md)
 
 ### Python Integration
 
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_to_sdfg.py b/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_to_sdfg.py
index 8d23aa42..99388336 100644
--- a/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_to_sdfg.py
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_to_sdfg.py
@@ -29,6 +29,7 @@ from gt4py.next.iterator.type_system import inference as gtir_type_inference
 from gt4py.next.program_processors.runners.dace_fieldview import (
     gtir_builtin_translators,
     gtir_to_tasklet,
+    transformations as gtx_transformations,
     utility as dace_fieldview_util,
 )
 from gt4py.next.type_system import type_specifications as ts, type_translation as tt
@@ -534,5 +535,5 @@ def build_sdfg_from_gtir(
     sdfg = sdfg_genenerator.visit(program)
     assert isinstance(sdfg, dace.SDFG)
 
-    sdfg.simplify()
+    gtx_transformations.gt_simplify(sdfg)
     return sdfg
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/__init__.py b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/__init__.py
new file mode 100644
index 00000000..53fa1eee
--- /dev/null
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/__init__.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
+
+"""Transformation and optimization pipeline for the DaCe backend in GT4Py.
+
+Please also see [ADR0018](https://github.com/GridTools/gt4py/tree/main/docs/development/ADRs/0018-Canonical_SDFG_in_GT4Py_Transformations.md)
+that explains the general structure and requirements on the SDFGs.
+"""
+
+from .auto_opt import gt_auto_optimize, gt_set_iteration_order, gt_simplify
+from .gpu_utils import GPUSetBlockSize, gt_gpu_transformation, gt_set_gpu_blocksize
+from .loop_blocking import LoopBlocking
+from .map_orderer import MapIterationOrder
+from .map_promoter import SerialMapPromoter
+from .map_serial_fusion import SerialMapFusion
+
+
+__all__ = [
+    "GPUSetBlockSize",
+    "LoopBlocking",
+    "MapIterationOrder",
+    "SerialMapFusion",
+    "SerialMapPromoter",
+    "SerialMapPromoterGPU",
+    "gt_auto_optimize",
+    "gt_gpu_transformation",
+    "gt_set_iteration_order",
+    "gt_set_gpu_blocksize",
+    "gt_simplify",
+]
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/auto_opt.py b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/auto_opt.py
new file mode 100644
index 00000000..e19dccc6
--- /dev/null
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/auto_opt.py
@@ -0,0 +1,367 @@
+# 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
+
+"""Fast access to the auto optimization on DaCe."""
+
+from typing import Any, Optional, Sequence
+
+import dace
+from dace.transformation import dataflow as dace_dataflow
+from dace.transformation.auto import auto_optimize as dace_aoptimize
+from dace.transformation.passes import simplify as dace_passes_simplify
+
+from gt4py.next import common as gtx_common
+from gt4py.next.program_processors.runners.dace_fieldview import (
+    transformations as gtx_transformations,
+)
+
+
+def gt_simplify(
+    sdfg: dace.SDFG,
+    validate: bool = True,
+    validate_all: bool = False,
+    skip: Optional[set[str]] = None,
+) -> Any:
+    """Performs simplifications on the SDFG in place.
+
+    Instead of calling `sdfg.simplify()` directly, you should use this function,
+    as it is specially tuned for GridTool based SDFGs.
+
+    Args:
+        sdfg: The SDFG to optimize.
+        validate: Perform validation after the pass has run.
+        validate_all: Perform extensive validation.
+        skip: List of simplify passes that should not be applied.
+
+    Note:
+        The reason for this function is that we can influence how simplify works.
+        Since some parts in simplify might break things in the SDFG.
+        However, currently nothing is customized yet, and the function just calls
+        the simplification pass directly.
+    """
+
+    return dace_passes_simplify.SimplifyPass(
+        validate=validate,
+        validate_all=validate_all,
+        verbose=False,
+        skip=skip,
+    ).apply_pass(sdfg, {})
+
+
+def gt_set_iteration_order(
+    sdfg: dace.SDFG,
+    leading_dim: gtx_common.Dimension,
+    validate: bool = True,
+    validate_all: bool = False,
+) -> Any:
+    """Set the iteration order of the Maps correctly.
+
+    Modifies the order of the Map parameters such that `leading_dim`
+    is the fastest varying one, the order of the other dimensions in
+    a Map is unspecific. `leading_dim` should be the dimensions were
+    the stride is one.
+
+    Args:
+        sdfg: The SDFG to process.
+        leading_dim: The leading dimensions.
+        validate: Perform validation during the steps.
+        validate_all: Perform extensive validation.
+    """
+    return sdfg.apply_transformations_once_everywhere(
+        gtx_transformations.MapIterationOrder(
+            leading_dim=leading_dim,
+        ),
+        validate=validate,
+        validate_all=validate_all,
+    )
+
+
+def gt_auto_optimize(
+    sdfg: dace.SDFG,
+    gpu: bool,
+    leading_dim: Optional[gtx_common.Dimension] = None,
+    aggressive_fusion: bool = True,
+    max_optimization_rounds_p2: int = 100,
+    make_persistent: bool = True,
+    gpu_block_size: Optional[Sequence[int | str] | str] = None,
+    blocking_dim: Optional[gtx_common.Dimension] = None,
+    blocking_size: int = 10,
+    reuse_transients: bool = False,
+    gpu_launch_bounds: Optional[int | str] = None,
+    gpu_launch_factor: Optional[int] = None,
+    validate: bool = True,
+    validate_all: bool = False,
+    **kwargs: Any,
+) -> dace.SDFG:
+    """Performs GT4Py specific optimizations on the SDFG in place.
+
+    This function expects that the input SDFG follows the principles that are
+    outlined in [ADR0018](https://github.com/GridTools/gt4py/tree/main/docs/development/ADRs/0018-Canonical_SDFG_in_GT4Py_Transformations.md).
+    It is important to notice, that if `reuse_transients` is active then the
+    optimized SDFG no longer conforms to these rules.
+
+    The auto optimization works in different phases, that focuses each on
+    different aspects of the SDFG. The initial SDFG is assumed to have a
+    very large number of rather simple Maps.
+
+    1. Some general simplification transformations, beyond classical simplify,
+        are applied to the SDFG.
+    2. Tries to create larger kernels by fusing smaller ones, see
+        `gt_auto_fuse_top_level_maps()` for more details.
+    3. After the function created big kernels/maps it will apply some optimization,
+        inside the kernels itself. For example fuse maps inside them.
+    4. Afterwards it will process the map ranges and iteration order. For this
+        the function assumes that the dimension indicated by `leading_dim` is the
+        one with stride one.
+    5. If requested the function will now apply loop blocking, on the dimension
+        indicated by `leading_dim`.
+    6. If requested the SDFG will be transformed to GPU. For this the
+        `gt_gpu_transformation()` function is used, that might apply several other
+        optimizations.
+    7. Afterwards some general transformations to the SDFG are applied.
+        This includes:
+        - Use fast implementation for library nodes.
+        - Move small transients to stack.
+        - Make transients persistent (if requested).
+        - Apply DaCe's `TransientReuse` transformation (if requested).
+
+    Args:
+        sdfg: The SDFG that should be optimized in place.
+        gpu: Optimize for GPU or CPU.
+        leading_dim: Leading dimension, indicates where the stride is 1.
+        aggressive_fusion: Be more aggressive during phase 2, will lead to the promotion
+            of certain maps (over computation) but will lead to larger kernels.
+        max_optimization_rounds_p2: Maximum number of optimization rounds in phase 2.
+        make_persistent: Turn all transients to persistent lifetime, thus they are
+            allocated over the whole lifetime of the program, even if the kernel exits.
+            Thus the SDFG can not be called by different threads.
+        gpu_block_size: The thread block size for maps in GPU mode, currently only
+            one for all.
+        blocking_dim: On which dimension blocking should be applied.
+        blocking_size: How many elements each block should process.
+        reuse_transients: Run the `TransientReuse` transformation, might reduce memory footprint.
+        gpu_launch_bounds: Use this value as `__launch_bounds__` for _all_ GPU Maps.
+        gpu_launch_factor: Use the number of threads times this value as `__launch_bounds__`
+            for _all_ GPU Maps.
+        validate: Perform validation during the steps.
+        validate_all: Perform extensive validation.
+
+    Todo:
+        - Make sure that `SDFG.simplify()` is not called indirectly, by temporarily
+            overwriting it with `gt_simplify()`.
+        - Specify arguments to set the size of GPU thread blocks depending on the
+            dimensions. I.e. be able to use a different size for 1D than 2D Maps.
+        - Add a parallel version of Map fusion.
+        - Implement some model to further guide to determine what we want to fuse.
+            Something along the line "Fuse if operational intensity goes up, but
+            not if we have too much internal space (register pressure).
+        - Create a custom array elimination pass that honors rule 1.
+        - Check if a pipeline could be used to speed up some computations.
+    """
+    device = dace.DeviceType.GPU if gpu else dace.DeviceType.CPU
+
+    with dace.config.temporary_config():
+        dace.Config.set("optimizer", "match_exception", value=True)
+        dace.Config.set("store_history", value=False)
+
+        # TODO(phimuell): Should there be a zeroth phase, in which we generate
+        #       a chanonical form of the SDFG, for example move all local maps
+        #       to internal serial maps, such that they do not block fusion?
+
+        # Phase 1: Initial Cleanup
+        gt_simplify(
+            sdfg=sdfg,
+            validate=validate,
+            validate_all=validate_all,
+        )
+        sdfg.apply_transformations_repeated(
+            [
+                dace_dataflow.TrivialMapElimination,
+                # TODO(phimuell): Investigate if these two are appropriate.
+                dace_dataflow.MapReduceFusion,
+                dace_dataflow.MapWCRFusion,
+            ],
+            validate=validate,
+            validate_all=validate_all,
+        )
+
+        # Phase 2: Kernel Creation
+        #   Try to create kernels as large as possible.
+        sdfg = gt_auto_fuse_top_level_maps(
+            sdfg=sdfg,
+            aggressive_fusion=aggressive_fusion,
+            max_optimization_rounds=max_optimization_rounds_p2,
+            validate=validate,
+            validate_all=validate_all,
+        )
+
+        # Phase 3: Optimizing the kernels, i.e. the larger maps, themselves.
+        #   Currently this only applies fusion inside Maps.
+        sdfg.apply_transformations_repeated(
+            gtx_transformations.SerialMapFusion(
+                only_inner_maps=True,
+            ),
+            validate=validate,
+            validate_all=validate_all,
+        )
+        gt_simplify(sdfg)
+
+        # Phase 4: Iteration Space
+        #   This essentially ensures that the stride 1 dimensions are handled
+        #   by the inner most loop nest (CPU) or x-block (GPU)
+        if leading_dim is not None:
+            gt_set_iteration_order(
+                sdfg=sdfg,
+                leading_dim=leading_dim,
+                validate=validate,
+                validate_all=validate_all,
+            )
+
+        # Phase 5: Apply blocking
+        if blocking_dim is not None:
+            sdfg.apply_transformations_once_everywhere(
+                gtx_transformations.LoopBlocking(
+                    blocking_size=blocking_size,
+                    blocking_parameter=blocking_dim,
+                ),
+                validate=validate,
+                validate_all=validate_all,
+            )
+
+        # Phase 6: Going to GPU
+        if gpu:
+            # TODO(phimuell): The GPU function might modify the map iteration order.
+            #                   This is because how it is implemented (promotion and
+            #                   fusion). However, because of its current state, this
+            #                   should not happen, but we have to look into it.
+            gtx_transformations.gt_gpu_transformation(
+                sdfg,
+                gpu_block_size=gpu_block_size,
+                gpu_launch_bounds=gpu_launch_bounds,
+                gpu_launch_factor=gpu_launch_factor,
+                validate=validate,
+                validate_all=validate_all,
+                try_removing_trivial_maps=True,
+            )
+
+        # Phase 7: General Optimizations
+        #   The following operations apply regardless if we have a GPU or CPU.
+        #   The DaCe auto optimizer also uses them. Note that the reuse transient
+        #   is not done by DaCe.
+        if reuse_transients:
+            # TODO(phimuell): Investigate if we should enable it, it may make things
+            #                   harder for the compiler. Maybe write our own to
+            #                   only consider big transients and not small ones (~60B)
+            transient_reuse = dace.transformation.passes.TransientReuse()
+            transient_reuse.apply_pass(sdfg, {})
+
+        # Set the implementation of the library nodes.
+        dace_aoptimize.set_fast_implementations(sdfg, device)
+        # TODO(phimuell): Fix the bug, it uses the tile value and not the stack array value.
+        dace_aoptimize.move_small_arrays_to_stack(sdfg)
+        if make_persistent:
+            # TODO(phimuell): Allow to also to set the lifetime to `SDFG`.
+            dace_aoptimize.make_transients_persistent(sdfg, device)
+
+        return sdfg
+
+
+def gt_auto_fuse_top_level_maps(
+    sdfg: dace.SDFG,
+    aggressive_fusion: bool = True,
+    max_optimization_rounds: int = 100,
+    validate: bool = True,
+    validate_all: bool = False,
+) -> dace.SDFG:
+    """Performs the second phase of the auto optimization process.
+
+    As noted in the doc of `gt_auto_optimize()` the function tries to reduce the
+    number of kernels/maps by fusing maps. This process essentially builds on
+    the map fusion transformations and some clean up transformations.
+
+    It is important to note that the fusion will only affect top level maps, i.e.
+    nested maps are ignored. Furthermore, the function will iteratively perform
+    optimizations until a fix point is reached. If this does not happen within
+    `max_optimization_rounds` iterations an error is generated.
+
+    Return:
+        The function optimizes the SDFG in place and returns it.
+
+    Args:
+        sdfg: The SDFG to optimize.
+        aggressive_fusion: allow more aggressive fusion by promoting maps (over
+            computing).
+        max_optimization_rounds: Maximal number of optimization rounds should be
+            performed.
+        validate: Perform validation during the steps.
+        validate_all: Perform extensive validation.
+
+    Note:
+        Calling this function directly is most likely an error. Instead you should
+        call `gt_auto_optimize()` directly.
+    """
+    # Compute the SDFG hash to see if something has changed.
+    sdfg_hash = sdfg.hash_sdfg()
+
+    # We use a loop to optimize because we are using multiple transformations
+    #  after the other, thus new opportunities might arise in the next round.
+    #  We use the hash of the SDFG to detect if we have reached a fix point.
+    for _ in range(max_optimization_rounds):
+        # Use map fusion to reduce their number and to create big kernels
+        # TODO(phimuell): Use a cost measurement to decide if fusion should be done.
+        # TODO(phimuell): Add parallel fusion transformation. Should it run after
+        #                   or with the serial one?
+        sdfg.apply_transformations_repeated(
+            gtx_transformations.SerialMapFusion(
+                only_toplevel_maps=True,
+            ),
+            validate=validate,
+            validate_all=validate_all,
+        )
+
+        # Now do some cleanup task, that may enable further fusion opportunities.
+        #  Note for performance reasons simplify is deferred.
+        phase2_cleanup = []
+        phase2_cleanup.append(dace_dataflow.TrivialTaskletElimination())
+
+        # If requested perform map promotion, this will lead to more fusion.
+        if aggressive_fusion:
+            # TODO(phimuell): Should we do this all the time or only once?
+            # TODO(phimuell): Add a criteria to decide if we should promote or not.
+            # TODO(phimuell): Add parallel map promotion?
+            phase2_cleanup.append(
+                gtx_transformations.SerialMapPromoter(
+                    only_toplevel_maps=True,
+                    promote_vertical=True,
+                    promote_horizontal=False,
+                    promote_local=False,
+                )
+            )
+
+        # Perform the phase 2 cleanup.
+        sdfg.apply_transformations_once_everywhere(
+            phase2_cleanup,
+            validate=validate,
+            validate_all=validate_all,
+        )
+
+        # Use the hash to determine if the transformations did modify the SDFG.
+        #  If not we have optimized the SDFG as much as we could, in this phase.
+        old_sdfg_hash = sdfg_hash
+        sdfg_hash = sdfg.hash_sdfg()
+        if old_sdfg_hash == sdfg_hash:
+            break
+
+        # The SDFG was modified by the transformations above. The SDFG was
+        #  modified. Call Simplify and try again to further optimize.
+        gt_simplify(sdfg, validate=validate, validate_all=validate_all)
+
+    else:
+        raise RuntimeWarning("Optimization of the SDFG did not converge.")
+
+    return sdfg
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/gpu_utils.py b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/gpu_utils.py
new file mode 100644
index 00000000..06a3e869
--- /dev/null
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/gpu_utils.py
@@ -0,0 +1,392 @@
+# 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
+
+"""Functions for turning an SDFG into a GPU SDFG."""
+
+import copy
+from typing import Any, Optional, Sequence, Union
+
+import dace
+from dace import properties as dace_properties, transformation as dace_transformation
+from dace.sdfg import nodes as dace_nodes
+
+from gt4py.next.program_processors.runners.dace_fieldview import (
+    transformations as gtx_transformations,
+)
+
+
+def gt_gpu_transformation(
+    sdfg: dace.SDFG,
+    try_removing_trivial_maps: bool = True,
+    use_gpu_storage: bool = True,
+    gpu_block_size: Optional[Sequence[int | str] | str] = None,
+    gpu_launch_bounds: Optional[int | str] = None,
+    gpu_launch_factor: Optional[int] = None,
+    validate: bool = True,
+    validate_all: bool = False,
+    **kwargs: Any,
+) -> dace.SDFG:
+    """Transform an SDFG into a GPU SDFG.
+
+    The transformation expects a rather optimized SDFG and turn it into an SDFG
+    capable of running on the GPU.
+    The function performs the following steps:
+    - If requested, modify the storage location of the non transient arrays such
+        that they reside in GPU memory.
+    - Call the normal GPU transform function followed by simplify.
+    - If requested try to remove trivial kernels.
+    - If specified, set the `gpu_block_size` parameters of the Maps to the given value.
+
+    Args:
+        sdfg: The SDFG that should be processed.
+        try_removing_trivial_maps: Try to get rid of trivial maps by incorporating them.
+        use_gpu_storage: Assume that the non global memory is already on the GPU. This
+            will avoid the data copy from host to GPU memory.
+        gpu_block_size: The size of a thread block on the GPU.
+        gpu_launch_bounds: Use this value as `__launch_bounds__` for _all_ GPU Maps.
+        gpu_launch_factor: Use the number of threads times this value as `__launch_bounds__`
+        validate: Perform validation during the steps.
+        validate_all: Perform extensive validation.
+
+    Notes:
+        The function might modify the order of the iteration variables of some
+        maps.
+        In addition it might fuse Maps together that should not be fused. To prevent
+        that you should set `try_removing_trivial_maps` to `False`.
+
+    Todo:
+        - Solve the fusing problem.
+        - Currently only one block size for all maps is given, add more options.
+    """
+    assert (
+        len(kwargs) == 0
+    ), f"gt_gpu_transformation(): found unknown arguments: {', '.join(arg for arg in kwargs.keys())}"
+
+    # Turn all global arrays (which we identify as input) into GPU memory.
+    #  This way the GPU transformation will not create this copying stuff.
+    if use_gpu_storage:
+        for desc in sdfg.arrays.values():
+            if isinstance(desc, dace.data.Array) and not desc.transient:
+                desc.storage = dace.dtypes.StorageType.GPU_Global
+
+    # Now turn it into a GPU SDFG
+    sdfg.apply_gpu_transformations(
+        validate=validate,
+        validate_all=validate_all,
+        simplify=False,
+    )
+    # The documentation recommends to run simplify afterwards
+    gtx_transformations.gt_simplify(sdfg)
+
+    if try_removing_trivial_maps:
+        # A Tasklet, outside of a Map, that writes into an array on GPU can not work
+        #  `sdfg.appyl_gpu_transformations()` puts Map around it (if said Tasklet
+        #  would write into a Scalar that then goes into a GPU Map, nothing would
+        #  happen. So we might end up with lot of these trivial Maps, that results
+        #  in a single kernel launch. To prevent this we will try to fuse them.
+        # NOTE: The current implementation has a bug, because promotion and fusion
+        #   are two different steps. Because of this the function will implicitly
+        #   fuse everything together it can find.
+        # TODO(phimuell): Fix the issue described above.
+        sdfg.apply_transformations_once_everywhere(
+            TrivialGPUMapPromoter(),
+            validate=False,
+            validate_all=False,
+        )
+        sdfg.apply_transformations_repeated(
+            gtx_transformations.SerialMapFusion(
+                only_toplevel_maps=True,
+            ),
+            validate=validate,
+            validate_all=validate_all,
+        )
+
+    # Set the GPU block size if it is known.
+    if gpu_block_size is not None:
+        gt_set_gpu_blocksize(
+            sdfg=sdfg,
+            gpu_block_size=gpu_block_size,
+            gpu_launch_bounds=gpu_launch_bounds,
+            gpu_launch_factor=gpu_launch_factor,
+        )
+
+    return sdfg
+
+
+def gt_set_gpu_blocksize(
+    sdfg: dace.SDFG,
+    gpu_block_size: Optional[Sequence[int | str] | str],
+    gpu_launch_bounds: Optional[int | str] = None,
+    gpu_launch_factor: Optional[int] = None,
+) -> Any:
+    """Set the block size related properties of _all_ Maps.
+
+    See `GPUSetBlockSize` for more information.
+
+    Args:
+        sdfg: The SDFG to process.
+        gpu_block_size: The size of a thread block on the GPU.
+        launch_bounds: The value for the launch bound that should be used.
+        launch_factor: If no `launch_bounds` was given use the number of threads
+            in a block multiplied by this number.
+    """
+    xform = GPUSetBlockSize(
+        block_size=gpu_block_size,
+        launch_bounds=gpu_launch_bounds,
+        launch_factor=gpu_launch_factor,
+    )
+    return sdfg.apply_transformations_once_everywhere([xform])
+
+
+def _gpu_block_parser(
+    self: "GPUSetBlockSize",
+    val: Any,
+) -> None:
+    """Used by the setter of `GPUSetBlockSize.block_size`."""
+    org_val = val
+    if isinstance(val, (tuple | list)):
+        pass
+    elif isinstance(val, str):
+        val = tuple(x.strip() for x in val.split(","))
+    elif isinstance(val, int):
+        val = (val,)
+    else:
+        raise TypeError(
+            f"Does not know how to transform '{type(org_val).__name__}' into a proper GPU block size."
+        )
+    if 0 < len(val) <= 3:
+        val = [*val, *([1] * (3 - len(val)))]
+    else:
+        raise ValueError(f"Can not parse block size '{org_val}': wrong length")
+    try:
+        val = [int(x) for x in val]
+    except ValueError:
+        raise TypeError(
+            f"Currently only block sizes convertible to int are supported, you passed '{val}'."
+        ) from None
+    self._block_size = val
+
+
+def _gpu_block_getter(
+    self: "GPUSetBlockSize",
+) -> tuple[int, int, int]:
+    """Used as getter in the `GPUSetBlockSize.block_size` property."""
+    assert isinstance(self._block_size, (tuple, list)) and len(self._block_size) == 3
+    assert all(isinstance(x, int) for x in self._block_size)
+    return tuple(self._block_size)
+
+
+@dace_properties.make_properties
+class GPUSetBlockSize(dace_transformation.SingleStateTransformation):
+    """Sets the GPU block size on GPU Maps.
+
+    The transformation will apply to all Maps that have a GPU schedule, regardless
+    of their dimensionality.
+
+    The `gpu_block_size` is either a sequence, of up to three integers or a string
+    of up to three numbers, separated by comma (`,`).
+    The first number is the size of the block in `x` direction, the second for the
+    `y` direction and the third for the `z` direction. Missing values will be filled
+    with `1`.
+
+    Args:
+        block_size: The size of a thread block on the GPU.
+        launch_bounds: The value for the launch bound that should be used.
+        launch_factor: If no `launch_bounds` was given use the number of threads
+            in a block multiplied by this number.
+
+    Todo:
+        Add the possibility to specify other bounds for 1, 2, or 3 dimensional maps.
+    """
+
+    block_size = dace_properties.Property(
+        dtype=None,
+        allow_none=False,
+        default=(32, 1, 1),
+        setter=_gpu_block_parser,
+        getter=_gpu_block_getter,
+        desc="Size of the block size a GPU Map should have.",
+    )
+
+    launch_bounds = dace_properties.Property(
+        dtype=str,
+        allow_none=True,
+        default=None,
+        desc="Set the launch bound property of the map.",
+    )
+
+    map_entry = dace_transformation.transformation.PatternNode(dace_nodes.MapEntry)
+
+    def __init__(
+        self,
+        block_size: Sequence[int | str] | str | None = None,
+        launch_bounds: int | str | None = None,
+        launch_factor: int | None = None,
+    ) -> None:
+        super().__init__()
+        if block_size is not None:
+            self.block_size = block_size
+
+        if launch_factor is not None:
+            assert launch_bounds is None
+            self.launch_bounds = str(
+                int(launch_factor) * self.block_size[0] * self.block_size[1] * self.block_size[2]
+            )
+        elif launch_bounds is None:
+            self.launch_bounds = None
+        elif isinstance(launch_bounds, (str, int)):
+            self.launch_bounds = str(launch_bounds)
+        else:
+            raise TypeError(
+                f"Does not know how to parse '{launch_bounds}' as 'launch_bounds' argument."
+            )
+
+    @classmethod
+    def expressions(cls) -> Any:
+        return [dace.sdfg.utils.node_path_graph(cls.map_entry)]
+
+    def can_be_applied(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        expr_index: int,
+        sdfg: dace.SDFG,
+        permissive: bool = False,
+    ) -> bool:
+        """Test if the block size can be set.
+
+        The function tests:
+        - If the block size of the map is already set.
+        - If the map is at global scope.
+        - If if the schedule of the map is correct.
+        """
+
+        scope = graph.scope_dict()
+        if scope[self.map_entry] is not None:
+            return False
+        if self.map_entry.map.schedule not in dace.dtypes.GPU_SCHEDULES:
+            return False
+        if self.map_entry.map.gpu_block_size is not None:
+            return False
+        return True
+
+    def apply(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        sdfg: dace.SDFG,
+    ) -> None:
+        """Modify the map as requested."""
+        self.map_entry.map.gpu_block_size = self.block_size
+        if self.launch_bounds is not None:  # Note empty string has a meaning in DaCe
+            self.map_entry.map.gpu_launch_bounds = self.launch_bounds
+
+
+@dace_properties.make_properties
+class TrivialGPUMapPromoter(dace_transformation.SingleStateTransformation):
+    """Serial Map promoter for empty GPU maps.
+
+    In CPU mode a Tasklet can be outside of a map, however, this is not
+    possible in GPU mode. For this reason DaCe wraps such Tasklets in a
+    trivial Map.
+    This transformation will look for such Maps and promote them, such
+    that they can be fused with downstream maps.
+
+    Note:
+        - This transformation should not be run on its own, instead it
+            is run within the context of `gt_gpu_transformation()`.
+        - This transformation must be run after the GPU Transformation.
+        - Currently the transformation does not do the fusion on its own.
+            Instead map fusion must be run afterwards.
+        - The transformation assumes that the upper Map is a trivial Tasklet.
+            Which should be the majority of all cases.
+    """
+
+    # Pattern Matching
+    trivial_map_exit = dace_transformation.transformation.PatternNode(dace_nodes.MapExit)
+    access_node = dace_transformation.transformation.PatternNode(dace_nodes.AccessNode)
+    second_map_entry = dace_transformation.transformation.PatternNode(dace_nodes.MapEntry)
+
+    @classmethod
+    def expressions(cls) -> Any:
+        return [
+            dace.sdfg.utils.node_path_graph(
+                cls.trivial_map_exit, cls.access_node, cls.second_map_entry
+            )
+        ]
+
+    def can_be_applied(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        expr_index: int,
+        sdfg: dace.SDFG,
+        permissive: bool = False,
+    ) -> bool:
+        """Tests if the promotion is possible.
+
+        The tests includes:
+        - Schedule of the maps.
+        - If the map is trivial.
+        - If the trivial map was not used to define a symbol.
+        - Intermediate access node can only have in and out degree of 1.
+        - The trivial map exit can only have one output.
+        """
+        trivial_map_exit: dace_nodes.MapExit = self.trivial_map_exit
+        trivial_map: dace_nodes.Map = trivial_map_exit.map
+        trivial_map_entry: dace_nodes.MapEntry = graph.entry_node(trivial_map_exit)
+        second_map: dace_nodes.Map = self.second_map_entry.map
+        access_node: dace_nodes.AccessNode = self.access_node
+
+        # The kind of maps we are interested only have one parameter.
+        if len(trivial_map.params) != 1:
+            return False
+
+        # Check if it is a GPU map
+        for map_to_check in [trivial_map, second_map]:
+            if map_to_check.schedule not in [
+                dace.dtypes.ScheduleType.GPU_Device,
+                dace.dtypes.ScheduleType.GPU_Default,
+            ]:
+                return False
+
+        # Check if the map is trivial.
+        for rng in trivial_map.range.ranges:
+            if rng[0] != rng[1]:
+                return False
+
+        # Now we have to ensure that the symbol is not used inside the scope of the
+        #  map, if it is, then the symbol is just there to define a symbol.
+        scope_view = graph.scope_subgraph(
+            trivial_map_entry,
+            include_entry=False,
+            include_exit=False,
+        )
+        if any(map_param in scope_view.free_symbols for map_param in trivial_map.params):
+            return False
+
+        # ensuring that the trivial map exit and the intermediate node have degree
+        #  one is a cheap way to ensure that the map can be merged into the
+        #  second map.
+        if graph.in_degree(access_node) != 1:
+            return False
+        if graph.out_degree(access_node) != 1:
+            return False
+        if graph.out_degree(trivial_map_exit) != 1:
+            return False
+
+        return True
+
+    def apply(self, graph: Union[dace.SDFGState, dace.SDFG], sdfg: dace.SDFG) -> None:
+        """Performs the Map Promoting.
+
+        The function essentially copies the parameters and the ranges from the
+        bottom map to the top one.
+        """
+        trivial_map: dace_nodes.Map = self.trivial_map_exit.map
+        second_map: dace_nodes.Map = self.second_map_entry.map
+
+        trivial_map.params = copy.deepcopy(second_map.params)
+        trivial_map.range = copy.deepcopy(second_map.range)
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/loop_blocking.py b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/loop_blocking.py
new file mode 100644
index 00000000..7acd997a
--- /dev/null
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/loop_blocking.py
@@ -0,0 +1,618 @@
+# 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 copy
+from typing import Any, Optional, Union
+
+import dace
+from dace import (
+    properties as dace_properties,
+    subsets as dace_subsets,
+    transformation as dace_transformation,
+)
+from dace.sdfg import graph as dace_graph, nodes as dace_nodes
+from dace.transformation import helpers as dace_helpers
+
+from gt4py.next import common as gtx_common
+from gt4py.next.program_processors.runners.dace_fieldview import utility as gtx_dace_fieldview_util
+
+
+@dace_properties.make_properties
+class LoopBlocking(dace_transformation.SingleStateTransformation):
+    """Applies loop blocking, also known as k-blocking, on a Map.
+
+    This transformation takes a multidimensional Map and performs blocking on a
+    single dimension, the loop variable is called `I` here. All dimensions except
+    `I` are unaffected by this transformation. In the outer Map will be replace the
+    `I` range, currently `I = 0:N`, with `__coarse_I = 0:N:B`, where `N` is the
+    original size of the range and `B` is the blocking size. The transformation
+    will then create an inner sequential map with `I = __coarse_I:(__coarse_I + B)`.
+
+    What makes this transformation different from simple blocking, is that
+    the inner map will not just be inserted right after the outer Map.
+    Instead the transformation will first identify all nodes that does not depend
+    on the blocking parameter `I` and relocate them between the outer and inner map.
+    Thus these operations will only be performed once, per inner loop.
+
+    Args:
+        blocking_size: The size of the block, denoted as `B` above.
+        blocking_parameter: On which parameter should we block.
+
+    Todo:
+        - Modify the inner map such that it always starts at zero.
+        - Allow more than one parameter on which we block.
+    """
+
+    blocking_size = dace_properties.Property(
+        dtype=int,
+        allow_none=True,
+        desc="Size of the inner blocks; 'B' in the above description.",
+    )
+    blocking_parameter = dace_properties.Property(
+        dtype=str,
+        allow_none=True,
+        desc="Name of the iteration variable on which to block (must be an exact match);"
+        " 'I' in the above description.",
+    )
+    independent_nodes = dace_properties.Property(
+        dtype=set,
+        allow_none=True,
+        default=None,
+        optional=True,
+        optional_condition=lambda _: False,
+        desc="Set of nodes that are independent of the blocking parameter.",
+    )
+    dependent_nodes = dace_properties.Property(
+        dtype=set,
+        allow_none=True,
+        default=None,
+        optional=True,
+        optional_condition=lambda _: False,
+        desc="Set of nodes that are dependent on the blocking parameter.",
+    )
+
+    outer_entry = dace_transformation.transformation.PatternNode(dace_nodes.MapEntry)
+
+    def __init__(
+        self,
+        blocking_size: Optional[int] = None,
+        blocking_parameter: Optional[Union[gtx_common.Dimension, str]] = None,
+    ) -> None:
+        super().__init__()
+        if isinstance(blocking_parameter, gtx_common.Dimension):
+            blocking_parameter = gtx_dace_fieldview_util.get_map_variable(blocking_parameter)
+        if blocking_parameter is not None:
+            self.blocking_parameter = blocking_parameter
+        if blocking_size is not None:
+            self.blocking_size = blocking_size
+
+    @classmethod
+    def expressions(cls) -> Any:
+        return [dace.sdfg.utils.node_path_graph(cls.outer_entry)]
+
+    def can_be_applied(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        expr_index: int,
+        sdfg: dace.SDFG,
+        permissive: bool = False,
+    ) -> bool:
+        """Test if the map can be blocked.
+
+        The test involves:
+        - The map must be at the top level.
+        - The block dimension must be present (exact string match).
+        - The map range must have step size of 1.
+        - The partition must exists (see `partition_map_output()`).
+        """
+        if self.blocking_parameter is None:
+            raise ValueError("The blocking dimension was not specified.")
+        elif self.blocking_size is None:
+            raise ValueError("The blocking size was not specified.")
+
+        outer_entry: dace_nodes.MapEntry = self.outer_entry
+        map_params: list[str] = outer_entry.map.params
+        map_range: dace_subsets.Range = outer_entry.map.range
+        block_var: str = self.blocking_parameter
+
+        scope = graph.scope_dict()
+        if scope[outer_entry] is not None:
+            return False
+        if block_var not in outer_entry.map.params:
+            return False
+        if map_range[map_params.index(block_var)][2] != 1:
+            return False
+        if not self.partition_map_output(graph, sdfg):
+            return False
+
+        return True
+
+    def apply(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        sdfg: dace.SDFG,
+    ) -> None:
+        """Creates a blocking map.
+
+        Performs the operation described in the doc string.
+        """
+
+        # Now compute the partitions of the nodes.
+        self.partition_map_output(graph, sdfg)
+
+        # Modify the outer map and create the inner map.
+        (outer_entry, outer_exit), (inner_entry, inner_exit) = self._prepare_inner_outer_maps(graph)
+
+        # Reconnect the edges
+        self._rewire_map_scope(
+            outer_entry=outer_entry,
+            outer_exit=outer_exit,
+            inner_entry=inner_entry,
+            inner_exit=inner_exit,
+            state=graph,
+            sdfg=sdfg,
+        )
+
+        # Clear the old partitions
+        self.independent_nodes = None
+        self.dependent_nodes = None
+
+    def _prepare_inner_outer_maps(
+        self,
+        state: dace.SDFGState,
+    ) -> tuple[
+        tuple[dace_nodes.MapEntry, dace_nodes.MapExit],
+        tuple[dace_nodes.MapEntry, dace_nodes.MapExit],
+    ]:
+        """Prepare the maps for the blocking.
+
+        The function modifies the outer map, `self.outer_entry`, by replacing the
+        blocking parameter, `self.blocking_parameter`, with a coarsened version
+        of it. In addition the function will then create the inner map, that
+        iterates over the blocking parameter, and these bounds are determined
+        by the coarsened blocking parameter of the outer map.
+
+        Args:
+            state: The state on which we operate.
+
+        Return:
+            The function returns a tuple of length two, the first element is the
+            modified outer map and the second element is the newly created
+            inner map. Each element consist of a pair containing the map entry
+            and map exit nodes of the corresponding maps.
+        """
+        outer_entry: dace_nodes.MapEntry = self.outer_entry
+        outer_exit: dace_nodes.MapExit = state.exit_node(outer_entry)
+        outer_map: dace_nodes.Map = outer_entry.map
+        outer_range: dace_subsets.Range = outer_entry.map.range
+        outer_params: list[str] = outer_entry.map.params
+        blocking_parameter_dim = outer_params.index(self.blocking_parameter)
+
+        # This is the name of the iterator that we use in the outer map for the
+        #  blocked dimension
+        coarse_block_var = "__coarse_" + self.blocking_parameter
+
+        # Generate the sequential inner map
+        rng_start = outer_range[blocking_parameter_dim][0]
+        rng_stop = outer_range[blocking_parameter_dim][1]
+        inner_label = f"inner_{outer_map.label}"
+        inner_range = {
+            self.blocking_parameter: dace_subsets.Range.from_string(
+                f"({coarse_block_var} * {self.blocking_size} + {rng_start})"
+                + ":"
+                + f"min(({rng_start} + {coarse_block_var} + 1) * {self.blocking_size}, {rng_stop} + 1)"
+            )
+        }
+        inner_entry, inner_exit = state.add_map(
+            name=inner_label,
+            ndrange=inner_range,
+            schedule=dace.dtypes.ScheduleType.Sequential,
+        )
+
+        # TODO(phimuell): Investigate if we want to prevent unrolling here
+
+        # Now we modify the properties of the outer map.
+        coarse_block_range = dace_subsets.Range.from_string(
+            f"0:int_ceil(({rng_stop} + 1) - {rng_start}, {self.blocking_size})"
+        ).ranges[0]
+        outer_map.params[blocking_parameter_dim] = coarse_block_var
+        outer_map.range[blocking_parameter_dim] = coarse_block_range
+
+        return ((outer_entry, outer_exit), (inner_entry, inner_exit))
+
+    def partition_map_output(
+        self,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> bool:
+        """Computes the partition the of the nodes of the Map.
+
+        The function divides the nodes into two sets, defined as:
+        - The independent nodes `\mathcal{I}`:
+            These are the nodes, whose output does not depend on the blocked
+            dimension. These nodes can be relocated between the outer and inner map.
+            Nodes in these set does not necessarily have a direct edge to map entry.
+            However, the exists a path from `outer_entry` to any node in this set.
+        - The dependent nodes `\mathcal{D}`:
+            These are the nodes, whose output depend on the blocked dimension.
+            Thus they can not be relocated between the two maps, but will remain
+            inside the inner scope. All these nodes have at least one edge to map entry.
+
+        The function uses an iterative method to compute the set of independent nodes.
+        In each iteration the function will look classify all nodes that have an
+        incoming edge originating either at outer_entry or from a node that was
+        already classified as independent. This is repeated until no new independent
+        nodes are found. This means that independent nodes does not necessarily have
+        a direct connection to map entry.
+
+        The dependent nodes on the other side always have a direct edge to outer_entry.
+        As they are the set of nodes that are adjacent to outer_entry but are not
+        independent.
+
+        For the sake of arguments, all nodes, except the map entry and exit nodes,
+        that are inside the scope and are not classified as dependent or independent
+        are known as "inner nodes".
+
+        If the function is able to compute the partition `True` is returned and the
+        member variables are updated. If the partition does not exists the function
+        will return `False` and the respective member variables will be `None`.
+
+        Args:
+            state: The state on which we operate.
+            sdfg: The SDFG in which we operate on.
+
+        Note:
+            - The function will only inspect the direct children of the map entry.
+            - Currently this function only considers the input Memlets and the
+                `used_symbol` properties of a Tasklet to determine if a Tasklet is dependent.
+        """
+
+        # Clear the previous partition.
+        self.independent_nodes = set()
+        self.dependent_nodes = None
+
+        while True:
+            # Find all the nodes that we have to classify in this iteration.
+            #  - All nodes adjacent to `outer_entry`
+            #  - All nodes adjacent to independent nodes.
+            nodes_to_classify: set[dace_nodes.Node] = {
+                edge.dst for edge in state.out_edges(self.outer_entry)
+            }
+            for independent_node in self.independent_nodes:
+                nodes_to_classify.update({edge.dst for edge in state.out_edges(independent_node)})
+            nodes_to_classify.difference_update(self.independent_nodes)
+
+            # Now classify each node
+            found_new_independent_node = False
+            for node_to_classify in nodes_to_classify:
+                class_res = self._classify_node(
+                    node_to_classify=node_to_classify,
+                    state=state,
+                    sdfg=sdfg,
+                )
+
+                # Check if the partition exists.
+                if class_res is None:
+                    self.independent_nodes = None
+                    return False
+                if class_res is True:
+                    found_new_independent_node = True
+
+            # If we found a new independent node then we have to continue.
+            if not found_new_independent_node:
+                break
+
+        # After the independent set is computed compute the set of dependent nodes
+        #  as the set of all nodes adjacent to `outer_entry` that are not dependent.
+        self.dependent_nodes = {
+            edge.dst
+            for edge in state.out_edges(self.outer_entry)
+            if edge.dst not in self.independent_nodes
+        }
+
+        return True
+
+    def _classify_node(
+        self,
+        node_to_classify: dace_nodes.Node,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> bool | None:
+        """Internal function for classifying a single node.
+
+        The general rule to classify if a node is independent are:
+        - The node must be a Tasklet or an AccessNode, in all other cases the
+            partition does not exist.
+        - `free_symbols` of the nodes shall not contain the blocking parameter.
+        - All incoming _empty_ edges must be connected to the map entry.
+        - A node either has only empty Memlets or none of them.
+        - Incoming Memlets does not depend on the blocking parameter.
+        - All incoming edges must start either at `self.outer_entry` or at dependent nodes.
+        - All output Memlets are non empty.
+
+        Returns:
+            The function returns `True` if `node_to_classify` is considered independent.
+            In this case the function will add the node to `self.independent_nodes`.
+            If the function returns `False` the node was classified as a dependent node.
+            The function will return `None` if the node can not be classified, in this
+            case the partition does not exist.
+
+        Args:
+            node_to_classify: The node that should be classified.
+            state: The state containing the map.
+            sdfg: The SDFG that is processed.
+        """
+        outer_entry: dace_nodes.MapEntry = self.outer_entry  # for caching.
+
+        # We are only able to handle certain kind of nodes, so screening them.
+        if isinstance(node_to_classify, dace_nodes.Tasklet):
+            if node_to_classify.side_effects:
+                # TODO(phimuell): Think of handling it.
+                return None
+        elif isinstance(node_to_classify, dace_nodes.AccessNode):
+            # AccessNodes need to have some special properties.
+            node_desc: dace.data.Data = node_to_classify.desc(sdfg)
+
+            if isinstance(node_desc, dace.data.View):
+                # Views are forbidden.
+                return None
+            if node_desc.lifetime != dace.dtypes.AllocationLifetime.Scope:
+                # The access node has to life fully within the scope.
+                return None
+        else:
+            # Any other node type we can not handle, so the partition can not exist.
+            # TODO(phimuell): Try to handle certain kind of library nodes.
+            return None
+
+        # Now we have to understand how the node generates its data. For this we have
+        #  to look at all the incoming edges.
+        in_edges: list[dace_graph.MultiConnectorEdge[dace.Memlet]] = list(
+            state.in_edges(node_to_classify)
+        )
+
+        # In a first phase we will only look if the partition exists or not.
+        #  We will therefore not check if the node is independent or not, since
+        #  for these classification to make sense the partition has to exist in the
+        #  first place.
+
+        # Either all incoming edges of a node are empty or none of them. If it has
+        #  empty edges, they are only allowed to come from the map entry.
+        found_empty_edges, found_nonempty_edges = False, False
+        for in_edge in in_edges:
+            if in_edge.data.is_empty():
+                found_empty_edges = True
+                if in_edge.src is not outer_entry:
+                    # TODO(phimuell): Lift this restriction.
+                    return None
+            else:
+                found_nonempty_edges = True
+
+        # Test if we found a mixture of empty and nonempty edges.
+        if found_empty_edges and found_nonempty_edges:
+            return None
+        assert (
+            found_empty_edges or found_nonempty_edges
+        ), f"Node '{node_to_classify}' inside '{outer_entry}' without an input connection."
+
+        # Requiring that all output Memlets are non empty implies, because we are
+        #  inside a scope, that there exists an output.
+        if any(out_edge.data.is_empty() for out_edge in state.out_edges(node_to_classify)):
+            return None
+
+        # Now we have ensured that the partition exists, thus we will now evaluate
+        #  if the node is independent or dependent.
+
+        # Test if the body of the Tasklet depends on the block variable.
+        if (
+            isinstance(node_to_classify, dace_nodes.Tasklet)
+            and self.blocking_parameter in node_to_classify.free_symbols
+        ):
+            return False
+
+        # Now we have to look at incoming edges individually.
+        #  We will inspect the subset of the Memlet to see if they depend on the
+        #  block variable. If this loop ends normally, then we classify the node
+        #  as independent and the node is added to `independent_nodes`.
+        for in_edge in in_edges:
+            memlet: dace.Memlet = in_edge.data
+            src_subset: dace_subsets.Subset | None = memlet.src_subset
+            dst_subset: dace_subsets.Subset | None = memlet.dst_subset
+
+            if memlet.is_empty():  # Empty Memlets do not impose any restrictions.
+                continue
+
+            # Now we have to look at the source and destination set of the Memlet.
+            subsets_to_inspect: list[dace_subsets.Subset] = []
+            if dst_subset is not None:
+                subsets_to_inspect.append(dst_subset)
+            if src_subset is not None:
+                subsets_to_inspect.append(src_subset)
+
+            # If a subset needs the block variable then the node is not independent
+            #  but dependent.
+            if any(self.blocking_parameter in subset.free_symbols for subset in subsets_to_inspect):
+                return False
+
+            # The edge must either originate from `outer_entry` or from an independent
+            #  node if not it is dependent.
+            if not (in_edge.src is outer_entry or in_edge.src in self.independent_nodes):
+                return False
+
+        # Loop ended normally, thus we classify the node as independent.
+        self.independent_nodes.add(node_to_classify)
+        return True
+
+    def _rewire_map_scope(
+        self,
+        outer_entry: dace_nodes.MapEntry,
+        outer_exit: dace_nodes.MapExit,
+        inner_entry: dace_nodes.MapEntry,
+        inner_exit: dace_nodes.MapExit,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> None:
+        """Rewire the edges inside the scope defined by the outer map.
+
+        The function assumes that the outer and inner map were obtained by a call
+        to `_prepare_inner_outer_maps()`. The function will now rewire the connections of these
+        nodes such that the dependent nodes are inside the scope of the inner map,
+        while the independent nodes remain outside.
+
+        Args:
+            outer_entry: The entry node of the outer map.
+            outer_exit: The exit node of the outer map.
+            inner_entry: The entry node of the inner map.
+            inner_exit: The exit node of the inner map.
+            state: The state of the map.
+            sdfg: The SDFG we operate on.
+        """
+
+        # Contains the nodes that are already have been handled.
+        relocated_nodes: set[dace_nodes.Node] = set()
+
+        # We now handle all independent nodes, this means that all of their
+        #  _output_ edges have to go through the new inner map and the Memlets need
+        #  modifications, because of the block parameter.
+        for independent_node in self.independent_nodes:
+            for out_edge in state.out_edges(independent_node):
+                edge_dst: dace_nodes.Node = out_edge.dst
+                relocated_nodes.add(edge_dst)
+
+                # If destination of this edge is also independent we do not need
+                #  to handle it, because that node will also be before the new
+                #  inner serial map.
+                if edge_dst in self.independent_nodes:
+                    continue
+
+                # Now split `out_edge` such that it passes through the new inner entry.
+                #  We do not need to modify the subsets, i.e. replacing the variable
+                #  on which we block, because the node is independent and the outgoing
+                #  new inner map entry iterate over the blocked variable.
+                new_map_conn = inner_entry.next_connector()
+                dace_helpers.redirect_edge(
+                    state=state,
+                    edge=out_edge,
+                    new_dst=inner_entry,
+                    new_dst_conn="IN_" + new_map_conn,
+                )
+                # TODO(phimuell): Check if there might be a subset error.
+                state.add_edge(
+                    inner_entry,
+                    "OUT_" + new_map_conn,
+                    out_edge.dst,
+                    out_edge.dst_conn,
+                    copy.deepcopy(out_edge.data),
+                )
+                inner_entry.add_in_connector("IN_" + new_map_conn)
+                inner_entry.add_out_connector("OUT_" + new_map_conn)
+
+        # Now we handle the dependent nodes, they differ from the independent nodes
+        #  in that they _after_ the new inner map entry. Thus, we will modify incoming edges.
+        for dependent_node in self.dependent_nodes:
+            for in_edge in state.in_edges(dependent_node):
+                edge_src: dace_nodes.Node = in_edge.src
+
+                # Since the independent nodes were already processed, and they process
+                #  their output we have to check for this. We do this by checking if
+                #  the source of the edge is the new inner map entry.
+                if edge_src is inner_entry:
+                    assert dependent_node in relocated_nodes
+                    continue
+
+                # A dependent node has at least one connection to the outer map entry.
+                #  And these are the only connections that we must handle, since other
+                #  connections come from independent nodes, and were already handled
+                #  or are inner nodes.
+                if edge_src is not outer_entry:
+                    continue
+
+                # If we encounter an empty Memlet we just just attach it to the
+                #  new inner map entry. Note the partition function ensures that
+                #  either all edges are empty or non.
+                if in_edge.data.is_empty():
+                    assert (
+                        edge_src is outer_entry
+                    ), f"Found an empty edge that does not go to the outer map entry, but to '{edge_src}'."
+                    dace_helpers.redirect_edge(state=state, edge=in_edge, new_src=inner_entry)
+                    continue
+
+                # Because of the definition of a dependent node and the processing
+                #  order, their incoming edges either point to the outer map or
+                #  are already handled.
+                assert (
+                    edge_src is outer_entry
+                ), f"Expected to find source '{outer_entry}' but found '{edge_src}'."
+                edge_conn: str = in_edge.src_conn[4:]
+
+                # Must be before the handling of the modification below
+                #  Note that this will remove the original edge from the SDFG.
+                dace_helpers.redirect_edge(
+                    state=state,
+                    edge=in_edge,
+                    new_src=inner_entry,
+                    new_src_conn="OUT_" + edge_conn,
+                )
+
+                # In a valid SDFG only one edge can go into an input connector of a Map.
+                if "IN_" + edge_conn in inner_entry.in_connectors:
+                    # We have found this edge multiple times already.
+                    #  To ensure that there is no error, we will create a new
+                    #  Memlet that reads the whole array.
+                    piping_edge = next(state.in_edges_by_connector(inner_entry, "IN_" + edge_conn))
+                    data_name = piping_edge.data.data
+                    piping_edge.data = dace.Memlet.from_array(
+                        data_name, sdfg.arrays[data_name], piping_edge.data.wcr
+                    )
+
+                else:
+                    # This is the first time we found this connection.
+                    #  so we just create the edge.
+                    state.add_edge(
+                        outer_entry,
+                        "OUT_" + edge_conn,
+                        inner_entry,
+                        "IN_" + edge_conn,
+                        copy.deepcopy(in_edge.data),
+                    )
+                    inner_entry.add_in_connector("IN_" + edge_conn)
+                    inner_entry.add_out_connector("OUT_" + edge_conn)
+
+        # In certain cases it might happen that we need to create an empty
+        #  Memlet between the outer map entry and the inner one.
+        if state.in_degree(inner_entry) == 0:
+            state.add_edge(
+                outer_entry,
+                None,
+                inner_entry,
+                None,
+                dace.Memlet(),
+            )
+
+        # Handle the Map exits
+        #  This is simple reconnecting, there would be possibilities for improvements
+        #  but we do not use them for now.
+        for in_edge in state.in_edges(outer_exit):
+            edge_conn = in_edge.dst_conn[3:]
+            dace_helpers.redirect_edge(
+                state=state,
+                edge=in_edge,
+                new_dst=inner_exit,
+                new_dst_conn="IN_" + edge_conn,
+            )
+            state.add_edge(
+                inner_exit,
+                "OUT_" + edge_conn,
+                outer_exit,
+                in_edge.dst_conn,
+                copy.deepcopy(in_edge.data),
+            )
+            inner_exit.add_in_connector("IN_" + edge_conn)
+            inner_exit.add_out_connector("OUT_" + edge_conn)
+
+        # TODO(phimuell): Use a less expensive method.
+        dace.sdfg.propagation.propagate_memlets_state(sdfg, state)
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_fusion_helper.py b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_fusion_helper.py
new file mode 100644
index 00000000..ec33e7ea
--- /dev/null
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_fusion_helper.py
@@ -0,0 +1,566 @@
+# 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
+
+"""Implements helper functions for the map fusion transformations.
+
+Note:
+    After DaCe [PR#1629](https://github.com/spcl/dace/pull/1629), that implements
+    a better map fusion transformation is merged, this file will be deleted.
+"""
+
+import functools
+import itertools
+from typing import Any, Optional, Sequence, Union
+
+import dace
+from dace import (
+    data as dace_data,
+    properties as dace_properties,
+    subsets as dace_subsets,
+    transformation as dace_transformation,
+)
+from dace.sdfg import graph as dace_graph, nodes as dace_nodes, validation as dace_validation
+from dace.transformation import helpers as dace_helpers
+
+from gt4py.next.program_processors.runners.dace_fieldview.transformations import util
+
+
+@dace_properties.make_properties
+class MapFusionHelper(dace_transformation.SingleStateTransformation):
+    """Contains common part of the fusion for parallel and serial Map fusion.
+
+    The transformation assumes that the SDFG obeys the principals outlined in
+    [ADR0018](https://github.com/GridTools/gt4py/tree/main/docs/development/ADRs/0018-Canonical_SDFG_in_GT4Py_Transformations.md).
+    The main advantage of this structure is, that it is rather easy to determine
+    if a transient is used anywhere else. This check, performed by
+    `is_interstate_transient()`. It is further speeded up by cashing some computation,
+    thus such an object should not be used after interstate optimizations were applied
+    to the SDFG.
+
+    Args:
+        only_inner_maps: Only match Maps that are internal, i.e. inside another Map.
+        only_toplevel_maps: Only consider Maps that are at the top.
+    """
+
+    only_toplevel_maps = dace_properties.Property(
+        dtype=bool,
+        default=False,
+        allow_none=False,
+        desc="Only perform fusing if the Maps are in the top level.",
+    )
+    only_inner_maps = dace_properties.Property(
+        dtype=bool,
+        default=False,
+        allow_none=False,
+        desc="Only perform fusing if the Maps are inner Maps, i.e. does not have top level scope.",
+    )
+    shared_transients = dace_properties.DictProperty(
+        key_type=dace.SDFG,
+        value_type=set[str],
+        default=None,
+        allow_none=True,
+        desc="Maps SDFGs to the set of array transients that can not be removed. "
+        "The variable acts as a cache, and is managed by 'is_interstate_transient()'.",
+    )
+
+    def __init__(
+        self,
+        only_inner_maps: Optional[bool] = None,
+        only_toplevel_maps: Optional[bool] = None,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(**kwargs)
+        if only_toplevel_maps is not None:
+            self.only_toplevel_maps = bool(only_toplevel_maps)
+        if only_inner_maps is not None:
+            self.only_inner_maps = bool(only_inner_maps)
+        self.shared_transients = {}
+
+    @classmethod
+    def expressions(cls) -> bool:
+        raise RuntimeError("The `_MapFusionHelper` is not a transformation on its own.")
+
+    def can_be_fused(
+        self,
+        map_entry_1: dace_nodes.MapEntry,
+        map_entry_2: dace_nodes.MapEntry,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        sdfg: dace.SDFG,
+        permissive: bool = False,
+    ) -> bool:
+        """Performs basic checks if the maps can be fused.
+
+        This function only checks constrains that are common between serial and
+        parallel map fusion process, which includes:
+        - The scope of the maps.
+        - The scheduling of the maps.
+        - The map parameters.
+
+        However, for performance reasons, the function does not check if the node
+        decomposition exists.
+
+        Args:
+            map_entry_1: The entry of the first (in serial case the top) map.
+            map_exit_2: The entry of the second (in serial case the bottom) map.
+            graph: The SDFGState in which the maps are located.
+            sdfg: The SDFG itself.
+            permissive: Currently unused.
+        """
+        if self.only_inner_maps and self.only_toplevel_maps:
+            raise ValueError("You specified both `only_inner_maps` and `only_toplevel_maps`.")
+
+        # Ensure that both have the same schedule
+        if map_entry_1.map.schedule != map_entry_2.map.schedule:
+            return False
+
+        # Fusing is only possible if the two entries are in the same scope.
+        scope = graph.scope_dict()
+        if scope[map_entry_1] != scope[map_entry_2]:
+            return False
+        elif self.only_inner_maps:
+            if scope[map_entry_1] is None:
+                return False
+        elif self.only_toplevel_maps:
+            if scope[map_entry_1] is not None:
+                return False
+            # TODO(phimuell): Figuring out why this is here.
+            elif util.is_nested_sdfg(sdfg):
+                return False
+
+        # We will now check if there exists a "remapping" that we can use.
+        #  NOTE: The serial map promoter depends on the fact that this is the
+        #           last check.
+        if not self.map_parameter_compatible(
+            map_1=map_entry_1.map, map_2=map_entry_2.map, state=graph, sdfg=sdfg
+        ):
+            return False
+
+        return True
+
+    @staticmethod
+    def relocate_nodes(
+        from_node: Union[dace_nodes.MapExit, dace_nodes.MapEntry],
+        to_node: Union[dace_nodes.MapExit, dace_nodes.MapEntry],
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> None:
+        """Move the connectors and edges from `from_node` to `to_nodes` node.
+
+        This function will only rewire the edges, it does not remove the nodes
+        themselves. Furthermore, this function should be called twice per Map,
+        once for the entry and then for the exit.
+        While it does not remove the node themselves if guarantees that the
+        `from_node` has degree zero.
+
+        Args:
+            from_node: Node from which the edges should be removed.
+            to_node: Node to which the edges should reconnect.
+            state: The state in which the operation happens.
+            sdfg: The SDFG that is modified.
+        """
+
+        # Now we relocate empty Memlets, from the `from_node` to the `to_node`
+        for empty_edge in filter(lambda e: e.data.is_empty(), state.out_edges(from_node)):
+            dace_helpers.redirect_edge(state, empty_edge, new_src=to_node)
+        for empty_edge in filter(lambda e: e.data.is_empty(), state.in_edges(from_node)):
+            dace_helpers.redirect_edge(state, empty_edge, new_dst=to_node)
+
+        # We now ensure that there is only one empty Memlet from the `to_node` to any other node.
+        #  Although it is allowed, we try to prevent it.
+        empty_targets: set[dace_nodes.Node] = set()
+        for empty_edge in filter(lambda e: e.data.is_empty(), state.all_edges(to_node)):
+            if empty_edge.dst in empty_targets:
+                state.remove_edge(empty_edge)
+            empty_targets.add(empty_edge.dst)
+
+        # We now determine which edges we have to migrate, for this we are looking at
+        #  the incoming edges, because this allows us also to detect dynamic map ranges.
+        for edge_to_move in state.in_edges(from_node):
+            assert isinstance(edge_to_move.dst_conn, str)
+
+            if not edge_to_move.dst_conn.startswith("IN_"):
+                # Dynamic Map Range
+                #  The connector name simply defines a variable name that is used,
+                #  inside the Map scope to define a variable. We handle it directly.
+                dmr_symbol = edge_to_move.dst_conn
+
+                # TODO(phimuell): Check if the symbol is really unused in the target scope.
+                if dmr_symbol in to_node.in_connectors:
+                    raise NotImplementedError(
+                        f"Tried to move the dynamic map range '{dmr_symbol}' from {from_node}'"
+                        f" to '{to_node}', but the symbol is already known there, but the"
+                        " renaming is not implemented."
+                    )
+                if not to_node.add_in_connector(dmr_symbol, force=False):
+                    raise RuntimeError(  # Might fail because of out connectors.
+                        f"Failed to add the dynamic map range symbol '{dmr_symbol}' to '{to_node}'."
+                    )
+                dace_helpers.redirect_edge(state=state, edge=edge_to_move, new_dst=to_node)
+                from_node.remove_in_connector(dmr_symbol)
+
+                # There is no other edge that we have to consider, so we just end here
+                continue
+
+            # We have a Passthrough connection, i.e. there exists a matching `OUT_`.
+            old_conn = edge_to_move.dst_conn[3:]  # The connection name without prefix
+            new_conn = to_node.next_connector(old_conn)
+
+            to_node.add_in_connector("IN_" + new_conn)
+            for e in state.in_edges_by_connector(from_node, "IN_" + old_conn):
+                dace_helpers.redirect_edge(state, e, new_dst=to_node, new_dst_conn="IN_" + new_conn)
+            to_node.add_out_connector("OUT_" + new_conn)
+            for e in state.out_edges_by_connector(from_node, "OUT_" + old_conn):
+                dace_helpers.redirect_edge(
+                    state, e, new_src=to_node, new_src_conn="OUT_" + new_conn
+                )
+            from_node.remove_in_connector("IN_" + old_conn)
+            from_node.remove_out_connector("OUT_" + old_conn)
+
+        # Check if we succeeded.
+        if state.out_degree(from_node) != 0:
+            raise dace_validation.InvalidSDFGError(
+                f"Failed to relocate the outgoing edges from `{from_node}`, there are still `{state.out_edges(from_node)}`",
+                sdfg,
+                sdfg.node_id(state),
+            )
+        if state.in_degree(from_node) != 0:
+            raise dace_validation.InvalidSDFGError(
+                f"Failed to relocate the incoming edges from `{from_node}`, there are still `{state.in_edges(from_node)}`",
+                sdfg,
+                sdfg.node_id(state),
+            )
+        assert len(from_node.in_connectors) == 0
+        assert len(from_node.out_connectors) == 0
+
+    @staticmethod
+    def map_parameter_compatible(
+        map_1: dace_nodes.Map,
+        map_2: dace_nodes.Map,
+        state: Union[dace.SDFGState, dace.SDFG],
+        sdfg: dace.SDFG,
+    ) -> bool:
+        """Checks if the parameters of `map_1` are compatible with `map_2`.
+
+        The check follows the following rules:
+        - The names of the map variables must be the same, i.e. no renaming
+            is performed.
+        - The ranges must be the same.
+        """
+        range_1: dace_subsets.Range = map_1.range
+        params_1: Sequence[str] = map_1.params
+        range_2: dace_subsets.Range = map_2.range
+        params_2: Sequence[str] = map_2.params
+
+        # The maps are only fuseable if we have an exact match in the parameter names
+        #  this is because we do not do any renaming. This is in accordance with the
+        #  rules.
+        if set(params_1) != set(params_2):
+            return False
+
+        # Maps the name of a parameter to the dimension index
+        param_dim_map_1: dict[str, int] = {pname: i for i, pname in enumerate(params_1)}
+        param_dim_map_2: dict[str, int] = {pname: i for i, pname in enumerate(params_2)}
+
+        # To fuse the two maps the ranges must have the same ranges
+        for pname in params_1:
+            idx_1 = param_dim_map_1[pname]
+            idx_2 = param_dim_map_2[pname]
+            # TODO(phimuell): do we need to call simplify?
+            if range_1[idx_1] != range_2[idx_2]:
+                return False
+
+        return True
+
+    def is_interstate_transient(
+        self,
+        transient: Union[str, dace_nodes.AccessNode],
+        sdfg: dace.SDFG,
+        state: dace.SDFGState,
+    ) -> bool:
+        """Tests if `transient` is an interstate transient, an can not be removed.
+
+        Essentially this function checks if a transient might be needed in a
+        different state in the SDFG, because it transmit information from
+        one state to the other.
+        If only the name of the data container is passed the function will
+        first look for an corresponding access node.
+
+        The set of these "interstate transients" is computed once per SDFG.
+        The result is then cached internally for later reuse.
+
+        Args:
+            transient: The transient that should be checked.
+            sdfg: The SDFG containing the array.
+            state: If given the state the node is located in.
+        """
+
+        # The following builds upon the HACK MD document and not on ADR0018.
+        #   Therefore the numbers are slightly different, but both documents
+        #   essentially describes the same SDFG.
+        # According to [rule 6](https://hackmd.io/klvzLnzMR6GZBWtRU8HbDg#Requirements-on-SDFG)
+        #  the set of such transients is partially given by all source access dace_nodes.
+        #  Because of rule 3 we also include all scalars in this set, as an over
+        #  approximation. Furthermore, because simplify might violate rule 3,
+        #  we also include the sink dace_nodes.
+
+        # See if we have already computed the set
+        if sdfg in self.shared_transients:
+            shared_sdfg_transients: set[str] = self.shared_transients[sdfg]
+        else:
+            # SDFG is not known so we have to compute the set.
+            shared_sdfg_transients = set()
+            for state_to_scan in sdfg.all_states():
+                # TODO(phimuell): Use `all_nodes_recursive()` once it is available.
+                shared_sdfg_transients.update(
+                    [
+                        node.data
+                        for node in itertools.chain(
+                            state_to_scan.source_nodes(), state_to_scan.sink_nodes()
+                        )
+                        if isinstance(node, dace_nodes.AccessNode)
+                        and sdfg.arrays[node.data].transient
+                    ]
+                )
+            self.shared_transients[sdfg] = shared_sdfg_transients
+
+        if isinstance(transient, str):
+            name = transient
+            matching_access_nodes = [node for node in state.data_nodes() if node.data == name]
+            # Rule 8: There is only one access node per state for data.
+            assert len(matching_access_nodes) == 1
+            transient = matching_access_nodes[0]
+        else:
+            assert isinstance(transient, dace_nodes.AccessNode)
+            name = transient.data
+
+        desc: dace_data.Data = sdfg.arrays[name]
+        if not desc.transient:
+            return True
+        if isinstance(desc, dace_data.Scalar):
+            return True  # Scalars can not be removed by fusion anyway.
+
+        # Rule 8: If degree larger than one then it is used within the state.
+        if state.out_degree(transient) > 1:
+            return True
+
+        # Now we check if it is used in a different state.
+        return name in shared_sdfg_transients
+
+    def partition_first_outputs(
+        self,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+        map_exit_1: dace_nodes.MapExit,
+        map_entry_2: dace_nodes.MapEntry,
+    ) -> Union[
+        tuple[
+            set[dace_graph.MultiConnectorEdge[dace.Memlet]],
+            set[dace_graph.MultiConnectorEdge[dace.Memlet]],
+            set[dace_graph.MultiConnectorEdge[dace.Memlet]],
+        ],
+        None,
+    ]:
+        """Partition the output edges of `map_exit_1` for serial map fusion.
+
+        The output edges of the first map are partitioned into three distinct sets,
+        defined as follows:
+
+        - Pure Output Set `\mathbb{P}`:
+            These edges exits the first map and does not enter the second map. These
+            outputs will be simply be moved to the output of the second map.
+        - Exclusive Intermediate Set `\mathbb{E}`:
+            Edges in this set leaves the first map exit, enters an access node, from
+            where a Memlet then leads immediately to the second map. The memory
+            referenced by this access node is not used anywhere else, thus it can
+            be removed.
+        - Shared Intermediate Set `\mathbb{S}`:
+            These edges are very similar to the one in `\mathbb{E}` except that they
+            are used somewhere else, thus they can not be removed and have to be
+            recreated as output of the second map.
+
+        Returns:
+            If such a decomposition exists the function will return the three sets
+            mentioned above in the same order.
+            In case the decomposition does not exist, i.e. the maps can not be fused
+            the function returns `None`.
+
+        Args:
+            state: The in which the two maps are located.
+            sdfg: The full SDFG in whcih we operate.
+            map_exit_1: The exit node of the first map.
+            map_entry_2: The entry node of the second map.
+        """
+        # The three outputs set.
+        pure_outputs: set[dace_graph.MultiConnectorEdge[dace.Memlet]] = set()
+        exclusive_outputs: set[dace_graph.MultiConnectorEdge[dace.Memlet]] = set()
+        shared_outputs: set[dace_graph.MultiConnectorEdge[dace.Memlet]] = set()
+
+        # Set of intermediate nodes that we have already processed.
+        processed_inter_nodes: set[dace_nodes.Node] = set()
+
+        # Now scan all output edges of the first exit and classify them
+        for out_edge in state.out_edges(map_exit_1):
+            intermediate_node: dace_nodes.Node = out_edge.dst
+
+            # We already processed the node, this should indicate that we should
+            #  run simplify again, or we should start implementing this case.
+            if intermediate_node in processed_inter_nodes:
+                return None
+            processed_inter_nodes.add(intermediate_node)
+
+            # Now let's look at all nodes that are downstream of the intermediate node.
+            #  This, among other things, will tell us, how we have to handle this node.
+            downstream_nodes = util.all_nodes_between(
+                graph=state,
+                begin=intermediate_node,
+                end=map_entry_2,
+            )
+
+            # If `downstream_nodes` is `None` this means that `map_entry_2` was never
+            #  reached, thus `intermediate_node` does not enter the second map and
+            #  the node is a pure output node.
+            if downstream_nodes is None:
+                pure_outputs.add(out_edge)
+                continue
+
+            # The following tests are _after_ we have determined if we have a pure
+            #  output node, because this allows us to handle more exotic pure node
+            #  cases, as handling them is essentially rerouting an edge, whereas
+            #  handling intermediate nodes is much more complicated.
+
+            # Empty Memlets are only allowed if they are in `\mathbb{P}`, which
+            #  is also the only place they really make sense (for a map exit).
+            #  Thus if we now found an empty Memlet we reject it.
+            if out_edge.data.is_empty():
+                return None
+
+            # In case the intermediate has more than one entry, all must come from the
+            #  first map, otherwise we can not fuse them. Currently we restrict this
+            #  even further by saying that it has only one incoming Memlet.
+            if state.in_degree(intermediate_node) != 1:
+                return None
+
+            # It can happen that multiple edges converges at the `IN_` connector
+            #  of the first map exit, but there is only one edge leaving the exit.
+            #  It is complicate to handle this, so for now we ignore it.
+            # TODO(phimuell): Handle this case properly.
+            inner_collector_edges = list(
+                state.in_edges_by_connector(intermediate_node, "IN_" + out_edge.src_conn[3:])
+            )
+            if len(inner_collector_edges) > 1:
+                return None
+
+            # For us an intermediate node must always be an access node, because
+            #  everything else we do not know how to handle. It is important that
+            #  we do not test for non transient data here, because they can be
+            #  handled has shared intermediates.
+            if not isinstance(intermediate_node, dace_nodes.AccessNode):
+                return None
+            intermediate_desc: dace_data.Data = intermediate_node.desc(sdfg)
+            if isinstance(intermediate_desc, dace_data.View):
+                return None
+
+            # There are some restrictions we have on intermediate dace_nodes. The first one
+            #  is that we do not allow WCR, this is because they need special handling
+            #  which is currently not implement (the DaCe transformation has this
+            #  restriction as well). The second one is that we can reduce the
+            #  intermediate node and only feed a part into the second map, consider
+            #  the case `b = a + 1; return b + 2`, where we have arrays. In this
+            #  example only a single element must be available to the second map.
+            #  However, this is hard to check so we will make a simplification.
+            #  First, we will not check it at the producer, but at the consumer point.
+            #  There we assume if the consumer does _not consume the whole_
+            #  intermediate array, then we can decompose the intermediate, by setting
+            #  the map iteration index to zero and recover the shape, see
+            #  implementation in the actual fusion routine.
+            #  This is an assumption that is in most cases correct, but not always.
+            #  However, doing it correctly is extremely complex.
+            for _, produce_edge in util.find_upstream_producers(state, out_edge):
+                if produce_edge.data.wcr is not None:
+                    return None
+
+            if len(downstream_nodes) == 0:
+                # There is nothing between intermediate node and the entry of the
+                #  second map, thus the edge belongs either in `\mathbb{S}` or
+                #  `\mathbb{E}`.
+
+                # This is a very special situation, i.e. the access node has many
+                #  different connections to the second map entry, this is a special
+                #  case that we do not handle.
+                # TODO(phimuell): Handle this case.
+                if state.out_degree(intermediate_node) != 1:
+                    return None
+
+                # Certain nodes need more than one element as input. As explained
+                #  above, in this situation we assume that we can naturally decompose
+                #  them iff the node does not consume that whole intermediate.
+                #  Furthermore, it can not be a dynamic map range or a library node.
+                intermediate_size = functools.reduce(lambda a, b: a * b, intermediate_desc.shape)
+                consumers = util.find_downstream_consumers(state=state, begin=intermediate_node)
+                for consumer_node, feed_edge in consumers:
+                    # TODO(phimuell): Improve this approximation.
+                    if (
+                        intermediate_size != 1
+                    ) and feed_edge.data.num_elements() == intermediate_size:
+                        return None
+                    if consumer_node is map_entry_2:  # Dynamic map range.
+                        return None
+                    if isinstance(consumer_node, dace_nodes.LibraryNode):
+                        # TODO(phimuell): Allow some library dace_nodes.
+                        return None
+
+                # Note that "remove" has a special meaning here, regardless of the
+                #  output of the check function, from within the second map we remove
+                #  the intermediate, it has more the meaning of "do we need to
+                #  reconstruct it after the second map again?"
+                if self.is_interstate_transient(intermediate_node, sdfg, state):
+                    shared_outputs.add(out_edge)
+                else:
+                    exclusive_outputs.add(out_edge)
+                continue
+
+            else:
+                # There is not only a single connection from the intermediate node to
+                #  the second map, but the intermediate has more connections, thus
+                #  the node might belong to the shared output. Of the many different
+                #  possibilities, we only consider a single case:
+                #  - The intermediate has a single connection to the second map, that
+                #       fulfills the restriction outlined above.
+                #  - All other connections have no connection to the second map.
+                found_second_entry = False
+                intermediate_size = functools.reduce(lambda a, b: a * b, intermediate_desc.shape)
+                for edge in state.out_edges(intermediate_node):
+                    if edge.dst is map_entry_2:
+                        if found_second_entry:  # The second map was found again.
+                            return None
+                        found_second_entry = True
+                        consumers = util.find_downstream_consumers(state=state, begin=edge)
+                        for consumer_node, feed_edge in consumers:
+                            if feed_edge.data.num_elements() == intermediate_size:
+                                return None
+                            if consumer_node is map_entry_2:  # Dynamic map range
+                                return None
+                            if isinstance(consumer_node, dace_nodes.LibraryNode):
+                                # TODO(phimuell): Allow some library dace_nodes.
+                                return None
+                    else:
+                        # Ensure that there is no path that leads to the second map.
+                        after_intermdiate_node = util.all_nodes_between(
+                            graph=state, begin=edge.dst, end=map_entry_2
+                        )
+                        if after_intermdiate_node is not None:
+                            return None
+                # If we are here, then we know that the node is a shared output
+                shared_outputs.add(out_edge)
+                continue
+
+        assert exclusive_outputs or shared_outputs or pure_outputs
+        assert len(processed_inter_nodes) == sum(
+            len(x) for x in [pure_outputs, exclusive_outputs, shared_outputs]
+        )
+        return (pure_outputs, exclusive_outputs, shared_outputs)
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_orderer.py b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_orderer.py
new file mode 100644
index 00000000..4b34dd6a
--- /dev/null
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_orderer.py
@@ -0,0 +1,125 @@
+# 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
+
+from typing import Any, Optional, Sequence, Union
+
+import dace
+from dace import properties as dace_properties, transformation as dace_transformation
+from dace.sdfg import nodes as dace_nodes
+
+from gt4py.next import common as gtx_common
+from gt4py.next.program_processors.runners.dace_fieldview import utility as gtx_dace_fieldview_util
+
+
+@dace_properties.make_properties
+class MapIterationOrder(dace_transformation.SingleStateTransformation):
+    """Modify the order of the iteration variables.
+
+    The iteration order, while irrelevant from an SDFG point of view, is highly
+    relevant in code, and the fastest varying index ("inner most loop" in CPU or
+    "x block dimension" in GPU) should be associated with the stride 1 dimension
+    of the array.
+    This transformation will reorder the map indexes such that this is the case.
+
+    While the place of the leading dimension is clearly defined, the order of the
+    other loop indexes, after this transformation is unspecified.
+
+    Args:
+        leading_dim: A GT4Py dimension object that identifies the dimension that
+            is supposed to have stride 1.
+
+    Note:
+        The transformation does follow the rules outlines in
+        [ADR0018](https://github.com/GridTools/gt4py/tree/main/docs/development/ADRs/0018-Canonical_SDFG_in_GT4Py_Transformations.md)
+        especially rule 11, regarding the names.
+
+    Todo:
+        - Extend that different dimensions can be specified to be leading
+            dimensions, with some priority mechanism.
+        - Maybe also process the parameters to bring them in a canonical order.
+    """
+
+    leading_dim = dace_properties.Property(
+        dtype=str,
+        allow_none=True,
+        desc="Dimension that should become the leading dimension.",
+    )
+
+    map_entry = dace_transformation.transformation.PatternNode(dace_nodes.MapEntry)
+
+    def __init__(
+        self,
+        leading_dim: Optional[Union[gtx_common.Dimension, str]] = None,
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
+        if isinstance(leading_dim, gtx_common.Dimension):
+            self.leading_dim = gtx_dace_fieldview_util.get_map_variable(leading_dim)
+        elif leading_dim is not None:
+            self.leading_dim = leading_dim
+
+    @classmethod
+    def expressions(cls) -> Any:
+        return [dace.sdfg.utils.node_path_graph(cls.map_entry)]
+
+    def can_be_applied(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        expr_index: int,
+        sdfg: dace.SDFG,
+        permissive: bool = False,
+    ) -> bool:
+        """Test if the map can be reordered.
+
+        Essentially the function checks if the selected dimension is inside the map,
+        and if so, if it is on the right place.
+        """
+
+        if self.leading_dim is None:
+            return False
+        map_entry: dace_nodes.MapEntry = self.map_entry
+        map_params: Sequence[str] = map_entry.map.params
+        map_var: str = self.leading_dim
+
+        if map_var not in map_params:
+            return False
+        if map_params[-1] == map_var:  # Already at the correct location
+            return False
+        return True
+
+    def apply(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        sdfg: dace.SDFG,
+    ) -> None:
+        """Performs the actual parameter reordering.
+
+        The function will make the map variable, that corresponds to
+        `self.leading_dim` the last map variable (this is given by the structure of
+        DaCe's code generator).
+        """
+        map_entry: dace_nodes.MapEntry = self.map_entry
+        map_params: list[str] = map_entry.map.params
+        map_var: str = self.leading_dim
+
+        # This implementation will just swap the variable that is currently the last
+        #  with the one that should be the last.
+        dst_idx = -1
+        src_idx = map_params.index(map_var)
+
+        for to_process in [
+            map_entry.map.params,
+            map_entry.map.range.ranges,
+            map_entry.map.range.tile_sizes,
+        ]:
+            assert isinstance(to_process, list)
+            src_val = to_process[src_idx]
+            dst_val = to_process[dst_idx]
+            to_process[dst_idx] = src_val
+            to_process[src_idx] = dst_val
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_promoter.py b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_promoter.py
new file mode 100644
index 00000000..19818fd3
--- /dev/null
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_promoter.py
@@ -0,0 +1,427 @@
+# 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 copy
+from typing import Any, Mapping, Optional, Union
+
+import dace
+from dace import (
+    properties as dace_properties,
+    subsets as dace_subsets,
+    transformation as dace_transformation,
+)
+from dace.sdfg import nodes as dace_nodes
+
+from gt4py.next.program_processors.runners.dace_fieldview import (
+    transformations as gtx_transformations,
+)
+
+
+@dace_properties.make_properties
+class BaseMapPromoter(dace_transformation.SingleStateTransformation):
+    """Base transformation to add certain missing dimension to a map.
+
+    By adding certain dimension to a Map, it might became possible to use the Map
+    in more transformations. This class acts as a base and the actual matching and
+    checking must be implemented by a concrete implementation.
+    But it provides some basic check functionality and the actual promotion logic.
+
+    The transformation operates on two Maps, first the "source map". This map
+    describes the Map that should be used as template. The second one is "map to
+    promote". After the transformation the "map to promote" will have the same
+    map parameter as the "source map" has.
+
+    In order to properly work, the parameters of "source map" must be a strict
+    superset of the ones of "map to promote". Furthermore, this transformation
+    builds upon the structure defined [ADR0018](https://github.com/GridTools/gt4py/tree/main/docs/development/ADRs/0018-Canonical_SDFG_in_GT4Py_Transformations.md)
+    especially rule 11, regarding the names. Thus the function uses the names
+    of the map parameters to determine what a local, horizontal, vertical or
+    unknown dimension is. It also uses rule 12, therefore it will not perform
+    a renaming and iteration variables must be a match.
+
+    To influence what to promote the user must implement the `map_to_promote()`
+    and `source_map()` function. They have to return the map entry node.
+
+    The order of the parameter the map to promote has is unspecific, while the
+    source map is not modified.
+
+    Args:
+        only_inner_maps: Only match Maps that are internal, i.e. inside another Map.
+        only_toplevel_maps: Only consider Maps that are at the top.
+        promote_vertical: If `True` promote vertical dimensions, i.e. add them
+            to the map to promote; `True` by default.
+        promote_local: If `True` promote local dimensions, i.e. add them to the
+            map to promote; `True` by default.
+        promote_horizontal: If `True` promote horizontal dimensions, i.e. add
+            them to the map to promote; `False` by default.
+        promote_all: Do not impose any restriction on what to promote. The only
+            reasonable value is `True` or `None`.
+
+    Note:
+        This ignores tiling.
+    """
+
+    only_toplevel_maps = dace_properties.Property(
+        dtype=bool,
+        default=False,
+        allow_none=False,
+        desc="Only perform fusing if the Maps are on the top level.",
+    )
+    only_inner_maps = dace_properties.Property(
+        dtype=bool,
+        default=False,
+        allow_none=False,
+        desc="Only perform fusing if the Maps are inner Maps, i.e. does not have top level scope.",
+    )
+    promote_vertical = dace_properties.Property(
+        dtype=bool,
+        default=True,
+        desc="If `True` promote vertical dimensions.",
+    )
+    promote_local = dace_properties.Property(
+        dtype=bool,
+        default=True,
+        desc="If `True` promote local dimensions.",
+    )
+    promote_horizontal = dace_properties.Property(
+        dtype=bool,
+        default=False,
+        desc="If `True` promote horizontal dimensions.",
+    )
+    promote_all = dace_properties.Property(
+        dtype=bool,
+        default=False,
+        desc="If `True` perform any promotion. Takes precedence over all other selectors.",
+    )
+
+    def map_to_promote(
+        self,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> dace_nodes.MapEntry:
+        """Returns the map entry that should be promoted."""
+        raise NotImplementedError(f"{type(self).__name__} must implement 'map_to_promote'.")
+
+    def source_map(
+        self,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> dace_nodes.MapEntry:
+        """Returns the map entry that is used as source/template."""
+        raise NotImplementedError(f"{type(self).__name__} must implement 'source_map'.")
+
+    @classmethod
+    def expressions(cls) -> Any:
+        raise TypeError("You must implement 'expressions' yourself.")
+
+    def __init__(
+        self,
+        only_inner_maps: Optional[bool] = None,
+        only_toplevel_maps: Optional[bool] = None,
+        promote_local: Optional[bool] = None,
+        promote_vertical: Optional[bool] = None,
+        promote_horizontal: Optional[bool] = None,
+        promote_all: Optional[bool] = None,
+        *args: Any,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(*args, **kwargs)
+        if only_inner_maps is not None:
+            self.only_inner_maps = bool(only_inner_maps)
+        if only_toplevel_maps is not None:
+            self.only_toplevel_maps = bool(only_toplevel_maps)
+        if promote_local is not None:
+            self.promote_local = bool(promote_local)
+        if promote_vertical is not None:
+            self.promote_vertical = bool(promote_vertical)
+        if promote_horizontal is not None:
+            self.promote_horizontal = bool(promote_horizontal)
+        if promote_all is not None:
+            self.promote_all = bool(promote_all)
+            self.promote_horizontal = False
+            self.promote_vertical = False
+            self.promote_local = False
+        if only_inner_maps and only_toplevel_maps:
+            raise ValueError("You specified both `only_inner_maps` and `only_toplevel_maps`.")
+        if not (
+            self.promote_local
+            or self.promote_vertical
+            or self.promote_horizontal
+            or self.promote_all
+        ):
+            raise ValueError(
+                "You must select at least one class of dimension that should be promoted."
+            )
+
+    def can_be_applied(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        expr_index: int,
+        sdfg: dace.SDFG,
+        permissive: bool = False,
+    ) -> bool:
+        """Perform some basic structural tests on the map.
+
+        A subclass should call this function before checking anything else. If a
+        subclass has not called this function, the behaviour will be undefined.
+        The function checks:
+        - If the map to promote is in the right scope.
+        - If the parameter of the second map are compatible with each other.
+        - If a dimension would be promoted that should not.
+        """
+        assert self.expr_index == expr_index
+        map_to_promote_entry: dace_nodes.MapEntry = self.map_to_promote(state=graph, sdfg=sdfg)
+        map_to_promote: dace_nodes.Map = map_to_promote_entry.map
+        source_map_entry: dace_nodes.MapEntry = self.source_map(state=graph, sdfg=sdfg)
+        source_map: dace_nodes.Map = source_map_entry.map
+
+        # Test the scope of the promotee.
+        #  Because of the nature of the transformation, it is not needed that the
+        #  two maps are in the same scope. However, they should be in the same state
+        #  to ensure that the symbols are the same and all. But this is guaranteed by
+        #  the nature of this transformation (single state).
+        if self.only_inner_maps or self.only_toplevel_maps:
+            scope_dict: Mapping[dace_nodes.Node, Union[dace_nodes.Node, None]] = graph.scope_dict()
+            if self.only_inner_maps and (scope_dict[map_to_promote_entry] is None):
+                return False
+            if self.only_toplevel_maps and (scope_dict[map_to_promote_entry] is not None):
+                return False
+
+        # Test if the map ranges and parameter are compatible with each other
+        missing_map_parameters: list[str] | None = self.missing_map_params(
+            map_to_promote=map_to_promote,
+            source_map=source_map,
+            be_strict=True,
+        )
+        if not missing_map_parameters:
+            return False
+
+        # We now know which dimensions we have to add to the promotee map.
+        #  Now we must test if we are also allowed to make that promotion in the first place.
+        if not self.promote_all:
+            dimension_identifier: list[str] = []
+            if self.promote_local:
+                dimension_identifier.append("_gtx_localdim")
+            if self.promote_vertical:
+                dimension_identifier.append("_gtx_vertical")
+            if self.promote_horizontal:
+                dimension_identifier.append("_gtx_horizontal")
+            if not dimension_identifier:
+                return False
+            for missing_map_param in missing_map_parameters:
+                # Check if all missing parameter match a specified pattern. Note
+                #  unknown iteration parameter, such as `__hansi_meier` will be
+                #  rejected and can not be promoted.
+                if not any(
+                    missing_map_param.endswith(dim_identifier)
+                    for dim_identifier in dimension_identifier
+                ):
+                    return False
+
+        return True
+
+    def apply(self, graph: Union[dace.SDFGState, dace.SDFG], sdfg: dace.SDFG) -> None:
+        """Performs the actual Map promoting.
+
+        After this call the map to promote will have the same map parameters
+        and ranges as the source map has. The function assumes that `can_be_applied()`
+        returned `True`.
+        """
+        map_to_promote: dace_nodes.Map = self.map_to_promote(state=graph, sdfg=sdfg).map
+        source_map: dace_nodes.Map = self.source_map(state=graph, sdfg=sdfg).map
+
+        # The simplest implementation is just to copy the important parts.
+        #  Note that we only copy the ranges and parameters all other stuff in the
+        #  associated Map object is not modified.
+        map_to_promote.params = copy.deepcopy(source_map.params)
+        map_to_promote.range = copy.deepcopy(source_map.range)
+
+    def missing_map_params(
+        self,
+        map_to_promote: dace_nodes.Map,
+        source_map: dace_nodes.Map,
+        be_strict: bool = True,
+    ) -> list[str] | None:
+        """Returns the parameter that are missing in the map that should be promoted.
+
+        The returned sequence is empty if they are already have the same parameters.
+        The function will return `None` is promoting is not possible.
+        By setting `be_strict` to `False` the function will only check the names.
+
+        Args:
+            map_to_promote: The map that should be promoted.
+            source_map: The map acting as template.
+            be_strict: Ensure that the ranges that are already there are correct.
+        """
+        source_params_set: set[str] = set(source_map.params)
+        curr_params_set: set[str] = set(map_to_promote.params)
+
+        # The promotion can only work if the source map's parameters
+        #  if a superset of the ones the map that should be promoted is.
+        if not source_params_set.issuperset(curr_params_set):
+            return None
+
+        if be_strict:
+            # Check if the parameters that are already in the map to promote have
+            #  the same range as in the source map.
+            source_ranges: dace_subsets.Range = source_map.range
+            curr_ranges: dace_subsets.Range = map_to_promote.range
+            curr_param_to_idx: dict[str, int] = {p: i for i, p in enumerate(map_to_promote.params)}
+            source_param_to_idx: dict[str, int] = {p: i for i, p in enumerate(source_map.params)}
+            for param_to_check in curr_params_set:
+                curr_range = curr_ranges[curr_param_to_idx[param_to_check]]
+                source_range = source_ranges[source_param_to_idx[param_to_check]]
+                # TODO(phimuell): Use simplify
+                if curr_range != source_range:
+                    return None
+        return list(source_params_set - curr_params_set)
+
+
+@dace_properties.make_properties
+class SerialMapPromoter(BaseMapPromoter):
+    """Promote a map such that it can be fused serially.
+
+    A condition for fusing serial Maps is that they cover the same range. This
+    transformation is able to promote a Map, i.e. adding the missing dimensions,
+    such that the maps can be fused.
+    For more information see the `BaseMapPromoter` class.
+
+    Notes:
+        The transformation does not perform the fusing on its one.
+
+    Todo:
+        The map should do the fusing on its own directly.
+    """
+
+    # Pattern Matching
+    exit_first_map = dace_transformation.transformation.PatternNode(dace_nodes.MapExit)
+    access_node = dace_transformation.transformation.PatternNode(dace_nodes.AccessNode)
+    entry_second_map = dace_transformation.transformation.PatternNode(dace_nodes.MapEntry)
+
+    @classmethod
+    def expressions(cls) -> Any:
+        """Get the match expressions.
+
+        The function generates two match expressions. The first match describes
+        the case where the top map must be promoted, while the second case is
+        the second/lower map must be promoted.
+        """
+        return [
+            dace.sdfg.utils.node_path_graph(
+                cls.exit_first_map, cls.access_node, cls.entry_second_map
+            ),
+            dace.sdfg.utils.node_path_graph(
+                cls.exit_first_map, cls.access_node, cls.entry_second_map
+            ),
+        ]
+
+    def can_be_applied(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        expr_index: int,
+        sdfg: dace.SDFG,
+        permissive: bool = False,
+    ) -> bool:
+        """Tests if the Maps really can be fused."""
+
+        # Test if the promotion could be done.
+        if not super().can_be_applied(graph, expr_index, sdfg, permissive):
+            return False
+
+        # Test if after the promotion the maps could be fused.
+        if not self._test_if_promoted_maps_can_be_fused(graph, sdfg):
+            return False
+
+        return True
+
+    def _test_if_promoted_maps_can_be_fused(
+        self,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> bool:
+        """This function checks if the promoted maps can be fused by map fusion.
+
+        This function assumes that `self.can_be_applied()` returned `True`.
+
+        Args:
+            state: The state in which we operate.
+            sdfg: The SDFG we process.
+
+        Note:
+            The current implementation uses a very hacky way to test this.
+
+        Todo:
+            Find a better way of doing it.
+        """
+        first_map_exit: dace_nodes.MapExit = self.exit_first_map
+        access_node: dace_nodes.AccessNode = self.access_node
+        second_map_entry: dace_nodes.MapEntry = self.entry_second_map
+
+        map_to_promote: dace_nodes.MapEntry = self.map_to_promote(state=state, sdfg=sdfg).map
+
+        # Since we force a promotion of the map we have to store the old parameters
+        #  of the map such that we can later restore them.
+        org_map_to_promote_params = copy.deepcopy(map_to_promote.params)
+        org_map_to_promote_ranges = copy.deepcopy(map_to_promote.range)
+
+        try:
+            # This will lead to a promotion of the map, this is needed that
+            #  Map fusion can actually inspect them.
+            self.apply(graph=state, sdfg=sdfg)
+
+            # Now create the map fusion object that we can then use to check if
+            #  the fusion is possible or not.
+            serial_fuser = gtx_transformations.SerialMapFusion(
+                only_inner_maps=self.only_inner_maps,
+                only_toplevel_maps=self.only_toplevel_maps,
+            )
+            candidate = {
+                type(serial_fuser).map_exit1: first_map_exit,
+                type(serial_fuser).access_node: access_node,
+                type(serial_fuser).map_entry2: second_map_entry,
+            }
+            state_id = sdfg.node_id(state)
+            serial_fuser.setup_match(sdfg, sdfg.cfg_id, state_id, candidate, 0, override=True)
+
+            # Now use the serial fuser to see if fusion would succeed
+            if not serial_fuser.can_be_applied(state, 0, sdfg):
+                return False
+
+        finally:
+            # Restore the parameters of the map that we promoted before.
+            map_to_promote.params = org_map_to_promote_params
+            map_to_promote.range = org_map_to_promote_ranges
+
+        return True
+
+    def map_to_promote(
+        self,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> dace_nodes.MapEntry:
+        if self.expr_index == 0:
+            # The first the top map will be promoted.
+            return state.entry_node(self.exit_first_map)
+        assert self.expr_index == 1
+
+        # The second map will be promoted.
+        return self.entry_second_map
+
+    def source_map(
+        self,
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+    ) -> dace_nodes.MapEntry:
+        """Returns the map entry that is used as source/template."""
+        if self.expr_index == 0:
+            # The first the top map will be promoted, so the second map is the source.
+            return self.entry_second_map
+        assert self.expr_index == 1
+
+        # The second map will be promoted, so the first is used as source
+        return state.entry_node(self.exit_first_map)
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_serial_fusion.py b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_serial_fusion.py
new file mode 100644
index 00000000..bca5aa22
--- /dev/null
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/map_serial_fusion.py
@@ -0,0 +1,483 @@
+# 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
+
+"""Implements the serial map fusing transformation.
+
+Note:
+    After DaCe [PR#1629](https://github.com/spcl/dace/pull/1629), that implements
+    a better map fusion transformation is merged, this file will be deleted.
+"""
+
+import copy
+from typing import Any, Union
+
+import dace
+from dace import (
+    dtypes as dace_dtypes,
+    properties as dace_properties,
+    subsets as dace_subsets,
+    symbolic as dace_symbolic,
+    transformation as dace_transformation,
+)
+from dace.sdfg import graph as dace_graph, nodes as dace_nodes
+
+from gt4py.next.program_processors.runners.dace_fieldview.transformations import map_fusion_helper
+
+
+@dace_properties.make_properties
+class SerialMapFusion(map_fusion_helper.MapFusionHelper):
+    """Specialized replacement for the map fusion transformation that is provided by DaCe.
+
+    As its name is indicating this transformation is only able to handle Maps that
+    are in sequence. Compared to the native DaCe transformation, this one is able
+    to handle more complex cases of connection between the maps. In that sense, it
+    is much more similar to DaCe's `SubgraphFusion` transformation.
+
+    Things that are improved, compared to the native DaCe implementation:
+    - Nested Maps.
+    - Temporary arrays and the correct propagation of their Memlets.
+    - Top Maps that have multiple outputs.
+
+    Conceptually this transformation removes the exit of the first or upper map
+    and the entry of the lower or second map and then rewrites the connections
+    appropriately.
+
+    This transformation assumes that an SDFG obeys the structure that is outlined
+    [here](https://hackmd.io/klvzLnzMR6GZBWtRU8HbDg#Requirements-on-SDFG). For that
+    reason it is not true replacement of the native DaCe transformation.
+
+    Args:
+        only_inner_maps: Only match Maps that are internal, i.e. inside another Map.
+        only_toplevel_maps: Only consider Maps that are at the top.
+
+    Notes:
+        - This transformation modifies more nodes than it matches!
+    """
+
+    map_exit1 = dace_transformation.transformation.PatternNode(dace_nodes.MapExit)
+    access_node = dace_transformation.transformation.PatternNode(dace_nodes.AccessNode)
+    map_entry2 = dace_transformation.transformation.PatternNode(dace_nodes.MapEntry)
+
+    def __init__(
+        self,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(**kwargs)
+
+    @classmethod
+    def expressions(cls) -> Any:
+        """Get the match expression.
+
+        The transformation matches the exit node of the top Map that is connected to
+        an access node that again is connected to the entry node of the second Map.
+        An important note is, that the transformation operates not just on the
+        matched nodes, but more or less on anything that has an incoming connection
+        from the first Map or an outgoing connection to the second Map entry.
+        """
+        return [dace.sdfg.utils.node_path_graph(cls.map_exit1, cls.access_node, cls.map_entry2)]
+
+    def can_be_applied(
+        self,
+        graph: Union[dace.SDFGState, dace.SDFG],
+        expr_index: int,
+        sdfg: dace.SDFG,
+        permissive: bool = False,
+    ) -> bool:
+        """Tests if the matched Maps can be merged.
+
+        The two Maps are mergeable iff:
+        - The `can_be_fused()` of the base succeed, which checks some basic constraints.
+        - The decomposition exists and at least one of the intermediate sets
+            is not empty.
+        """
+        assert isinstance(self.map_exit1, dace_nodes.MapExit)
+        assert isinstance(self.map_entry2, dace_nodes.MapEntry)
+        map_entry_1: dace_nodes.MapEntry = graph.entry_node(self.map_exit1)
+        map_entry_2: dace_nodes.MapEntry = self.map_entry2
+
+        # This essentially test the structural properties of the two Maps.
+        if not self.can_be_fused(
+            map_entry_1=map_entry_1, map_entry_2=map_entry_2, graph=graph, sdfg=sdfg
+        ):
+            return False
+
+        # Two maps can be serially fused if the node decomposition exists and
+        #  at least one of the intermediate output sets is not empty. The state
+        #  of the pure outputs is irrelevant for serial map fusion.
+        output_partition = self.partition_first_outputs(
+            state=graph,
+            sdfg=sdfg,
+            map_exit_1=self.map_exit1,
+            map_entry_2=self.map_entry2,
+        )
+        if output_partition is None:
+            return False
+        _, exclusive_outputs, shared_outputs = output_partition
+        if not (exclusive_outputs or shared_outputs):
+            return False
+        return True
+
+    def apply(self, graph: Union[dace.SDFGState, dace.SDFG], sdfg: dace.SDFG) -> None:
+        """Performs the serial Map fusing.
+
+        The function first computes the map decomposition and then handles the
+        three sets. The pure outputs are handled by `relocate_nodes()` while
+        the two intermediate sets are handled by `handle_intermediate_set()`.
+
+        By assumption we do not have to rename anything.
+
+        Args:
+            graph: The SDFG state we are operating on.
+            sdfg: The SDFG we are operating on.
+        """
+        # NOTE: `self.map_*` actually stores the ID of the node.
+        #  once we start adding and removing nodes it seems that their ID changes.
+        #  Thus we have to save them here, this is a known behaviour in DaCe.
+        assert isinstance(graph, dace.SDFGState)
+        assert isinstance(self.map_exit1, dace_nodes.MapExit)
+        assert isinstance(self.map_entry2, dace_nodes.MapEntry)
+        assert self.map_parameter_compatible(self.map_exit1.map, self.map_entry2.map, graph, sdfg)
+
+        map_exit_1: dace_nodes.MapExit = self.map_exit1
+        map_entry_2: dace_nodes.MapEntry = self.map_entry2
+        map_exit_2: dace_nodes.MapExit = graph.exit_node(self.map_entry2)
+        map_entry_1: dace_nodes.MapEntry = graph.entry_node(self.map_exit1)
+
+        output_partition = self.partition_first_outputs(
+            state=graph,
+            sdfg=sdfg,
+            map_exit_1=map_exit_1,
+            map_entry_2=map_entry_2,
+        )
+        assert output_partition is not None  # Make MyPy happy.
+        pure_outputs, exclusive_outputs, shared_outputs = output_partition
+
+        if len(exclusive_outputs) != 0:
+            self.handle_intermediate_set(
+                intermediate_outputs=exclusive_outputs,
+                state=graph,
+                sdfg=sdfg,
+                map_exit_1=map_exit_1,
+                map_entry_2=map_entry_2,
+                map_exit_2=map_exit_2,
+                is_exclusive_set=True,
+            )
+        if len(shared_outputs) != 0:
+            self.handle_intermediate_set(
+                intermediate_outputs=shared_outputs,
+                state=graph,
+                sdfg=sdfg,
+                map_exit_1=map_exit_1,
+                map_entry_2=map_entry_2,
+                map_exit_2=map_exit_2,
+                is_exclusive_set=False,
+            )
+        assert pure_outputs == set(graph.out_edges(map_exit_1))
+        if len(pure_outputs) != 0:
+            self.relocate_nodes(
+                from_node=map_exit_1,
+                to_node=map_exit_2,
+                state=graph,
+                sdfg=sdfg,
+            )
+
+        # Above we have handled the input of the second map and moved them
+        #  to the first map, now we must move the output of the first map
+        #  to the second one, as this one is used.
+        self.relocate_nodes(
+            from_node=map_entry_2,
+            to_node=map_entry_1,
+            state=graph,
+            sdfg=sdfg,
+        )
+
+        for node_to_remove in [map_exit_1, map_entry_2]:
+            assert graph.degree(node_to_remove) == 0
+            graph.remove_node(node_to_remove)
+
+        # Now turn the second output node into the output node of the first Map.
+        map_exit_2.map = map_entry_1.map
+
+    @staticmethod
+    def handle_intermediate_set(
+        intermediate_outputs: set[dace_graph.MultiConnectorEdge[dace.Memlet]],
+        state: dace.SDFGState,
+        sdfg: dace.SDFG,
+        map_exit_1: dace_nodes.MapExit,
+        map_entry_2: dace_nodes.MapEntry,
+        map_exit_2: dace_nodes.MapExit,
+        is_exclusive_set: bool,
+    ) -> None:
+        """This function handles the intermediate sets.
+
+        The function is able to handle both the shared and exclusive intermediate
+        output set, see `partition_first_outputs()`. The main difference is that
+        in exclusive mode the intermediate nodes will be fully removed from
+        the SDFG. While in shared mode the intermediate node will be preserved.
+
+        Args:
+            intermediate_outputs: The set of outputs, that should be processed.
+            state: The state in which the map is processed.
+            sdfg: The SDFG that should be optimized.
+            map_exit_1: The exit of the first/top map.
+            map_entry_2: The entry of the second map.
+            map_exit_2: The exit of the second map.
+            is_exclusive_set: If `True` `intermediate_outputs` is the exclusive set.
+
+        Notes:
+            Before the transformation the `state` does not have to be valid and
+            after this function has run the state is (most likely) invalid.
+
+        Todo:
+            Rewrite using `MemletTree`.
+        """
+
+        # Essentially this function removes the AccessNode between the two maps.
+        #  However, we still need some temporary memory that we can use, which is
+        #  just much smaller, i.e. a scalar. But all Memlets inside the second map
+        #  assumes that the intermediate memory has the bigger shape.
+        #  To fix that we will create this replacement dict that will replace all
+        #  occurrences of the iteration variables of the second map with zero.
+        #  Note that this is still not enough as the dimensionality might be different.
+        memlet_repl: dict[str, int] = {str(param): 0 for param in map_entry_2.map.params}
+
+        # Now we will iterate over all intermediate edges and process them.
+        #  If not stated otherwise the comments assume that we run in exclusive mode.
+        for out_edge in intermediate_outputs:
+            # This is the intermediate node that, that we want to get rid of.
+            #  In shared mode we want to recreate it after the second map.
+            inter_node: dace_nodes.AccessNode = out_edge.dst
+            inter_name = inter_node.data
+            inter_desc = inter_node.desc(sdfg)
+            inter_shape = inter_desc.shape
+
+            # Now we will determine the shape of the new intermediate. This size of
+            #  this temporary is given by the Memlet that goes into the first map exit.
+            pre_exit_edges = list(
+                state.in_edges_by_connector(map_exit_1, "IN_" + out_edge.src_conn[4:])
+            )
+            if len(pre_exit_edges) != 1:
+                raise NotImplementedError()
+            pre_exit_edge = pre_exit_edges[0]
+            new_inter_shape_raw = dace_symbolic.overapproximate(pre_exit_edge.data.subset.size())
+
+            # Over approximation will leave us with some unneeded size one dimensions.
+            #  That are known to cause some troubles, so we will now remove them.
+            squeezed_dims: list[int] = []  # These are the dimensions we removed.
+            new_inter_shape: list[int] = []  # This is the final shape of the new intermediate.
+            for dim, (proposed_dim_size, full_dim_size) in enumerate(
+                zip(new_inter_shape_raw, inter_shape)
+            ):
+                # Order of checks is important!
+                if full_dim_size == 1:  # Must be kept!
+                    new_inter_shape.append(proposed_dim_size)
+                elif proposed_dim_size == 1:  # This dimension was reduced, so we can remove it.
+                    squeezed_dims.append(dim)
+                else:
+                    new_inter_shape.append(proposed_dim_size)
+
+            # This is the name of the new "intermediate" node that we will create.
+            #  It will only have the shape `new_inter_shape` which is basically its
+            #  output within one Map iteration.
+            #  NOTE: The insertion process might generate a new name.
+            new_inter_name: str = f"__s{sdfg.node_id(state)}_n{state.node_id(out_edge.src)}{out_edge.src_conn}_n{state.node_id(out_edge.dst)}{out_edge.dst_conn}"
+
+            # Now generate the intermediate data container.
+            if len(new_inter_shape) == 0:
+                assert pre_exit_edge.data.subset.num_elements() == 1
+                is_scalar = True
+                new_inter_name, new_inter_desc = sdfg.add_scalar(
+                    new_inter_name,
+                    dtype=inter_desc.dtype,
+                    transient=True,
+                    storage=dace_dtypes.StorageType.Register,
+                    find_new_name=True,
+                )
+
+            else:
+                assert (pre_exit_edge.data.subset.num_elements() > 1) or all(
+                    x == 1 for x in new_inter_shape
+                )
+                is_scalar = False
+                new_inter_name, new_inter_desc = sdfg.add_transient(
+                    new_inter_name,
+                    shape=new_inter_shape,
+                    dtype=inter_desc.dtype,
+                    find_new_name=True,
+                )
+            new_inter_node: dace_nodes.AccessNode = state.add_access(new_inter_name)
+
+            # New we will reroute the output Memlet, thus it will no longer pass
+            #  through the Map exit but through the newly created intermediate.
+            #  we will delete the previous edge later.
+            pre_exit_memlet: dace.Memlet = pre_exit_edge.data
+            new_pre_exit_memlet = copy.deepcopy(pre_exit_memlet)
+
+            # We might operate on a different array, but the check below, ensures
+            #  that we do not change the direction of the Memlet.
+            assert pre_exit_memlet.data == inter_name
+            new_pre_exit_memlet.data = new_inter_name
+
+            # Now we have to modify the subset of the Memlet.
+            #  Before the subset of the Memlet was dependent on the Map variables,
+            #  however, this is no longer the case, as we removed them. This change
+            #  has to be reflected in the Memlet.
+            #  NOTE: Assert above ensures that the below is correct.
+            new_pre_exit_memlet.replace(memlet_repl)
+            if is_scalar:
+                new_pre_exit_memlet.subset = "0"
+                new_pre_exit_memlet.other_subset = None
+            else:
+                new_pre_exit_memlet.subset.pop(squeezed_dims)
+
+            # Now we create the new edge between the producer and the new output
+            #  (the new intermediate node). We will remove the old edge further down.
+            new_pre_exit_edge = state.add_edge(
+                pre_exit_edge.src,
+                pre_exit_edge.src_conn,
+                new_inter_node,
+                None,
+                new_pre_exit_memlet,
+            )
+
+            # We just have handled the last Memlet, but we must actually handle the
+            #  whole producer side, i.e. the scope of the top Map.
+            for producer_tree in state.memlet_tree(new_pre_exit_edge).traverse_children():
+                producer_edge = producer_tree.edge
+
+                # Ensure the correctness of the rerouting below.
+                # TODO(phimuell): Improve the code below to remove the check.
+                assert producer_edge.data.data == inter_name
+
+                # Will not change the direction, because of test above!
+                producer_edge.data.data = new_inter_name
+                producer_edge.data.replace(memlet_repl)
+                if is_scalar:
+                    producer_edge.data.dst_subset = "0"
+                elif producer_edge.data.dst_subset is not None:
+                    producer_edge.data.dst_subset.pop(squeezed_dims)
+
+            # Now after we have handled the input of the new intermediate node,
+            #  we must handle its output. For this we have to "inject" the newly
+            #  created intermediate into the second map. We do this by finding
+            #  the input connectors on the map entry, such that we know where we
+            #  have to reroute inside the Map.
+            # NOTE: Assumes that map (if connected is the direct neighbour).
+            conn_names: set[str] = set()
+            for inter_node_out_edge in state.out_edges(inter_node):
+                if inter_node_out_edge.dst == map_entry_2:
+                    assert inter_node_out_edge.dst_conn.startswith("IN_")
+                    conn_names.add(inter_node_out_edge.dst_conn)
+                else:
+                    # If we found another target than the second map entry from the
+                    #  intermediate node it means that the node _must_ survive,
+                    #  i.e. we are not in exclusive mode.
+                    assert not is_exclusive_set
+
+            # Now we will reroute the connections inside the second map, i.e.
+            #  instead of consuming the old intermediate node, they will now
+            #  consume the new intermediate node.
+            for in_conn_name in conn_names:
+                out_conn_name = "OUT_" + in_conn_name[3:]
+
+                for inner_edge in state.out_edges_by_connector(map_entry_2, out_conn_name):
+                    assert inner_edge.data.data == inter_name  # DIRECTION!!
+
+                    # The create the first Memlet to transmit information, within
+                    #  the second map, we do this again by copying and modifying
+                    #  the original Memlet.
+                    # NOTE: Test above is important to ensure the direction of the
+                    #       Memlet and the correctness of the code below.
+                    new_inner_memlet = copy.deepcopy(inner_edge.data)
+                    new_inner_memlet.replace(memlet_repl)
+                    new_inner_memlet.data = new_inter_name  # Because of the assert above, this will not change the direction.
+
+                    # Now remove the old edge, that started the second map entry.
+                    #  Also add the new edge that started at the new intermediate.
+                    state.remove_edge(inner_edge)
+                    new_inner_edge = state.add_edge(
+                        new_inter_node,
+                        None,
+                        inner_edge.dst,
+                        inner_edge.dst_conn,
+                        new_inner_memlet,
+                    )
+
+                    # Now we do subset modification to ensure that nothing failed.
+                    if is_scalar:
+                        new_inner_memlet.src_subset = "0"
+                    elif new_inner_memlet.src_subset is not None:
+                        new_inner_memlet.src_subset.pop(squeezed_dims)
+
+                    # Now clean the Memlets of that tree to use the new intermediate node.
+                    for consumer_tree in state.memlet_tree(new_inner_edge).traverse_children():
+                        consumer_edge = consumer_tree.edge
+                        assert consumer_edge.data.data == inter_name
+                        consumer_edge.data.data = new_inter_name
+                        consumer_edge.data.replace(memlet_repl)
+                        if is_scalar:
+                            consumer_edge.data.src_subset = "0"
+                        elif consumer_edge.data.subset is not None:
+                            consumer_edge.data.subset.pop(squeezed_dims)
+
+                # The edge that leaves the second map entry was already deleted.
+                #  We will now delete the edges that brought the data.
+                for edge in state.in_edges_by_connector(map_entry_2, in_conn_name):
+                    assert edge.src == inter_node
+                    state.remove_edge(edge)
+                map_entry_2.remove_in_connector(in_conn_name)
+                map_entry_2.remove_out_connector(out_conn_name)
+
+            if is_exclusive_set:
+                # In exclusive mode the old intermediate node is no longer needed.
+                assert state.degree(inter_node) == 1
+                state.remove_edge_and_connectors(out_edge)
+                state.remove_node(inter_node)
+
+                state.remove_edge(pre_exit_edge)
+                map_exit_1.remove_in_connector(pre_exit_edge.dst_conn)
+                map_exit_1.remove_out_connector(out_edge.src_conn)
+                del sdfg.arrays[inter_name]
+
+            else:
+                # This is the shared mode, so we have to recreate the intermediate
+                #  node, but this time it is at the exit of the second map.
+                state.remove_edge(pre_exit_edge)
+                map_exit_1.remove_in_connector(pre_exit_edge.dst_conn)
+
+                # This is the Memlet that goes from the map internal intermediate
+                #  temporary node to the Map output. This will essentially restore
+                #  or preserve the output for the intermediate node. It is important
+                #  that we use the data that `preExitEdge` was used.
+                new_exit_memlet = copy.deepcopy(pre_exit_edge.data)
+                assert new_exit_memlet.data == inter_name
+                new_exit_memlet.subset = pre_exit_edge.data.dst_subset
+                new_exit_memlet.other_subset = (
+                    "0" if is_scalar else dace_subsets.Range.from_array(inter_desc)
+                )
+
+                new_pre_exit_conn = map_exit_2.next_connector()
+                state.add_edge(
+                    new_inter_node,
+                    None,
+                    map_exit_2,
+                    "IN_" + new_pre_exit_conn,
+                    new_exit_memlet,
+                )
+                state.add_edge(
+                    map_exit_2,
+                    "OUT_" + new_pre_exit_conn,
+                    inter_node,
+                    out_edge.dst_conn,
+                    copy.deepcopy(out_edge.data),
+                )
+                map_exit_2.add_in_connector("IN_" + new_pre_exit_conn)
+                map_exit_2.add_out_connector("OUT_" + new_pre_exit_conn)
+
+                map_exit_1.remove_out_connector(out_edge.src_conn)
+                state.remove_edge(out_edge)
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/util.py b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/util.py
new file mode 100644
index 00000000..29bae7bb
--- /dev/null
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/transformations/util.py
@@ -0,0 +1,160 @@
+# 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
+
+"""Common functionality for the transformations/optimization pipeline."""
+
+from typing import Iterable, Union
+
+import dace
+from dace.sdfg import graph as dace_graph, nodes as dace_nodes
+
+
+def is_nested_sdfg(
+    sdfg: Union[dace.SDFG, dace.SDFGState, dace_nodes.NestedSDFG],
+) -> bool:
+    """Tests if `sdfg` is a NestedSDFG."""
+    if isinstance(sdfg, dace.SDFGState):
+        sdfg = sdfg.parent
+    if isinstance(sdfg, dace_nodes.NestedSDFG):
+        return True
+    elif isinstance(sdfg, dace.SDFG):
+        return sdfg.parent_nsdfg_node is not None
+    raise TypeError(f"Does not know how to handle '{type(sdfg).__name__}'.")
+
+
+def all_nodes_between(
+    graph: dace.SDFG | dace.SDFGState,
+    begin: dace_nodes.Node,
+    end: dace_nodes.Node,
+    reverse: bool = False,
+) -> set[dace_nodes.Node] | None:
+    """Find all nodes that are reachable from `begin` but bound by `end`.
+
+    Essentially the function starts a DFS at `begin`. If an edge is found that lead
+    to `end`, this edge is ignored. It will thus found any node that is reachable
+    from `begin` by a path that does not involve `end`. The returned set will
+    never contain `end` nor `begin`. In case `end` is never found the function
+    will return `None`.
+
+    If `reverse` is set to `True` the function will start exploring at `end` and
+    follows the outgoing edges, i.e. the meaning of `end` and `begin` are swapped.
+
+    Args:
+        graph: The graph to operate on.
+        begin: The start of the DFS.
+        end: The terminator node of the DFS.
+        reverse: Perform a backward DFS.
+
+    Notes:
+        - The returned set will also contain the nodes of path that starts at
+            `begin` and ends at a node that is not `end`.
+    """
+
+    def next_nodes(node: dace_nodes.Node) -> Iterable[dace_nodes.Node]:
+        return (
+            (edge.src for edge in graph.in_edges(node))
+            if reverse
+            else (edge.dst for edge in graph.out_edges(node))
+        )
+
+    if reverse:
+        begin, end = end, begin
+
+    to_visit: list[dace_nodes.Node] = [begin]
+    seen: set[dace_nodes.Node] = set()
+
+    while len(to_visit) > 0:
+        node: dace_nodes.Node = to_visit.pop()
+        if node != end and node not in seen:
+            to_visit.extend(next_nodes(node))
+        seen.add(node)
+
+    # If `end` was not found we have to return `None`  to indicate this.
+    if end not in seen:
+        return None
+
+    # `begin` and `end` are not included in the output set.
+    return seen - {begin, end}
+
+
+def find_downstream_consumers(
+    state: dace.SDFGState,
+    begin: dace_nodes.Node | dace_graph.MultiConnectorEdge[dace.Memlet],
+    only_tasklets: bool = False,
+    reverse: bool = False,
+) -> set[tuple[dace_nodes.Node, dace_graph.MultiConnectorEdge[dace.Memlet]]]:
+    """Find all downstream connectors of `begin`.
+
+    A consumer, in for this function, is any node that is neither an entry nor
+    an exit node. The function returns a set of pairs, the first element is the
+    node that acts as consumer and the second is the edge that leads to it.
+    By setting `only_tasklets` the nodes the function finds are only Tasklets.
+
+    To find this set the function starts a search at `begin`, however, it is also
+    possible to pass an edge as `begin`.
+    If `reverse` is `True` the function essentially finds the producers that are
+    upstream.
+
+    Args:
+        state: The state in which to look for the consumers.
+        begin: The initial node that from which the search starts.
+        only_tasklets: Return only Tasklets.
+        reverse: Follow the reverse direction.
+    """
+    if isinstance(begin, dace_graph.MultiConnectorEdge):
+        to_visit: list[dace_graph.MultiConnectorEdge[dace.Memlet]] = [begin]
+    else:
+        to_visit = state.in_edges(begin) if reverse else state.out_edges(begin)
+
+    seen: set[dace_graph.MultiConnectorEdge[dace.Memlet]] = set()
+    found: set[tuple[dace_nodes.Node, dace_graph.MultiConnectorEdge[dace.Memlet]]] = set()
+
+    while len(to_visit) > 0:
+        curr_edge: dace_graph.MultiConnectorEdge[dace.Memlet] = to_visit.pop()
+        next_node: dace_nodes.Node = curr_edge.src if reverse else curr_edge.dst
+
+        if curr_edge in seen:
+            continue
+        seen.add(curr_edge)
+
+        if isinstance(next_node, (dace_nodes.MapEntry, dace_nodes.MapExit)):
+            if not reverse:
+                # In forward mode a Map entry could also mean the definition of a
+                #  dynamic map range.
+                if isinstance(next_node, dace_nodes.MapEntry) and (
+                    not curr_edge.dst_conn.startswith("IN_")
+                ):
+                    if not only_tasklets:
+                        found.add((next_node, curr_edge))
+                    continue
+                target_conn = curr_edge.dst_conn[3:]
+                new_edges = state.out_edges_by_connector(curr_edge.dst, "OUT_" + target_conn)
+            else:
+                target_conn = curr_edge.src_conn[4:]
+                new_edges = state.in_edges_by_connector(curr_edge.src, "IN_" + target_conn)
+            to_visit.extend(new_edges)
+
+        elif isinstance(next_node, dace_nodes.Tasklet) or not only_tasklets:
+            # We have found a consumer.
+            found.add((next_node, curr_edge))
+
+    return found
+
+
+def find_upstream_producers(
+    state: dace.SDFGState,
+    begin: dace_nodes.Node | dace_graph.MultiConnectorEdge[dace.Memlet],
+    only_tasklets: bool = False,
+) -> set[tuple[dace_nodes.Node, dace_graph.MultiConnectorEdge[dace.Memlet]]]:
+    """Same as `find_downstream_consumers()` but with `reverse` set to `True`."""
+    return find_downstream_consumers(
+        state=state,
+        begin=begin,
+        only_tasklets=only_tasklets,
+        reverse=True,
+    )


Test patch:
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/__init__.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/__init__.py
new file mode 100644
index 00000000..abf4c3e2
--- /dev/null
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/__init__.py
@@ -0,0 +1,8 @@
+# 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
+
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/conftest.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/conftest.py
new file mode 100644
index 00000000..5d5fb992
--- /dev/null
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/conftest.py
@@ -0,0 +1,37 @@
+# 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
+
+from typing import Any, Optional, Sequence, Union, overload, Literal, Generator
+
+import pytest
+import dace
+import copy
+import numpy as np
+from dace.sdfg import nodes as dace_nodes
+from dace.transformation import dataflow as dace_dataflow
+
+from gt4py.next import common as gtx_common
+from gt4py.next.program_processors.runners.dace_fieldview import (
+    transformations as gtx_transformations,
+)
+
+
+@pytest.fixture()
+def set_dace_settings() -> Generator[None, None, None]:
+    """Sets the common DaCe settings for the tests.
+
+    The function will modify the following settings:
+    - `optimizer.match_exception` exceptions during the pattern matching stage,
+        especially inside `can_be_applied()` are not ignored.
+    - `compiler.allow_view_arguments` allow that NumPy views can be passed to
+        `CompiledSDFG` objects as arguments.
+    """
+    with dace.config.temporary_config():
+        dace.Config.set("optimizer", "match_exception", value=False)
+        dace.Config.set("compiler", "allow_view_arguments", value=True)
+        yield
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_gpu_utils.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_gpu_utils.py
new file mode 100644
index 00000000..d51c2429
--- /dev/null
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_gpu_utils.py
@@ -0,0 +1,150 @@
+# 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
+
+from typing import Callable
+import dace
+import copy
+import numpy as np
+import pytest
+
+from dace.sdfg import nodes as dace_nodes
+
+from gt4py.next.program_processors.runners.dace_fieldview.transformations import (
+    gpu_utils as gtx_dace_fieldview_gpu_utils,
+)
+from . import util
+
+pytestmark = pytest.mark.usefixtures("set_dace_settings")
+
+
+def _get_trivial_gpu_promotable(
+    tasklet_code: str,
+) -> tuple[dace.SDFG, dace_nodes.MapEntry, dace_nodes.MapEntry]:
+    """Returns an SDFG that is suitable to test the `TrivialGPUMapPromoter` promoter.
+
+    The first map is a trivial map (`Map[__trival_gpu_it=0]`) containing a Tasklet,
+    that does not have an output, but writes a scalar value into `tmp` (output
+    connector `__out`), the body of this Tasklet can be controlled through the
+    `tasklet_code` argument.
+    The second map (`Map[__i0=0:N]`) contains a Tasklet that computes the sum of its
+    two inputs, the first input is the scalar value inside `tmp` and the second one
+    is `a[__i0]`, the result is stored in `b[__i0]`.
+
+    Returns:
+        A tuple, the first element is the SDFG, the second element is the map entry
+        of the trivial map and the last element is the map entry of the second map.
+
+    Args:
+        tasklet_code: The body of the Tasklet inside the trivial map.
+    """
+    sdfg = dace.SDFG(util.unique_name("gpu_promotable_sdfg"))
+    state = sdfg.add_state("state", is_start_block=True)
+    sdfg.add_symbol("N", dace.int32)
+
+    storage_array = dace.dtypes.StorageType.GPU_Global
+    storage_scalar = dace.dtypes.StorageType.Register
+    schedule = dace.dtypes.ScheduleType.GPU_Device
+
+    sdfg.add_scalar("tmp", dace.float64, transient=True)
+    sdfg.add_array("a", shape=("N",), dtype=dace.float64, transient=False, storage=storage_array)
+    sdfg.add_array("b", shape=("N",), dtype=dace.float64, transient=False, storage=storage_array)
+    a, b, tmp = (state.add_access(name) for name in ["a", "b", "tmp"])
+
+    _, trivial_map_entry, _ = state.add_mapped_tasklet(
+        "trivail_top_tasklet",
+        map_ranges={"__trivial_gpu_it": "0"},
+        inputs={},
+        code=tasklet_code,
+        outputs={"__out": dace.Memlet("tmp[0]")},
+        output_nodes={"tmp": tmp},
+        external_edges=True,
+        schedule=schedule,
+    )
+    _, second_map_entry, _ = state.add_mapped_tasklet(
+        "non_trivial_tasklet",
+        map_ranges={"__i0": "0:N"},
+        inputs={
+            "__in0": dace.Memlet("a[__i0]"),
+            "__in1": dace.Memlet("tmp[0]"),
+        },
+        code="__out = __in0 + __in1",
+        outputs={"__out": dace.Memlet("b[__i0]")},
+        input_nodes={"a": a, "tmp": tmp},
+        output_nodes={"b": b},
+        external_edges=True,
+        schedule=schedule,
+    )
+    return sdfg, trivial_map_entry, second_map_entry
+
+
+def test_trivial_gpu_map_promoter():
+    """Tests if the GPU map promoter works.
+
+    By using a body such as `__out = 3.0`, the transformation will apply.
+    """
+    sdfg, trivial_map_entry, second_map_entry = _get_trivial_gpu_promotable("__out = 3.0")
+    org_second_map_params = list(second_map_entry.map.params)
+    org_second_map_ranges = copy.deepcopy(second_map_entry.map.range)
+
+    nb_runs = sdfg.apply_transformations_once_everywhere(
+        gtx_dace_fieldview_gpu_utils.TrivialGPUMapPromoter(),
+        validate=True,
+        validate_all=True,
+    )
+    assert (
+        nb_runs == 1
+    ), f"Expected that 'TrivialGPUMapPromoter' applies once but it applied {nb_runs}."
+    trivial_map_params = trivial_map_entry.map.params
+    trivial_map_ranges = trivial_map_ranges.map.range
+    second_map_params = second_map_entry.map.params
+    second_map_ranges = second_map_entry.map.range
+
+    assert (
+        second_map_params == org_second_map_params
+    ), "The transformation modified the parameter of the second map."
+    assert all(
+        org_rng == rng for org_rng, rng in zip(org_second_map_ranges, second_map_ranges)
+    ), "The transformation modified the range of the second map."
+    assert all(
+        t_rng == s_rng for t_rng, s_rng in zip(trivial_map_ranges, second_map_ranges, strict=True)
+    ), "Expected that the ranges are the same; trivial '{trivial_map_ranges}'; second '{second_map_ranges}'."
+    assert (
+        trivial_map_params == second_map_params
+    ), f"Expected the trivial map to have parameters '{second_map_params}', but it had '{trivial_map_params}'."
+    assert sdfg.is_valid()
+
+
+def test_trivial_gpu_map_promoter():
+    """Test if the GPU promoter does not fuse a special trivial map.
+
+    By using a body such as `__out = __trivial_gpu_it` inside the
+    Tasklet's body, the map parameter is now used, and thus can not be fused.
+    """
+    sdfg, trivial_map_entry, second_map_entry = _get_trivial_gpu_promotable(
+        "__out = __trivial_gpu_it"
+    )
+    org_trivial_map_params = list(trivial_map_entry.map.params)
+    org_second_map_params = list(second_map_entry.map.params)
+
+    nb_runs = sdfg.apply_transformations_once_everywhere(
+        gtx_dace_fieldview_gpu_utils.TrivialGPUMapPromoter(),
+        validate=True,
+        validate_all=True,
+    )
+    assert (
+        nb_runs == 0
+    ), f"Expected that 'TrivialGPUMapPromoter' does not apply but it applied {nb_runs}."
+    trivial_map_params = trivial_map_entry.map.params
+    second_map_params = second_map_entry.map.params
+    assert (
+        trivial_map_params == org_trivial_map_params
+    ), f"Expected the trivial map to have parameters '{org_trivial_map_params}', but it had '{trivial_map_params}'."
+    assert (
+        second_map_params == org_second_map_params
+    ), f"Expected the trivial map to have parameters '{org_trivial_map_params}', but it had '{trivial_map_params}'."
+    assert sdfg.is_valid()
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_loop_blocking.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_loop_blocking.py
new file mode 100644
index 00000000..b4fde413
--- /dev/null
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_loop_blocking.py
@@ -0,0 +1,305 @@
+# 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
+
+from typing import Callable
+import dace
+import copy
+import numpy as np
+import pytest
+
+from dace.sdfg import nodes as dace_nodes, propagation as dace_propagation
+
+from gt4py.next.program_processors.runners.dace_fieldview import (
+    transformations as gtx_transformations,
+)
+from . import util
+
+pytestmark = pytest.mark.usefixtures("set_dace_settings")
+
+
+def _get_simple_sdfg() -> tuple[dace.SDFG, Callable[[np.ndarray, np.ndarray], np.ndarray]]:
+    """Creates a simple SDFG.
+
+    The k blocking transformation can be applied to the SDFG, however no node
+    can be taken out. This is because how it is constructed. However, applying
+    some simplistic transformations this can be done.
+    """
+    sdfg = dace.SDFG(util.unique_name("simple_block_sdfg"))
+    state = sdfg.add_state("state", is_start_block=True)
+    sdfg.add_symbol("N", dace.int32)
+    sdfg.add_symbol("M", dace.int32)
+    _, a = sdfg.add_array("a", ("N", "M"), dace.float64, transient=False)
+    _, b = sdfg.add_array("b", ("N",), dace.float64, transient=False)
+    _, c = sdfg.add_array("c", ("N", "M"), dace.float64, transient=False)
+    state.add_mapped_tasklet(
+        name="comp",
+        map_ranges=dict(i=f"0:N", j=f"0:M"),
+        inputs=dict(__in0=dace.Memlet("a[i, j]"), __in1=dace.Memlet("b[i]")),
+        outputs=dict(__out=dace.Memlet("c[i, j]")),
+        code="__out = __in0 + __in1",
+        external_edges=True,
+    )
+    return sdfg, lambda a, b: a + b.reshape((-1, 1))
+
+
+def _get_chained_sdfg() -> tuple[dace.SDFG, Callable[[np.ndarray, np.ndarray], np.ndarray]]:
+    """Generates an SDFG that has chained Tasklets that are independent.
+
+    The bottom Tasklet is the only dependent Tasklet.
+    """
+    sdfg = dace.SDFG(util.unique_name("chained_block_sdfg"))
+    state = sdfg.add_state("state", is_start_block=True)
+    sdfg.add_symbol("N", dace.int32)
+    sdfg.add_symbol("M", dace.int32)
+    sdfg.add_array("a", ("N", "M"), dace.float64, transient=False)
+    sdfg.add_array("b", ("N",), dace.float64, transient=False)
+    sdfg.add_array("c", ("N", "M"), dace.float64, transient=False)
+    sdfg.add_scalar("tmp1", dtype=dace.float64, transient=True)
+    sdfg.add_scalar("tmp2", dtype=dace.float64, transient=True)
+    a, b, c, tmp1, tmp2 = (state.add_access(name) for name in ["a", "b", "c", "tmp1", "tmp2"])
+
+    # First independent Tasklet.
+    task1 = state.add_tasklet(
+        "task1_indepenent",
+        inputs={
+            "__in0",  # <- `b[i]`
+        },
+        outputs={
+            "__out0",  # <- `tmp1`
+        },
+        code="__out0 = __in0 + 3.0",
+    )
+
+    # This is the second independent Tasklet.
+    task2 = state.add_tasklet(
+        "task2_indepenent",
+        inputs={
+            "__in0",  # <- `tmp1`
+            "__in1",  # <- `b[i]`
+        },
+        outputs={
+            "__out0",  # <- `tmp2`
+        },
+        code="__out0 = __in0 + __in1",
+    )
+
+    # This is the third Tasklet, which is dependent.
+    task3 = state.add_tasklet(
+        "task3_dependent",
+        inputs={
+            "__in0",  # <- `tmp2`
+            "__in1",  # <- `a[i, j]`
+        },
+        outputs={
+            "__out0",  # <- `c[i, j]`.
+        },
+        code="__out0 = __in0 + __in1",
+    )
+
+    # Now create the map
+    mentry, mexit = state.add_map(
+        "map",
+        ndrange={"i": "0:N", "j": "0:M"},
+    )
+
+    # Now assemble everything.
+    state.add_edge(mentry, "OUT_b", task1, "__in0", dace.Memlet("b[i]"))
+    state.add_edge(task1, "__out0", tmp1, None, dace.Memlet("tmp1[0]"))
+
+    state.add_edge(tmp1, None, task2, "__in0", dace.Memlet("tmp1[0]"))
+    state.add_edge(mentry, "OUT_b", task2, "__in1", dace.Memlet("b[i]"))
+    state.add_edge(task2, "__out0", tmp2, None, dace.Memlet("tmp2[0]"))
+
+    state.add_edge(tmp2, None, task3, "__in0", dace.Memlet("tmp2[0]"))
+    state.add_edge(mentry, "OUT_a", task3, "__in1", dace.Memlet("a[i, j]"))
+    state.add_edge(task3, "__out0", mexit, "IN_c", dace.Memlet("c[i, j]"))
+
+    state.add_edge(a, None, mentry, "IN_a", sdfg.make_array_memlet("a"))
+    state.add_edge(b, None, mentry, "IN_b", sdfg.make_array_memlet("b"))
+    state.add_edge(mexit, "OUT_c", c, None, sdfg.make_array_memlet("c"))
+    for name in ["a", "b"]:
+        mentry.add_in_connector("IN_" + name)
+        mentry.add_out_connector("OUT_" + name)
+    mexit.add_in_connector("IN_c")
+    mexit.add_out_connector("OUT_c")
+
+    dace_propagation.propagate_states(sdfg)
+    sdfg.validate()
+
+    return sdfg, lambda a, b: (a + (2 * b.reshape((-1, 1)) + 3))
+
+
+def test_only_dependent():
+    """Just applying the transformation to the SDFG.
+
+    Because all of nodes (which is only a Tasklet) inside the map scope are
+    "dependent", see the transformation for explanation of terminology, the
+    transformation will only add an inner map.
+    """
+    sdfg, reff = _get_simple_sdfg()
+
+    N, M = 100, 10
+    a = np.random.rand(N, M)
+    b = np.random.rand(N)
+    c = np.zeros_like(a)
+    ref = reff(a, b)
+
+    # Apply the transformation
+    sdfg.apply_transformations_repeated(
+        gtx_transformations.LoopBlocking(blocking_size=10, blocking_parameter="j"),
+        validate=True,
+        validate_all=True,
+    )
+
+    assert len(sdfg.states()) == 1
+    state = sdfg.states()[0]
+    source_nodes = state.source_nodes()
+    assert len(source_nodes) == 2
+    assert all(isinstance(x, dace_nodes.AccessNode) for x in source_nodes)
+    source_node = source_nodes[0]  # Unspecific which one it is, but it does not matter.
+    assert state.out_degree(source_node) == 1
+    outer_map: dace_nodes.MapEntry = next(iter(state.out_edges(source_node))).dst
+    assert isinstance(outer_map, dace_nodes.MapEntry)
+    assert state.in_degree(outer_map) == 2
+    assert state.out_degree(outer_map) == 2
+    assert len(outer_map.map.params) == 2
+    assert "j" not in outer_map.map.params
+    assert all(isinstance(x.dst, dace_nodes.MapEntry) for x in state.out_edges(outer_map))
+    inner_map: dace_nodes.MapEntry = next(iter(state.out_edges(outer_map))).dst
+    assert len(inner_map.map.params) == 1
+    assert inner_map.map.params[0] == "j"
+    assert inner_map.map.schedule == dace.dtypes.ScheduleType.Sequential
+
+    sdfg(a=a, b=b, c=c, N=N, M=M)
+    assert np.allclose(ref, c)
+
+
+def test_intermediate_access_node():
+    """Test the lifting out, version "AccessNode".
+
+    The Tasklet of the SDFG generated by `_get_simple_sdfg()` has to be inside the
+    inner most loop because one of its input Memlet depends on `j`. However,
+    one of its input, `b[i]` does not. Instead of connecting `b` directly with the
+    Tasklet, this test will store `b[i]` inside a temporary inside the Map.
+    This access node is independent of `j` and can thus be moved out of the inner
+    most scope.
+    """
+    sdfg, reff = _get_simple_sdfg()
+
+    N, M = 100, 10
+    a = np.random.rand(N, M)
+    b = np.random.rand(N)
+    c = np.zeros_like(a)
+    ref = reff(a, b)
+
+    # Now make a small modification is such that the transformation does something.
+    state = sdfg.states()[0]
+    sdfg.add_scalar("tmp", dace.float64, transient=True)
+
+    tmp = state.add_access("tmp")
+    edge = next(
+        e for e in state.edges() if isinstance(e.src, dace_nodes.MapEntry) and e.data.data == "b"
+    )
+    state.add_edge(edge.src, edge.src_conn, tmp, None, copy.deepcopy(edge.data))
+    state.add_edge(tmp, None, edge.dst, edge.dst_conn, dace.Memlet("tmp[0]"))
+    state.remove_edge(edge)
+
+    # Test if after the modification the SDFG still works
+    sdfg(a=a, b=b, c=c, N=N, M=M)
+    assert np.allclose(ref, c)
+
+    # Apply the transformation.
+    sdfg.apply_transformations_repeated(
+        gtx_transformations.LoopBlocking(blocking_size=10, blocking_parameter="j"),
+        validate=True,
+        validate_all=True,
+    )
+
+    # Inspect if the SDFG was modified correctly.
+    #  We only inspect `tmp` which now has to be between the two maps.
+    assert state.in_degree(tmp) == 1
+    assert state.out_degree(tmp) == 1
+    top_node = next(iter(state.in_edges(tmp))).src
+    bottom_node = next(iter(state.out_edges(tmp))).dst
+    assert isinstance(top_node, dace_nodes.MapEntry)
+    assert isinstance(bottom_node, dace_nodes.MapEntry)
+    assert bottom_node is not top_node
+
+    c[:] = 0
+    sdfg(a=a, b=b, c=c, N=N, M=M)
+    assert np.allclose(ref, c)
+
+
+def test_chained_access() -> None:
+    """Test if chained access works."""
+    sdfg, reff = _get_chained_sdfg()
+    state: dace.SDFGState = next(iter(sdfg.states()))
+
+    N, M = 100, 10
+    a = np.random.rand(N, M)
+    b = np.random.rand(N)
+    c = np.zeros_like(a)
+    ref = reff(a, b)
+
+    # Before the optimization
+    sdfg(a=a, b=b, c=c, M=M, N=N)
+    assert np.allclose(c, ref)
+    c[:] = 0
+
+    # Apply the transformation.
+    ret = sdfg.apply_transformations_repeated(
+        gtx_transformations.LoopBlocking(blocking_size=10, blocking_parameter="j"),
+        validate=True,
+        validate_all=True,
+    )
+    assert ret == 1, f"Expected that the transformation was applied 1 time, but it was {ret}."
+
+    # Now run the SDFG to see if it is still the same
+    sdfg(a=a, b=b, c=c, M=M, N=N)
+    assert np.allclose(c, ref)
+
+    # Now look for the outer map.
+    outer_map = None
+    for node in state.nodes():
+        if not isinstance(node, dace_nodes.MapEntry):
+            continue
+        if state.scope_dict()[node] is None:
+            assert (
+                outer_map is None
+            ), f"Found multiple outer maps, first '{outer_map}', second '{node}'."
+            outer_map = node
+    assert outer_map is not None, "Could not found the outer map."
+    assert len(outer_map.map.params) == 2
+
+    # Now inspect the SDFG if the transformation was applied correctly.
+    first_level_tasklets: list[dace_nodes.Tasklet] = []
+    inner_map: list[dace_nodes.MapEntry] = []
+
+    for edge in state.out_edges(outer_map):
+        node: dace_nodes.Node = edge.dst
+        if isinstance(node, dace_nodes.Tasklet):
+            first_level_tasklets.append(node)
+        elif isinstance(node, dace_nodes.MapEntry):
+            inner_map.append(node)
+        else:
+            assert False, f"Found unexpected node '{type(node).__name__}'."
+
+    # Test what we found
+    assert len(first_level_tasklets) == 2
+    assert len(set(first_level_tasklets)) == 2
+    assert len(inner_map) == 1
+    assert state.scope_dict()[inner_map[0]] is outer_map
+
+    # Now we look inside the inner map
+    #  There we expect to find one Tasklet.
+    inner_scope = state.scope_subgraph(next(iter(inner_map)), False, False)
+    assert inner_scope.number_of_nodes() == 1
+    inner_tasklet = next(iter(inner_scope.nodes()))
+
+    assert isinstance(inner_tasklet, dace_nodes.Tasklet)
+    assert inner_tasklet not in first_level_tasklets
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_map_fusion.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_map_fusion.py
new file mode 100644
index 00000000..54319b27
--- /dev/null
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_map_fusion.py
@@ -0,0 +1,458 @@
+# 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
+
+from typing import Any, Optional, Sequence, Union, Literal, overload
+
+import pytest
+import dace
+import copy
+import numpy as np
+from dace.sdfg import nodes as dace_nodes
+from dace.transformation import dataflow as dace_dataflow
+
+from gt4py.next import common as gtx_common
+from gt4py.next.program_processors.runners.dace_fieldview import (
+    transformations as gtx_transformations,
+)
+from . import util
+
+pytestmark = pytest.mark.usefixtures("set_dace_settings")
+
+
+def _make_serial_sdfg_1(
+    N: str | int,
+) -> dace.SDFG:
+    """Create the "serial_1_sdfg".
+
+    This is an SDFG with a single state containing two maps. It has the input
+    `a` and the output `b`, each two dimensional arrays, with size `0:N`.
+    The first map adds 1 to the input and writes it into `tmp`. The second map
+    adds another 3 to `tmp` and writes it back inside `b`.
+
+    Args:
+        N: The size of the arrays.
+    """
+    shape = (N, N)
+    sdfg = dace.SDFG(util.unique_name("serial_sdfg1"))
+    state = sdfg.add_state(is_start_block=True)
+
+    for name in ["a", "b", "tmp"]:
+        sdfg.add_array(
+            name=name,
+            shape=shape,
+            dtype=dace.float64,
+            transient=False,
+        )
+    sdfg.arrays["tmp"].transient = True
+    tmp = state.add_access("tmp")
+
+    state.add_mapped_tasklet(
+        name="first_computation",
+        map_ranges=[("__i0", f"0:{N}"), ("__i1", f"0:{N}")],
+        inputs={"__in0": dace.Memlet("a[__i0, __i1]")},
+        code="__out = __in0 + 1.0",
+        outputs={"__out": dace.Memlet("tmp[__i0, __i1]")},
+        output_nodes={"tmp": tmp},
+        external_edges=True,
+    )
+
+    state.add_mapped_tasklet(
+        name="second_computation",
+        map_ranges=[("__i0", f"0:{N}"), ("__i1", f"0:{N}")],
+        input_nodes={"tmp": tmp},
+        inputs={"__in0": dace.Memlet("tmp[__i0, __i1]")},
+        code="__out = __in0 + 3.0",
+        outputs={"__out": dace.Memlet("b[__i0, __i1]")},
+        external_edges=True,
+    )
+
+    return sdfg
+
+
+def _make_serial_sdfg_2(
+    N: str | int,
+) -> dace.SDFG:
+    """Create the "serial_2_sdfg".
+
+    The generated SDFG uses `a` and input and has two outputs `b := a + 4` and
+    `c := a - 4`. There is a top map with a single Single Tasklet, that has
+    two outputs, the first one computes `a + 1` and stores that in `tmp_1`.
+    The second output computes `a - 1` and stores it `tmp_2`.
+    Below the top map are two (parallel) map, one compute `b := tmp_1 + 3`, while
+    the other compute `c := tmp_2 - 3`. This means that there are two map fusions.
+    The main important thing is that, the second map fusion will involve a pure
+    fusion (because the processing order is indeterministic, one does not know
+    which one in advance).
+
+    Args:
+        N: The size of the arrays.
+    """
+    shape = (N, N)
+    sdfg = dace.SDFG(util.unique_name("serial_sdfg2"))
+    state = sdfg.add_state(is_start_block=True)
+
+    for name in ["a", "b", "c", "tmp_1", "tmp_2"]:
+        sdfg.add_array(
+            name=name,
+            shape=shape,
+            dtype=dace.float64,
+            transient=False,
+        )
+    sdfg.arrays["tmp_1"].transient = True
+    sdfg.arrays["tmp_2"].transient = True
+    tmp_1 = state.add_access("tmp_1")
+    tmp_2 = state.add_access("tmp_2")
+
+    state.add_mapped_tasklet(
+        name="first_computation",
+        map_ranges=[("__i0", f"0:{N}"), ("__i1", f"0:{N}")],
+        inputs={"__in0": dace.Memlet("a[__i0, __i1]")},
+        code="__out0 = __in0 + 1.0\n__out1 = __in0 - 1.0",
+        outputs={
+            "__out0": dace.Memlet("tmp_1[__i0, __i1]"),
+            "__out1": dace.Memlet("tmp_2[__i0, __i1]"),
+        },
+        output_nodes={
+            "tmp_1": tmp_1,
+            "tmp_2": tmp_2,
+        },
+        external_edges=True,
+    )
+
+    state.add_mapped_tasklet(
+        name="first_computation",
+        map_ranges=[("__i0", f"0:{N}"), ("__i1", f"0:{N}")],
+        input_nodes={"tmp_1": tmp_1},
+        inputs={"__in0": dace.Memlet("tmp_1[__i0, __i1]")},
+        code="__out = __in0 + 3.0",
+        outputs={"__out": dace.Memlet("b[__i0, __i1]")},
+        external_edges=True,
+    )
+    state.add_mapped_tasklet(
+        name="second_computation",
+        map_ranges=[("__i0", f"0:{N}"), ("__i1", f"0:{N}")],
+        input_nodes={"tmp_2": tmp_2},
+        inputs={"__in0": dace.Memlet("tmp_2[__i0, __i1]")},
+        code="__out = __in0 - 3.0",
+        outputs={"__out": dace.Memlet("c[__i0, __i1]")},
+        external_edges=True,
+    )
+
+    return sdfg
+
+
+def _make_serial_sdfg_3(
+    N_input: str | int,
+    N_output: str | int,
+) -> dace.SDFG:
+    """Creates a serial SDFG that has an indirect access Tasklet in the second map.
+
+    The SDFG has three inputs `a`, `b` and `idx`. The first two are 1 dimensional
+    arrays, and the second is am array containing integers.
+    The top map computes `a + b` and stores that in `tmp`.
+    The second map then uses the elements of `idx` to make indirect accesses into
+    `tmp`, which are stored inside `c`.
+
+    Args:
+        N_input: The length of `a` and `b`.
+        N_output: The length of `c` and `idx`.
+    """
+    input_shape = (N_input,)
+    output_shape = (N_output,)
+
+    sdfg = dace.SDFG(util.unique_name("serial_sdfg3"))
+    state = sdfg.add_state(is_start_block=True)
+
+    for name, shape in [
+        ("a", input_shape),
+        ("b", input_shape),
+        ("c", output_shape),
+        ("idx", output_shape),
+        ("tmp", input_shape),
+    ]:
+        sdfg.add_array(
+            name=name,
+            shape=shape,
+            dtype=dace.int32 if name == "idx" else dace.float64,
+            transient=False,
+        )
+    sdfg.arrays["tmp"].transient = True
+    tmp = state.add_access("tmp")
+
+    state.add_mapped_tasklet(
+        name="first_computation",
+        map_ranges=[("__i0", f"0:{N_input}")],
+        inputs={
+            "__in0": dace.Memlet("a[__i0]"),
+            "__in1": dace.Memlet("b[__i0]"),
+        },
+        code="__out = __in0 + __in1",
+        outputs={"__out": dace.Memlet("tmp[__i0]")},
+        output_nodes={"tmp": tmp},
+        external_edges=True,
+    )
+
+    state.add_mapped_tasklet(
+        name="indirect_access",
+        map_ranges=[("__i0", f"0:{N_output}")],
+        input_nodes={"tmp": tmp},
+        inputs={
+            "__index": dace.Memlet("idx[__i0]"),
+            "__array": dace.Memlet.simple("tmp", subset_str=f"0:{N_input}", num_accesses=1),
+        },
+        code="__out = __array[__index]",
+        outputs={"__out": dace.Memlet("c[__i0]")},
+        external_edges=True,
+    )
+
+    return sdfg
+
+
+def test_exclusive_itermediate():
+    """Tests if the exclusive intermediate branch works."""
+    N = 10
+    sdfg = _make_serial_sdfg_1(N)
+
+    # Now apply the optimizations.
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 2
+    sdfg.apply_transformations(
+        gtx_transformations.SerialMapFusion(),
+        validate=True,
+        validate_all=True,
+    )
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 1
+    assert "tmp" not in sdfg.arrays
+
+    # Test if the intermediate is a scalar
+    intermediate_nodes: list[dace_nodes.Node] = [
+        node
+        for node in util._count_nodes(sdfg, dace_nodes.AccessNode, True)
+        if node.data not in ["a", "b"]
+    ]
+    assert len(intermediate_nodes) == 1
+    assert all(isinstance(node.desc(sdfg), dace.data.Scalar) for node in intermediate_nodes)
+
+    a = np.random.rand(N, N)
+    b = np.empty_like(a)
+    ref = a + 4.0
+    sdfg(a=a, b=b)
+
+    assert np.allclose(b, ref)
+
+
+def test_shared_itermediate():
+    """Tests the shared intermediate path.
+
+    The function uses the `_make_serial_sdfg_1()` SDFG. However, it promotes `tmp`
+    to a global, and it thus became a shared intermediate, i.e. will survive.
+    """
+    N = 10
+    sdfg = _make_serial_sdfg_1(N)
+    sdfg.arrays["tmp"].transient = False
+
+    # Now apply the optimizations.
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 2
+    sdfg.apply_transformations(
+        gtx_transformations.SerialMapFusion(),
+        validate=True,
+        validate_all=True,
+    )
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 1
+    assert "tmp" in sdfg.arrays
+
+    # Test if the intermediate is a scalar
+    intermediate_nodes: list[dace_nodes.Node] = [
+        node
+        for node in util._count_nodes(sdfg, dace_nodes.AccessNode, True)
+        if node.data not in ["a", "b", "tmp"]
+    ]
+    assert len(intermediate_nodes) == 1
+    assert all(isinstance(node.desc(sdfg), dace.data.Scalar) for node in intermediate_nodes)
+
+    a = np.random.rand(N, N)
+    b = np.empty_like(a)
+    tmp = np.empty_like(a)
+
+    ref_b = a + 4.0
+    ref_tmp = a + 1.0
+    sdfg(a=a, b=b, tmp=tmp)
+
+    assert np.allclose(b, ref_b)
+    assert np.allclose(tmp, ref_tmp)
+
+
+def test_pure_output_node():
+    """Tests the path of a pure intermediate."""
+    N = 10
+    sdfg = _make_serial_sdfg_2(N)
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 3
+
+    # The first fusion will only bring it down to two maps.
+    sdfg.apply_transformations(
+        gtx_transformations.SerialMapFusion(),
+        validate=True,
+        validate_all=True,
+    )
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 2
+    sdfg.apply_transformations(
+        gtx_transformations.SerialMapFusion(),
+        validate=True,
+        validate_all=True,
+    )
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 1
+
+    a = np.random.rand(N, N)
+    b = np.empty_like(a)
+    c = np.empty_like(a)
+    ref_b = a + 4.0
+    ref_c = a - 4.0
+    sdfg(a=a, b=b, c=c)
+
+    assert np.allclose(b, ref_b)
+    assert np.allclose(c, ref_c)
+
+
+def test_array_intermediate():
+    """Tests the correct working if we have more than scalar intermediate.
+
+    The test used `_make_serial_sdfg_1()` to get an SDFG and then call `MapExpansion`.
+    Map fusion is then called only outer maps, thus the intermediate node, must
+    be an array.
+    """
+    N = 10
+    sdfg = _make_serial_sdfg_1(N)
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 2
+    sdfg.apply_transformations_repeated([dace_dataflow.MapExpansion])
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 4
+
+    # Now perform the fusion
+    sdfg.apply_transformations(
+        gtx_transformations.SerialMapFusion(only_toplevel_maps=True),
+        validate=True,
+        validate_all=True,
+    )
+    map_entries = util._count_nodes(sdfg, dace_nodes.MapEntry, return_nodes=True)
+
+    scope = next(iter(sdfg.states())).scope_dict()
+    assert len(map_entries) == 3
+    top_maps = [map_entry for map_entry in map_entries if scope[map_entry] is None]
+    assert len(top_maps) == 1
+    top_map = top_maps[0]
+    assert sum(scope[map_entry] is top_map for map_entry in map_entries) == 2
+
+    # Find the access node that is the new intermediate node.
+    inner_access_nodes: list[dace_nodes.AccessNode] = [
+        node
+        for node in util._count_nodes(sdfg, dace_nodes.AccessNode, True)
+        if scope[node] is not None
+    ]
+    assert len(inner_access_nodes) == 1
+    inner_access_node = inner_access_nodes[0]
+    inner_desc: dace.data.Data = inner_access_node.desc(sdfg)
+    assert inner_desc.shape == (N,)
+
+    a = np.random.rand(N, N)
+    b = np.empty_like(a)
+    ref_b = a + 4.0
+    sdfg(a=a, b=b)
+
+    assert np.allclose(ref_b, b)
+
+
+def test_interstate_transient():
+    """Tests if an interstate transient is handled properly.
+
+    This function uses the SDFG generated by `_make_serial_sdfg_2()`. It adds a second
+    state to SDFG in which `tmp_1` is read from and the result is written in `d` (new
+    variable). Thus `tmp_1` can not be removed.
+    """
+    N = 10
+    sdfg = _make_serial_sdfg_2(N)
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 3
+    assert sdfg.number_of_nodes() == 1
+
+    # Now add the new state and the new output.
+    sdfg.add_datadesc("d", copy.deepcopy(sdfg.arrays["b"]))
+    head_state = next(iter(sdfg.states()))
+    new_state = sdfg.add_state_after(head_state)
+
+    new_state.add_mapped_tasklet(
+        name="first_computation_second_state",
+        map_ranges=[("__i0", f"0:{N}"), ("__i1", f"0:{N}")],
+        inputs={"__in0": dace.Memlet("tmp_1[__i0, __i1]")},
+        code="__out = __in0 + 9.0",
+        outputs={"__out": dace.Memlet("d[__i0, __i1]")},
+        external_edges=True,
+    )
+
+    # Now apply the transformation
+    sdfg.apply_transformations_repeated(
+        gtx_transformations.SerialMapFusion(),
+        validate=True,
+        validate_all=True,
+    )
+    assert "tmp_1" in sdfg.arrays
+    assert "tmp_2" not in sdfg.arrays
+    assert sdfg.number_of_nodes() == 2
+    assert util._count_nodes(head_state, dace_nodes.MapEntry) == 1
+    assert util._count_nodes(new_state, dace_nodes.MapEntry) == 1
+
+    a = np.random.rand(N, N)
+    b = np.empty_like(a)
+    c = np.empty_like(a)
+    d = np.empty_like(a)
+    ref_b = a + 4.0
+    ref_c = a - 4.0
+    ref_d = a + 10.0
+
+    sdfg(a=a, b=b, c=c, d=d)
+    assert np.allclose(ref_b, b)
+    assert np.allclose(ref_c, c)
+    assert np.allclose(ref_d, d)
+
+
+def test_indirect_access():
+    """Tests if indirect accesses are handled.
+
+    Indirect accesses, a Tasklet dereferences the array, can not be fused, because
+    the array is accessed by the Tasklet.
+    """
+    N_input = 100
+    N_output = 1000
+    a = np.random.rand(N_input)
+    b = np.random.rand(N_input)
+    c = np.empty(N_output)
+    idx = np.random.randint(low=0, high=N_input, size=N_output, dtype=np.int32)
+    sdfg = _make_serial_sdfg_3(N_input=N_input, N_output=N_output)
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 2
+
+    def _ref(a, b, idx):
+        tmp = a + b
+        return tmp[idx]
+
+    ref = _ref(a, b, idx)
+
+    sdfg(a=a, b=b, idx=idx, c=c)
+    assert np.allclose(ref, c)
+
+    # Now "apply" the transformation
+    sdfg.apply_transformations_repeated(
+        gtx_transformations.SerialMapFusion(),
+        validate=True,
+        validate_all=True,
+    )
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 2
+
+    c[:] = -1.0
+    sdfg(a=a, b=b, idx=idx, c=c)
+    assert np.allclose(ref, c)
+
+
+def test_indirect_access_2():
+    # TODO(phimuell): Index should be computed and that map should be fusable.
+    pass
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_serial_map_promoter.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_serial_map_promoter.py
new file mode 100644
index 00000000..85846ef6
--- /dev/null
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/test_serial_map_promoter.py
@@ -0,0 +1,98 @@
+# 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
+
+from typing import Callable
+import dace
+import copy
+import numpy as np
+import pytest
+
+from dace.sdfg import nodes as dace_nodes
+
+from gt4py.next.program_processors.runners.dace_fieldview import (
+    transformations as gtx_transformations,
+)
+from . import util
+
+
+pytestmark = pytest.mark.usefixtures("set_dace_settings")
+
+
+def test_serial_map_promotion():
+    """Tests the serial Map promotion transformation."""
+    N = 10
+    shape_1d = (N,)
+    shape_2d = (N, N)
+    sdfg = dace.SDFG(util.unique_name("serial_promotable_sdfg"))
+    state = sdfg.add_state(is_start_block=True)
+
+    # 1D Arrays
+    for name in ["a", "tmp"]:
+        sdfg.add_array(
+            name=name,
+            shape=shape_1d,
+            dtype=dace.float64,
+            transient=False,
+        )
+    sdfg.arrays["tmp"].transient = True
+
+    # 2D Arrays
+    for name in ["b", "c"]:
+        sdfg.add_array(
+            name=name,
+            shape=shape_2d,
+            dtype=dace.float64,
+            transient=False,
+        )
+    tmp = state.add_access("tmp")
+
+    _, map_entry_1d, _ = state.add_mapped_tasklet(
+        name="one_d_map",
+        map_ranges=[("__i0", f"0:{N}")],
+        inputs={"__in0": dace.Memlet("a[__i0]")},
+        code="__out = __in0 + 1.0",
+        outputs={"__out": dace.Memlet("tmp[__i0]")},
+        output_nodes={"tmp": tmp},
+        external_edges=True,
+    )
+
+    _, map_entry_2d, _ = state.add_mapped_tasklet(
+        name="two_d_map",
+        map_ranges=[("__i0", f"0:{N}"), ("__i1", f"0:{N}")],
+        input_nodes={"tmp": tmp},
+        inputs={"__in0": dace.Memlet("tmp[__i0]"), "__in1": dace.Memlet("b[__i0, __i1]")},
+        code="__out = __in0 + __in1",
+        outputs={"__out": dace.Memlet("c[__i0, __i1]")},
+        external_edges=True,
+    )
+
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 2
+    assert len(map_entry_1d.map.params) == 1
+    assert len(map_entry_1d.map.range) == 1
+    assert len(map_entry_2d.map.params) == 2
+    assert len(map_entry_2d.map.range) == 2
+
+    # Now apply the promotion
+    sdfg.apply_transformations(
+        gtx_transformations.SerialMapPromoter(
+            promote_all=True,
+        ),
+        validate=True,
+        validate_all=True,
+    )
+
+    assert util._count_nodes(sdfg, dace_nodes.MapEntry) == 2
+    assert len(map_entry_1d.map.params) == 2
+    assert len(map_entry_1d.map.range) == 2
+    assert len(map_entry_2d.map.params) == 2
+    assert len(map_entry_2d.map.range) == 2
+    assert set(map_entry_1d.map.params) == set(map_entry_2d.map.params)
+    assert all(
+        rng_1d == rng_2d
+        for rng_1d, rng_2d in zip(map_entry_1d.map.range.ranges, map_entry_2d.map.range.ranges)
+    )
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/util.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/util.py
new file mode 100644
index 00000000..0caabcc7
--- /dev/null
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace/transformation_tests/util.py
@@ -0,0 +1,61 @@
+# 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
+
+from typing import Union, Literal, overload
+import uuid
+
+import dace
+from dace.sdfg import nodes as dace_nodes
+from dace.transformation import dataflow as dace_dataflow
+
+
+@overload
+def _count_nodes(
+    graph: Union[dace.SDFG, dace.SDFGState],
+    node_type: tuple[type, ...] | type,
+    return_nodes: Literal[False],
+) -> int: ...
+
+
+@overload
+def _count_nodes(
+    graph: Union[dace.SDFG, dace.SDFGState],
+    node_type: tuple[type, ...] | type,
+    return_nodes: Literal[True],
+) -> list[dace_nodes.Node]: ...
+
+
+def _count_nodes(
+    graph: Union[dace.SDFG, dace.SDFGState],
+    node_type: tuple[type, ...] | type,
+    return_nodes: bool = False,
+) -> Union[int, list[dace_nodes.Node]]:
+    """Counts the number of nodes in of a particular type in `graph`.
+
+    If `graph` is an SDFGState then only count the nodes inside this state,
+    but if `graph` is an SDFG count in all states.
+
+    Args:
+        graph: The graph to scan.
+        node_type: The type or sequence of types of nodes to look for.
+    """
+
+    states = graph.states() if isinstance(graph, dace.SDFG) else [graph]
+    found_nodes: list[dace_nodes.Node] = []
+    for state_nodes in states:
+        for node in state_nodes.nodes():
+            if isinstance(node, node_type):
+                found_nodes.append(node)
+    if return_nodes:
+        return found_nodes
+    return len(found_nodes)
+
+
+def unique_name(name: str) -> str:
+    """Adds a unique string to `name`."""
+    return f"{name}_{str(uuid.uuid1()).replace('-', '_')}"