Repository: GridTools/gt4py
Pull request #1702 (2024-10-22 20:15:06)

fext[next]: GTIR embedded backend (not active in tests)
with features from #1648


Patch:
diff --git a/docs/user/next/advanced/ToolchainWalkthrough.md b/docs/user/next/advanced/ToolchainWalkthrough.md
index b82dea1a..a5a63cb5 100644
--- a/docs/user/next/advanced/ToolchainWalkthrough.md
+++ b/docs/user/next/advanced/ToolchainWalkthrough.md
@@ -247,7 +247,7 @@ pprint.pprint(jit_args)
 ```
 
 ```python
-gtx.program_processors.runners.roundtrip.executor(pitir)(*jit_args.args, **jit_args.kwargs)
+gtx.program_processors.runners.roundtrip.Roundtrip()(pitir)(*jit_args.args, **jit_args.kwargs)
 ```
 
 ```python
@@ -290,7 +290,7 @@ assert pitir2 == pitir
 #### Pass The result to the compile workflow and execute
 
 ```python
-example_compiled = gtx.program_processors.runners.roundtrip.executor(pitir2)
+example_compiled = gtx.program_processors.runners.roundtrip.Roundtrip()(pitir2)
 ```
 
 ```python
diff --git a/src/gt4py/cartesian/frontend/gtscript_frontend.py b/src/gt4py/cartesian/frontend/gtscript_frontend.py
index e2aa98f3..ade05921 100644
--- a/src/gt4py/cartesian/frontend/gtscript_frontend.py
+++ b/src/gt4py/cartesian/frontend/gtscript_frontend.py
@@ -459,7 +459,8 @@ class CallInliner(ast.NodeTransformer):
                     call_args[name] = ast.Constant(value=arg_infos[name])
         except Exception as ex:
             raise GTScriptSyntaxError(
-                message="Invalid call signature", loc=nodes.Location.from_ast_node(node)
+                message=f"Invalid call signature when calling {call_name}",
+                loc=nodes.Location.from_ast_node(node),
             ) from ex
 
         # Rename local names in subroutine to avoid conflicts with caller context names
diff --git a/src/gt4py/next/ffront/decorator.py b/src/gt4py/next/ffront/decorator.py
index 52fe8d81..dc2421e1 100644
--- a/src/gt4py/next/ffront/decorator.py
+++ b/src/gt4py/next/ffront/decorator.py
@@ -34,6 +34,8 @@ from gt4py.next.common import Connectivity, Dimension, GridType
 from gt4py.next.embedded import operators as embedded_operators
 from gt4py.next.ffront import (
     field_operator_ast as foast,
+    foast_to_gtir,
+    foast_to_itir,
     past_process_args,
     signature,
     stages as ffront_stages,
@@ -560,10 +562,15 @@ class FieldOperator(GTCallable, Generic[OperatorNodeT]):
             self, definition_stage=dataclasses.replace(self.definition_stage, grid_type=grid_type)
         )
 
+    # TODO(tehrengruber): We can not use transforms from `self.backend` since this can be
+    #  a different backend than the one of the program that calls this field operator. Just use
+    #  the hard-coded lowering until this is cleaned up.
     def __gt_itir__(self) -> itir.FunctionDefinition:
-        return self._frontend_transforms.foast_to_itir(
-            toolchain.CompilableProgram(self.foast_stage, arguments.CompileTimeArgs.empty())
-        )
+        return foast_to_itir.foast_to_itir(self.foast_stage)
+
+    # FIXME[#1582](tehrengruber): remove after refactoring to GTIR
+    def __gt_gtir__(self) -> itir.FunctionDefinition:
+        return foast_to_gtir.foast_to_gtir(self.foast_stage)
 
     def __gt_closure_vars__(self) -> dict[str, Any]:
         return self.foast_stage.closure_vars
diff --git a/src/gt4py/next/ffront/foast_to_gtir.py b/src/gt4py/next/ffront/foast_to_gtir.py
index 9cb0ce05..0d0c3868 100644
--- a/src/gt4py/next/ffront/foast_to_gtir.py
+++ b/src/gt4py/next/ffront/foast_to_gtir.py
@@ -374,9 +374,7 @@ class FieldOperatorLowering(eve.PreserveLocationVisitor, eve.NodeTranslator):
 
     def _visit_broadcast(self, node: foast.Call, **kwargs: Any) -> itir.FunCall:
         expr = self.visit(node.args[0], **kwargs)
-        if isinstance(node.args[0].type, ts.ScalarType):
-            return im.as_fieldop(im.ref("deref"))(expr)
-        return expr
+        return im.as_fieldop(im.ref("deref"))(expr)
 
     def _visit_math_built_in(self, node: foast.Call, **kwargs: Any) -> itir.FunCall:
         return self._map(self.visit(node.func, **kwargs), *node.args)
diff --git a/src/gt4py/next/ffront/foast_to_past.py b/src/gt4py/next/ffront/foast_to_past.py
index 0844f632..312ac686 100644
--- a/src/gt4py/next/ffront/foast_to_past.py
+++ b/src/gt4py/next/ffront/foast_to_past.py
@@ -45,6 +45,11 @@ class ItirShim:
     def __gt_itir__(self) -> itir.Expr:
         return self.foast_to_itir(self.definition)
 
+    # FIXME[#1582](tehrengruber): remove after refactoring to GTIR
+    def __gt_gtir__(self) -> itir.Expr:
+        # backend should have self.foast_to_itir set to foast_to_gtir
+        return self.foast_to_itir(self.definition)
+
 
 @dataclasses.dataclass(frozen=True)
 class OperatorToProgram(workflow.Workflow[AOT_FOP, AOT_PRG]):
diff --git a/src/gt4py/next/ffront/gtcallable.py b/src/gt4py/next/ffront/gtcallable.py
index beaebb3a..cdfb2391 100644
--- a/src/gt4py/next/ffront/gtcallable.py
+++ b/src/gt4py/next/ffront/gtcallable.py
@@ -52,6 +52,16 @@ class GTCallable(typing.Protocol):
         """
         ...
 
+    # FIXME[#1582](tehrengruber): remove after refactoring to GTIR
+    @abc.abstractmethod
+    def __gt_gtir__(self) -> itir.FunctionDefinition:
+        """
+        Return iterator IR function definition representing the callable.
+        Used internally by the Program decorator to populate the function
+        definitions of the iterator IR.
+        """
+        ...
+
     # TODO(tehrengruber): For embedded execution a `__call__` method and for
     #  "truly" embedded execution arguably also a `from_function` method is
     #  required. Since field operators currently have a `__gt_type__` with a
diff --git a/src/gt4py/next/ffront/past_to_itir.py b/src/gt4py/next/ffront/past_to_itir.py
index a20c517c..14d70557 100644
--- a/src/gt4py/next/ffront/past_to_itir.py
+++ b/src/gt4py/next/ffront/past_to_itir.py
@@ -80,11 +80,18 @@ def past_to_itir(inp: AOT_PRG, to_gtir: bool = False) -> stages.CompilableProgra
     gt_callables = transform_utils._filter_closure_vars_by_type(
         all_closure_vars, gtcallable.GTCallable
     ).values()
+
+    # FIXME[#1582](tehrengruber): remove after refactoring to GTIR
     # TODO(ricoh): The following calls to .__gt_itir__, which will use whatever
-    # backend is set for each of these field operators (GTCallables). Instead
-    # we should use the current toolchain to lower these to ITIR. This will require
-    # making this step aware of the toolchain it is called by (it can be part of multiple).
-    lowered_funcs = [gt_callable.__gt_itir__() for gt_callable in gt_callables]
+    #  backend is set for each of these field operators (GTCallables). Instead
+    #  we should use the current toolchain to lower these to ITIR. This will require
+    #  making this step aware of the toolchain it is called by (it can be part of multiple).
+    lowered_funcs = []
+    for gt_callable in gt_callables:
+        if to_gtir:
+            lowered_funcs.append(gt_callable.__gt_gtir__())
+        else:
+            lowered_funcs.append(gt_callable.__gt_itir__())
 
     itir_program = ProgramLowering.apply(
         inp.data.past_node, function_definitions=lowered_funcs, grid_type=grid_type, to_gtir=to_gtir
diff --git a/src/gt4py/next/iterator/transforms/__init__.py b/src/gt4py/next/iterator/transforms/__init__.py
index 58678cfc..6f9651a3 100644
--- a/src/gt4py/next/iterator/transforms/__init__.py
+++ b/src/gt4py/next/iterator/transforms/__init__.py
@@ -6,7 +6,12 @@
 # Please, refer to the LICENSE file in the root directory.
 # SPDX-License-Identifier: BSD-3-Clause
 
-from gt4py.next.iterator.transforms.pass_manager import LiftMode, apply_common_transforms
+from gt4py.next.iterator.transforms.pass_manager import (
+    ITIRTransform,
+    LiftMode,
+    apply_common_transforms,
+    apply_fieldview_transforms,
+)
 
 
-__all__ = ["apply_common_transforms", "LiftMode"]
+__all__ = ["apply_common_transforms", "apply_fieldview_transforms", "LiftMode", "ITIRTransform"]
diff --git a/src/gt4py/next/iterator/transforms/pass_manager.py b/src/gt4py/next/iterator/transforms/pass_manager.py
index b3bb7bc6..7c35d552 100644
--- a/src/gt4py/next/iterator/transforms/pass_manager.py
+++ b/src/gt4py/next/iterator/transforms/pass_manager.py
@@ -7,11 +7,12 @@
 # SPDX-License-Identifier: BSD-3-Clause
 
 import enum
-from typing import Callable, Optional
+from typing import Callable, Optional, Protocol
 
 from gt4py.eve import utils as eve_utils
+from gt4py.next import common
 from gt4py.next.iterator import ir as itir
-from gt4py.next.iterator.transforms import fencil_to_program, inline_fundefs
+from gt4py.next.iterator.transforms import fencil_to_program, infer_domain, inline_fundefs
 from gt4py.next.iterator.transforms.collapse_list_get import CollapseListGet
 from gt4py.next.iterator.transforms.collapse_tuple import CollapseTuple
 from gt4py.next.iterator.transforms.constant_folding import ConstantFolding
@@ -29,6 +30,12 @@ from gt4py.next.iterator.transforms.scan_eta_reduction import ScanEtaReduction
 from gt4py.next.iterator.transforms.unroll_reduce import UnrollReduce
 
 
+class ITIRTransform(Protocol):
+    def __call__(
+        self, _: itir.Program | itir.FencilDefinition, *, offset_provider: common.OffsetProvider
+    ) -> itir.Program: ...
+
+
 @enum.unique
 class LiftMode(enum.Enum):
     FORCE_INLINE = enum.auto()
@@ -65,7 +72,7 @@ def _inline_into_scan(ir, *, max_iter=10):
 # TODO(tehrengruber): Revisit interface to configure temporary extraction. We currently forward
 #  `lift_mode` and `temporary_extraction_heuristics` which is inconvenient.
 def apply_common_transforms(
-    ir: itir.Node,
+    ir: itir.Program | itir.FencilDefinition,
     *,
     lift_mode=None,
     offset_provider=None,
@@ -115,10 +122,10 @@ def apply_common_transforms(
             #  other cases we want it anyway.
             force_inline_trivial_lift_args=True,
         )
-        inlined = ConstantFolding.apply(inlined)
+        inlined = ConstantFolding.apply(inlined)  # type: ignore[assignment] # still a `itir.Program`
         # This pass is required to be in the loop such that when an `if_` call with tuple arguments
         # is constant-folded the surrounding tuple_get calls can be removed.
-        inlined = CollapseTuple.apply(
+        inlined = CollapseTuple.apply(  # type: ignore[assignment] # still a `itir.Program`
             inlined,
             offset_provider=offset_provider,
             # TODO(tehrengruber): disabled since it increases compile-time too much right now
@@ -167,7 +174,7 @@ def apply_common_transforms(
     # larger than the number of closure outputs as given by the unconditional collapse, we can
     # only run the unconditional version here instead of in the loop above.
     if unconditionally_collapse_tuples:
-        ir = CollapseTuple.apply(
+        ir = CollapseTuple.apply(  # type: ignore[assignment] # still a `itir.Program`
             ir,
             ignore_tuple_size=True,
             offset_provider=offset_provider,
@@ -188,7 +195,7 @@ def apply_common_transforms(
             unrolled = UnrollReduce.apply(ir, offset_provider=offset_provider)
             if unrolled == ir:
                 break
-            ir = unrolled
+            ir = unrolled  # type: ignore[assignment] # still a `itir.Program`
             ir = CollapseListGet().visit(ir)
             ir = NormalizeShifts().visit(ir)
             ir = _inline_lifts(ir, LiftMode.FORCE_INLINE)
@@ -200,7 +207,7 @@ def apply_common_transforms(
     ir = ScanEtaReduction().visit(ir)
 
     if common_subexpression_elimination:
-        ir = CommonSubexpressionElimination.apply(ir, offset_provider=offset_provider)  # type: ignore[type-var]  # always an itir.Program
+        ir = CommonSubexpressionElimination.apply(ir, offset_provider=offset_provider)
         ir = MergeLet().visit(ir)
 
     ir = InlineLambdas.apply(
@@ -209,3 +216,17 @@ def apply_common_transforms(
 
     assert isinstance(ir, itir.Program)
     return ir
+
+
+def apply_fieldview_transforms(
+    ir: itir.Program, *, offset_provider: common.OffsetProvider
+) -> itir.Program:
+    ir = inline_fundefs.InlineFundefs().visit(ir)
+    ir = inline_fundefs.prune_unreferenced_fundefs(ir)
+    ir = InlineLambdas.apply(ir, opcount_preserving=True)
+    ir = infer_domain.infer_program(
+        ir,
+        offset_provider=offset_provider,
+    )
+    ir = CollapseTuple.apply(ir, offset_provider=offset_provider)  # type: ignore[assignment] # type is still `itir.Program`
+    return ir
diff --git a/src/gt4py/next/program_processors/formatters/lisp.py b/src/gt4py/next/program_processors/formatters/lisp.py
index c477795c..7b722a7c 100644
--- a/src/gt4py/next/program_processors/formatters/lisp.py
+++ b/src/gt4py/next/program_processors/formatters/lisp.py
@@ -50,7 +50,7 @@ class ToLispLike(TemplatedGenerator):
     )
 
     @classmethod
-    def apply(cls, root: itir.Node, **kwargs: Any) -> str:  # type: ignore[override]
+    def apply(cls, root: itir.FencilDefinition, **kwargs: Any) -> str:  # type: ignore[override]
         transformed = apply_common_transforms(
             root, lift_mode=kwargs.get("lift_mode"), offset_provider=kwargs["offset_provider"]
         )
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_builtin_translators.py b/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_builtin_translators.py
index a8ae1cc0..277d8a0c 100644
--- a/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_builtin_translators.py
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_builtin_translators.py
@@ -34,9 +34,25 @@ if TYPE_CHECKING:
 
 
 @dataclasses.dataclass(frozen=True)
-class Field:
-    data_node: dace.nodes.AccessNode
-    data_type: ts.FieldType | ts.ScalarType
+class FieldopData:
+    """
+    Abstraction to represent data (scalars, arrays) during the lowering to SDFG.
+
+    Attribute 'local_offset' must always be set for `FieldType` data with a local
+    dimension generated from neighbors access in unstructured domain, and indicates
+    the name of the offset provider used to generate the list of neighbor values.
+
+    Args:
+        dc_node: DaCe access node to the data storage.
+        gt_dtype: GT4Py type definition, which includes the field domain information.
+        local_offset: Provides information about the local dimension in`FieldType` data.
+            Set to 'None' for scalar data. Can be 'None' for `FieldType` data with
+            only global (horizontal or vertical) dimensions.
+    """
+
+    dc_node: dace.nodes.AccessNode
+    gt_dtype: ts.FieldType | ts.ScalarType
+    local_offset: Optional[str]
 
 
 FieldopDomain: TypeAlias = list[
@@ -50,7 +66,7 @@ Domain of a field operator represented as a list of tuples with 3 elements:
 """
 
 
-FieldopResult: TypeAlias = Field | tuple[Field | tuple, ...]
+FieldopResult: TypeAlias = FieldopData | tuple[FieldopData | tuple, ...]
 """Result of a field operator, can be either a field or a tuple fields."""
 
 
@@ -73,7 +89,7 @@ class PrimitiveTranslator(Protocol):
         This method is used by derived classes to build a specialized subgraph
         for a specific GTIR primitive function.
 
-        Arguments:
+        Args:
             node: The GTIR node describing the primitive to be lowered
             sdfg: The SDFG where the primitive subgraph should be instantiated
             state: The SDFG state where the result of the primitive function should be made available
@@ -108,24 +124,24 @@ def _parse_fieldop_arg(
     )
 
     # arguments passed to field operator should be plain fields, not tuples of fields
-    if not isinstance(arg, Field):
+    if not isinstance(arg, FieldopData):
         raise ValueError(f"Received {node} as argument to field operator, expected a field.")
 
-    if isinstance(arg.data_type, ts.ScalarType):
-        return gtir_dataflow.MemletExpr(arg.data_node, sbs.Indices([0]))
-    elif isinstance(arg.data_type, ts.FieldType):
-        indices: dict[gtx_common.Dimension, gtir_dataflow.ValueExpr] = {
+    if isinstance(arg.gt_dtype, ts.ScalarType):
+        return gtir_dataflow.MemletExpr(arg.dc_node, sbs.Indices([0]))
+    elif isinstance(arg.gt_dtype, ts.FieldType):
+        indices: dict[gtx_common.Dimension, gtir_dataflow.DataExpr] = {
             dim: gtir_dataflow.SymbolExpr(dace_gtir_utils.get_map_variable(dim), INDEX_DTYPE)
             for dim, _, _ in domain
         }
-        dims = arg.data_type.dims + (
+        dims = arg.gt_dtype.dims + (
             # we add an extra anonymous dimension in the iterator definition to enable
             # dereferencing elements in `ListType`
-            [gtx_common.Dimension("")] if isinstance(arg.data_type.dtype, itir_ts.ListType) else []
+            [gtx_common.Dimension("")] if isinstance(arg.gt_dtype.dtype, itir_ts.ListType) else []
         )
-        return gtir_dataflow.IteratorExpr(arg.data_node, dims, indices)
+        return gtir_dataflow.IteratorExpr(arg.dc_node, dims, indices, arg.local_offset)
     else:
-        raise NotImplementedError(f"Node type {type(arg.data_type)} not supported.")
+        raise NotImplementedError(f"Node type {type(arg.gt_dtype)} not supported.")
 
 
 def _create_temporary_field(
@@ -134,7 +150,7 @@ def _create_temporary_field(
     domain: FieldopDomain,
     node_type: ts.FieldType,
     dataflow_output: gtir_dataflow.DataflowOutputEdge,
-) -> Field:
+) -> FieldopData:
     """Helper method to allocate a temporary field where to write the output of a field operator."""
     domain_dims, _, domain_ubs = zip(*domain)
     field_dims = list(domain_dims)
@@ -148,7 +164,7 @@ def _create_temporary_field(
     # eliminate most of transient arrays.
     field_shape = list(domain_ubs)
 
-    output_desc = dataflow_output.result.node.desc(sdfg)
+    output_desc = dataflow_output.result.dc_node.desc(sdfg)
     if isinstance(output_desc, dace.data.Array):
         assert isinstance(node_type.dtype, itir_ts.ListType)
         assert isinstance(node_type.dtype.element_type, ts.ScalarType)
@@ -165,7 +181,7 @@ def _create_temporary_field(
     field_node = state.add_access(temp_name)
     field_type = ts.FieldType(field_dims, node_type.dtype)
 
-    return Field(field_node, field_type)
+    return FieldopData(field_node, field_type, local_offset=dataflow_output.result.local_offset)
 
 
 def extract_domain(node: gtir.Node) -> FieldopDomain:
@@ -284,7 +300,7 @@ def translate_as_fieldop(
     # represent the field operator as a mapped tasklet graph, which will range over the field domain
     taskgen = gtir_dataflow.LambdaToDataflow(sdfg, state, sdfg_builder, reduce_identity)
     input_edges, output = taskgen.visit(stencil_expr, args=stencil_args)
-    output_desc = output.result.node.desc(sdfg)
+    output_desc = output.result.dc_node.desc(sdfg)
 
     if isinstance(node.type.dtype, itir_ts.ListType):
         assert isinstance(output_desc, dace.data.Array)
@@ -314,7 +330,7 @@ def translate_as_fieldop(
         edge.connect(me)
 
     # and here the edge writing the result data through the map exit node
-    output.connect(mx, result_field.data_node, output_subset)
+    output.connect(mx, result_field.dc_node, output_subset)
 
     return result_field
 
@@ -353,7 +369,7 @@ def translate_broadcast_scalar(
     assert isinstance(scalar_expr, gtir_dataflow.MemletExpr)
     assert scalar_expr.subset == sbs.Indices.from_string("0")
     result = gtir_dataflow.DataflowOutputEdge(
-        state, gtir_dataflow.DataExpr(scalar_expr.node, node.args[0].type)
+        state, gtir_dataflow.ValueExpr(scalar_expr.dc_node, node.args[0].type)
     )
     result_field = _create_temporary_field(sdfg, state, domain, node.type, dataflow_output=result)
 
@@ -364,11 +380,11 @@ def translate_broadcast_scalar(
             dace_gtir_utils.get_map_variable(dim): f"{lower_bound}:{upper_bound}"
             for dim, lower_bound, upper_bound in domain
         },
-        inputs={"__inp": dace.Memlet(data=scalar_expr.node.data, subset="0")},
+        inputs={"__inp": dace.Memlet(data=scalar_expr.dc_node.data, subset="0")},
         code="__val = __inp",
-        outputs={"__val": dace.Memlet(data=result_field.data_node.data, subset=domain_indices)},
-        input_nodes={scalar_expr.node.data: scalar_expr.node},
-        output_nodes={result_field.data_node.data: result_field.data_node},
+        outputs={"__val": dace.Memlet(data=result_field.dc_node.data, subset=domain_indices)},
+        input_nodes={scalar_expr.dc_node.data: scalar_expr.dc_node},
+        output_nodes={result_field.dc_node.data: result_field.dc_node},
         external_edges=True,
     )
 
@@ -431,16 +447,16 @@ def translate_if(
         reduce_identity=reduce_identity,
     )
 
-    def make_temps(x: Field) -> Field:
-        desc = x.data_node.desc(sdfg)
+    def make_temps(output_data: FieldopData) -> FieldopData:
+        desc = output_data.dc_node.desc(sdfg)
         data_name, _ = sdfg.add_temp_transient_like(desc)
         data_node = state.add_access(data_name)
 
-        return Field(data_node, x.data_type)
+        return FieldopData(data_node, output_data.gt_dtype, output_data.local_offset)
 
     result_temps = gtx_utils.tree_map(make_temps)(true_br_args)
 
-    fields: Iterable[tuple[Field, Field, Field]] = zip(
+    fields: Iterable[tuple[FieldopData, FieldopData, FieldopData]] = zip(
         gtx_utils.flatten_nested_tuple((true_br_args,)),
         gtx_utils.flatten_nested_tuple((false_br_args,)),
         gtx_utils.flatten_nested_tuple((result_temps,)),
@@ -448,11 +464,11 @@ def translate_if(
     )
 
     for true_br, false_br, temp in fields:
-        assert true_br.data_type == false_br.data_type
-        true_br_node = true_br.data_node
-        false_br_node = false_br.data_node
+        assert true_br.gt_dtype == false_br.gt_dtype
+        true_br_node = true_br.dc_node
+        false_br_node = false_br.dc_node
 
-        temp_name = temp.data_node.data
+        temp_name = temp.dc_node.data
         true_br_output_node = true_state.add_access(temp_name)
         true_state.add_nedge(
             true_br_node,
@@ -479,16 +495,27 @@ def _get_data_nodes(
 ) -> FieldopResult:
     if isinstance(sym_type, ts.FieldType):
         sym_node = state.add_access(sym_name)
-        return Field(sym_node, sym_type)
-    elif isinstance(sym_type, ts.ScalarType):
-        if sym_name in sdfg.arrays:
-            # access the existing scalar container
-            sym_node = state.add_access(sym_name)
+        local_dims = [dim for dim in sym_type.dims if dim.kind == gtx_common.DimensionKind.LOCAL]
+        if len(local_dims) > 1:
+            raise ValueError(f"Field {sym_name} has more than one local dimension.")
+        elif len(local_dims) == 1:
+            # we ensure that the name of the local dimension corresponds to a valid
+            # connectivity-based offset provider
+            local_offset = next(iter(local_dims)).value
+            assert isinstance(
+                sdfg_builder.get_offset_provider(local_offset), gtx_common.Connectivity
+            )
         else:
+            local_offset = None
+        return FieldopData(sym_node, sym_type, local_offset)
+    elif isinstance(sym_type, ts.ScalarType):
+        if sym_name in sdfg.symbols:
             sym_node = _get_symbolic_value(
                 sdfg, state, sdfg_builder, sym_name, sym_type, temp_name=f"__{sym_name}"
             )
-        return Field(sym_node, sym_type)
+        else:
+            sym_node = state.add_access(sym_name)
+        return FieldopData(sym_node, sym_type, local_offset=None)
     elif isinstance(sym_type, ts.TupleType):
         tuple_fields = dace_gtir_utils.get_tuple_fields(sym_name, sym_type)
         return tuple(
@@ -544,7 +571,7 @@ def translate_literal(
     data_type = node.type
     data_node = _get_symbolic_value(sdfg, state, sdfg_builder, node.value, data_type)
 
-    return Field(data_node, data_type)
+    return FieldopData(data_node, data_type, local_offset=None)
 
 
 def translate_make_tuple(
@@ -587,13 +614,13 @@ def translate_tuple_get(
         head_state=state,
         reduce_identity=reduce_identity,
     )
-    if isinstance(data_nodes, Field):
+    if isinstance(data_nodes, FieldopData):
         raise ValueError(f"Invalid tuple expression {node}")
-    unused_arg_nodes: Iterable[Field] = gtx_utils.flatten_nested_tuple(
+    unused_arg_nodes: Iterable[FieldopData] = gtx_utils.flatten_nested_tuple(
         tuple(arg for i, arg in enumerate(data_nodes) if i != index)
     )
     state.remove_nodes_from(
-        [arg.data_node for arg in unused_arg_nodes if state.degree(arg.data_node) == 0]
+        [arg.dc_node for arg in unused_arg_nodes if state.degree(arg.dc_node) == 0]
     )
     return data_nodes[index]
 
@@ -612,7 +639,7 @@ def translate_scalar_expr(
     connectors = []
     scalar_expr_args = []
 
-    for arg_expr in node.args:
+    for i, arg_expr in enumerate(node.args):
         visit_expr = True
         if isinstance(arg_expr, gtir.SymRef):
             try:
@@ -634,10 +661,10 @@ def translate_scalar_expr(
                 head_state=state,
                 reduce_identity=reduce_identity,
             )
-            if not (isinstance(arg, Field) and isinstance(arg.data_type, ts.ScalarType)):
+            if not (isinstance(arg, FieldopData) and isinstance(arg.gt_dtype, ts.ScalarType)):
                 raise ValueError(f"Invalid argument to scalar expression {arg_expr}.")
-            param = f"__in_{arg.data_node.data}"
-            args.append(arg.data_node)
+            param = f"__arg{i}"
+            args.append(arg.dc_node)
             connectors.append(param)
             scalar_expr_args.append(gtir.SymRef(id=param))
         else:
@@ -679,7 +706,7 @@ def translate_scalar_expr(
         dace.Memlet(data=temp_name, subset="0"),
     )
 
-    return Field(temp_node, node.type)
+    return FieldopData(temp_node, node.type, local_offset=None)
 
 
 def translate_symbol_ref(
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_dataflow.py b/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_dataflow.py
index 0e571fc1..4f6a1e04 100644
--- a/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_dataflow.py
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_dataflow.py
@@ -30,19 +30,43 @@ from gt4py.next.type_system import type_specifications as ts
 
 
 @dataclasses.dataclass(frozen=True)
-class DataExpr:
-    """Local storage for the computation result returned by a tasklet node."""
+class ValueExpr:
+    """
+    Local storage for the values returned by dataflow computation.
+
+    This type is used in the context in a dataflow, that is a stencil expression.
+    Therefore, it contains either a scalar value (single elements in the fields) or
+    a list of values in a local dimension.
+    This is different from `gtir_builtin_translators.FieldopData` which represents
+    the result of a field operator, basically the data storage outside a global map.
 
-    node: dace.nodes.AccessNode
-    dtype: itir_ts.ListType | ts.ScalarType
+    Args:
+        dc_node: Access node to the data storage, can be either a scalar or a local list.
+        gt_dtype: GT4Py type definition, which includes the field domain information.
+        local_offset: Provides information about the local dimension in`FieldType` data.
+            For a more detailed explanation see `gtir_builtin_translators.FieldopData`.
+    """
+
+    dc_node: dace.nodes.AccessNode
+    gt_dtype: itir_ts.ListType | ts.ScalarType
+    local_offset: Optional[str] = None
 
 
 @dataclasses.dataclass(frozen=True)
 class MemletExpr:
-    """Scalar or array data access through a memlet."""
+    """
+    Scalar or array data access through a memlet.
 
-    node: dace.nodes.AccessNode
+    Args:
+        dc_node: Access node to the data storage, can be either a scalar or a local list.
+        subset: Represents the subset to use in memlet to access the above data.
+        local_offset: Provides information about the local dimension in`FieldType` data.
+            For a more detailed explanation see `gtir_builtin_translators.FieldopData`.
+    """
+
+    dc_node: dace.nodes.AccessNode
     subset: sbs.Indices | sbs.Range
+    local_offset: Optional[str] = None
 
 
 @dataclasses.dataclass(frozen=True)
@@ -50,10 +74,10 @@ class SymbolExpr:
     """Any symbolic expression that is constant in the context of current SDFG."""
 
     value: dace.symbolic.SymExpr
-    dtype: dace.typeclass
+    dc_dtype: dace.typeclass
 
 
-ValueExpr: TypeAlias = DataExpr | MemletExpr | SymbolExpr
+DataExpr: TypeAlias = ValueExpr | MemletExpr | SymbolExpr
 
 
 @dataclasses.dataclass(frozen=True)
@@ -62,18 +86,20 @@ class IteratorExpr:
     Iterator for field access to be consumed by `deref` or `shift` builtin functions.
 
     Args:
-        field: The field this iterator operates on.
-        dimensions: Field domain represented as a sorted list of dimensions.
-                    In order to dereference an element in the field, we need index values
-                    for all the dimensions in the right order.
+        field: Access node to the field this iterator operates on.
+        dimensions: Field domain represented as a sorted list of dimensions, needed
+            to order the map index variables and dereference an element in the field.
         indices: Maps each dimension to an index value, which could be either a symbolic value
-                 or the result of a tasklet computation like neighbors connectivity or dynamic offset.
+            or the result of a tasklet computation like neighbors connectivity or dynamic offset.
+        local_offset: Provides information about the local dimension in`FieldType` data.
+            For a more detailed explanation see `gtir_builtin_translators.FieldopData`.
 
     """
 
     field: dace.nodes.AccessNode
     dimensions: list[gtx_common.Dimension]
-    indices: dict[gtx_common.Dimension, ValueExpr]
+    indices: dict[gtx_common.Dimension, DataExpr]
+    local_offset: Optional[str] = None
 
 
 class DataflowInputEdge(Protocol):
@@ -119,7 +145,7 @@ class MemletInputEdge(DataflowInputEdge):
 @dataclasses.dataclass(frozen=True)
 class EmptyInputEdge(DataflowInputEdge):
     """
-    Allows to setup an edge from a map entry node to a tasklet with no arguements.
+    Allows to setup an edge from a map entry node to a tasklet with no arguments.
 
     The reason behind this kind of connection is that all nodes inside a map scope
     must have an in/out path that traverses the entry and exit nodes.
@@ -145,30 +171,30 @@ class DataflowOutputEdge:
     """
 
     state: dace.SDFGState
-    result: DataExpr
+    result: ValueExpr
 
     def connect(
         self,
         mx: dace.nodes.MapExit,
-        result_node: dace.nodes.AccessNode,
+        dest: dace.nodes.AccessNode,
         subset: sbs.Range,
     ) -> None:
         # retrieve the node which writes the result
-        last_node = self.state.in_edges(self.result.node)[0].src
+        last_node = self.state.in_edges(self.result.dc_node)[0].src
         if isinstance(last_node, dace.nodes.Tasklet):
             # the last transient node can be deleted
-            last_node_connector = self.state.in_edges(self.result.node)[0].src_conn
-            self.state.remove_node(self.result.node)
+            last_node_connector = self.state.in_edges(self.result.dc_node)[0].src_conn
+            self.state.remove_node(self.result.dc_node)
         else:
-            last_node = self.result.node
+            last_node = self.result.dc_node
             last_node_connector = None
 
         self.state.add_memlet_path(
             last_node,
             mx,
-            result_node,
+            dest,
             src_conn=last_node_connector,
-            memlet=dace.Memlet(data=result_node.data, subset=subset),
+            memlet=dace.Memlet(data=dest.data, subset=subset),
         )
 
 
@@ -187,7 +213,7 @@ DACE_REDUCTION_MAPPING: dict[str, dace.dtypes.ReductionType] = {
 
 def get_reduce_params(node: gtir.FunCall) -> tuple[str, SymbolExpr, SymbolExpr]:
     assert isinstance(node.type, ts.ScalarType)
-    dtype = dace_utils.as_dace_type(node.type)
+    dc_dtype = dace_utils.as_dace_type(node.type)
 
     assert isinstance(node.fun, gtir.FunCall)
     assert len(node.fun.args) == 2
@@ -195,12 +221,12 @@ def get_reduce_params(node: gtir.FunCall) -> tuple[str, SymbolExpr, SymbolExpr]:
     op_name = str(node.fun.args[0])
     assert isinstance(node.fun.args[1], gtir.Literal)
     assert node.fun.args[1].type == node.type
-    reduce_init = SymbolExpr(node.fun.args[1].value, dtype)
+    reduce_init = SymbolExpr(node.fun.args[1].value, dc_dtype)
 
     if op_name not in DACE_REDUCTION_MAPPING:
         raise RuntimeError(f"Reduction operation '{op_name}' not supported.")
-    identity_value = dace.dtypes.reduction_identity(dtype, DACE_REDUCTION_MAPPING[op_name])
-    reduce_identity = SymbolExpr(identity_value, dtype)
+    identity_value = dace.dtypes.reduction_identity(dc_dtype, DACE_REDUCTION_MAPPING[op_name])
+    reduce_identity = SymbolExpr(identity_value, dc_dtype)
 
     return op_name, reduce_init, reduce_identity
 
@@ -325,9 +351,9 @@ class LambdaToDataflow(eve.NodeVisitor):
             name, self.state, map_ranges, inputs, code, outputs, **kwargs
         )
 
-    def _construct_local_view(self, field: MemletExpr | DataExpr) -> DataExpr:
+    def _construct_local_view(self, field: MemletExpr | ValueExpr) -> ValueExpr:
         if isinstance(field, MemletExpr):
-            desc = field.node.desc(self.sdfg)
+            desc = field.dc_node.desc(self.sdfg)
             local_dim_indices = [i for i, size in enumerate(field.subset.size()) if size != 1]
             if len(local_dim_indices) == 0:
                 # we are accessing a single-element array with shape (1,)
@@ -337,36 +363,37 @@ class LambdaToDataflow(eve.NodeVisitor):
                 view_shape = tuple(desc.shape[i] for i in local_dim_indices)
                 view_strides = tuple(desc.strides[i] for i in local_dim_indices)
             view, _ = self.sdfg.add_view(
-                f"{field.node.data}_view",
+                f"{field.dc_node.data}_view",
                 view_shape,
                 desc.dtype,
                 strides=view_strides,
                 find_new_name=True,
             )
             local_view_node = self.state.add_access(view)
-            self._add_input_data_edge(field.node, field.subset, local_view_node)
+            self._add_input_data_edge(field.dc_node, field.subset, local_view_node)
 
-            return DataExpr(local_view_node, desc.dtype)
+            return ValueExpr(local_view_node, desc.dtype)
 
         else:
             return field
 
     def _construct_tasklet_result(
         self,
-        dtype: dace.typeclass,
+        dc_dtype: dace.typeclass,
         src_node: dace.nodes.Tasklet,
         src_connector: str,
+        local_offset: Optional[str] = None,
         use_array: bool = False,
-    ) -> DataExpr:
+    ) -> ValueExpr:
         temp_name = self.sdfg.temp_data_name()
         if use_array:
             # In some cases, such as result data with list-type annotation, we want
             # that output data is represented as an array (single-element 1D array)
             # in order to allow for composition of array shape in external memlets.
-            self.sdfg.add_array(temp_name, (1,), dtype, transient=True)
+            self.sdfg.add_array(temp_name, (1,), dc_dtype, transient=True)
         else:
-            self.sdfg.add_scalar(temp_name, dtype, transient=True)
-        data_type = dace_utils.as_itir_type(dtype)
+            self.sdfg.add_scalar(temp_name, dc_dtype, transient=True)
+        data_type = dace_utils.as_itir_type(dc_dtype)
         temp_node = self.state.add_access(temp_name)
         self._add_edge(
             src_node,
@@ -375,9 +402,9 @@ class LambdaToDataflow(eve.NodeVisitor):
             None,
             dace.Memlet(data=temp_name, subset="0"),
         )
-        return DataExpr(temp_node, data_type)
+        return ValueExpr(temp_node, data_type, local_offset)
 
-    def _visit_deref(self, node: gtir.FunCall) -> ValueExpr:
+    def _visit_deref(self, node: gtir.FunCall) -> DataExpr:
         """
         Visit a `deref` node, which represents dereferencing of an iterator.
         The iterator is the argument of this node.
@@ -409,7 +436,7 @@ class LambdaToDataflow(eve.NodeVisitor):
                     else (0, size - 1, 1)
                     for dim, size in zip(arg_expr.dimensions, field_desc.shape)
                 )
-                return MemletExpr(arg_expr.field, field_subset)
+                return MemletExpr(arg_expr.field, field_subset, arg_expr.local_offset)
 
             else:
                 # we use a tasklet to dereference an iterator when one or more indices are the result of some computation,
@@ -449,31 +476,33 @@ class LambdaToDataflow(eve.NodeVisitor):
                     deref_connector = IndexConnectorFmt.format(dim=dim.value)
                     if isinstance(index_expr, MemletExpr):
                         self._add_input_data_edge(
-                            index_expr.node,
+                            index_expr.dc_node,
                             index_expr.subset,
                             deref_node,
                             deref_connector,
                         )
 
-                    elif isinstance(index_expr, DataExpr):
+                    elif isinstance(index_expr, ValueExpr):
                         self._add_edge(
-                            index_expr.node,
+                            index_expr.dc_node,
                             None,
                             deref_node,
                             deref_connector,
-                            dace.Memlet(data=index_expr.node.data, subset="0"),
+                            dace.Memlet(data=index_expr.dc_node.data, subset="0"),
                         )
                     else:
                         assert isinstance(index_expr, SymbolExpr)
 
-                dtype = arg_expr.field.desc(self.sdfg).dtype
-                return self._construct_tasklet_result(dtype, deref_node, "val")
+                dc_dtype = arg_expr.field.desc(self.sdfg).dtype
+                return self._construct_tasklet_result(
+                    dc_dtype, deref_node, "val", arg_expr.local_offset
+                )
 
         else:
             # dereferencing a scalar or a literal node results in the node itself
             return arg_expr
 
-    def _visit_neighbors(self, node: gtir.FunCall) -> DataExpr:
+    def _visit_neighbors(self, node: gtir.FunCall) -> ValueExpr:
         assert len(node.args) == 2
         assert isinstance(node.type, itir_ts.ListType)
 
@@ -532,24 +561,26 @@ class LambdaToDataflow(eve.NodeVisitor):
         )
         neighbors_node = self.state.add_access(neighbors_temp)
 
-        offset_dim = gtx_common.Dimension(offset, kind=gtx_common.DimensionKind.LOCAL)
-        neighbor_idx = dace_gtir_utils.get_map_variable(offset_dim)
+        neighbor_idx = dace_gtir_utils.get_map_variable(offset)
 
         index_connector = "__index"
         output_connector = "__val"
         tasklet_expression = f"{output_connector} = __field[{index_connector}]"
         input_memlets = {
-            "__field": self.sdfg.make_array_memlet(field_slice.node.data),
-            index_connector: dace.Memlet(data=connectivity_slice.node.data, subset=neighbor_idx),
+            "__field": self.sdfg.make_array_memlet(field_slice.dc_node.data),
+            index_connector: dace.Memlet(data=connectivity_slice.dc_node.data, subset=neighbor_idx),
         }
         input_nodes = {
-            field_slice.node.data: field_slice.node,
-            connectivity_slice.node.data: connectivity_slice.node,
+            field_slice.dc_node.data: field_slice.dc_node,
+            connectivity_slice.dc_node.data: connectivity_slice.dc_node,
         }
 
         if offset_provider.has_skip_values:
-            assert self.reduce_identity is not None
-            assert self.reduce_identity.dtype == field_desc.dtype
+            if self.reduce_identity is None:
+                raise ValueError(
+                    f"Found local offset '{offset}' with skip values, but 'reduce_identity' is not set."
+                )
+            assert self.reduce_identity.dc_dtype == field_desc.dtype
             tasklet_expression += f" if {index_connector} != {gtx_common._DEFAULT_SKIP_VALUE} else {field_desc.dtype}({self.reduce_identity.value})"
 
         self._add_mapped_tasklet(
@@ -565,9 +596,9 @@ class LambdaToDataflow(eve.NodeVisitor):
             external_edges=True,
         )
 
-        return DataExpr(neighbors_node, node.type)
+        return ValueExpr(neighbors_node, node.type, offset)
 
-    def _visit_map(self, node: gtir.FunCall) -> DataExpr:
+    def _visit_map(self, node: gtir.FunCall) -> ValueExpr:
         """
         A map node defines an operation to be mapped on all elements of input arguments.
 
@@ -588,7 +619,7 @@ class LambdaToDataflow(eve.NodeVisitor):
         assert len(node.fun.args) == 1  # the operation to be mapped on the arguments
 
         assert isinstance(node.type.element_type, ts.ScalarType)
-        dtype = dace_utils.as_dace_type(node.type.element_type)
+        dc_dtype = dace_utils.as_dace_type(node.type.element_type)
 
         input_args = [self.visit(arg) for arg in node.args]
         input_connectors = [f"__arg{i}" for i in range(len(input_args))]
@@ -599,9 +630,27 @@ class LambdaToDataflow(eve.NodeVisitor):
         fun_python_code = gtir_python_codegen.get_source(fun_node)
         tasklet_expression = f"{output_connector} = {fun_python_code}"
 
-        # TODO(edopao): extract offset_dim from the input arguments
-        offset_dim = gtx_common.Dimension("", gtx_common.DimensionKind.LOCAL)
-        map_index = dace_gtir_utils.get_map_variable(offset_dim)
+        input_local_offsets = [
+            input_arg.local_offset for input_arg in input_args if input_arg.local_offset is not None
+        ]
+        if len(input_local_offsets) == 0:
+            raise ValueError(f"Missing information on local dimension for map node {node}.")
+
+        # GT4Py guarantees that all connectivities used to generate lists of neighbors
+        # have the same length, that is the same value of 'max_neighbors'.
+        local_connectivities = dace_utils.filter_connectivities(
+            {
+                offset: self.subgraph_builder.get_offset_provider(offset)
+                for offset in input_local_offsets
+            }
+        )
+        if len(set(table.max_neighbors for table in local_connectivities.values())) != 1:
+            raise ValueError(
+                "Unexpected arguments to map expression with different local dimensions."
+            )
+        local_offset, offset_provider = next(iter(local_connectivities.items()))
+        local_size = offset_provider.max_neighbors
+        map_index = dace_gtir_utils.get_map_variable(local_offset)
 
         # The dataflow we build in this class has some loose connections on input edges.
         # These edges are described as set of nodes, that will have to be connected to
@@ -611,9 +660,9 @@ class LambdaToDataflow(eve.NodeVisitor):
         # than representing map-to-map edges (which require memlets with 2 pass-nodes).
         input_memlets = {}
         input_nodes = {}
-        local_size: Optional[int] = None
+        skip_value_connectivities: dict[str, gtx_common.Connectivity] = {}
         for conn, input_expr in zip(input_connectors, input_args):
-            input_node = self._construct_local_view(input_expr).node
+            input_node = self._construct_local_view(input_expr).dc_node
             input_desc = input_node.desc(self.sdfg)
             # we assume that there is a single local dimension
             if len(input_desc.shape) != 1:
@@ -621,21 +670,59 @@ class LambdaToDataflow(eve.NodeVisitor):
             input_size = input_desc.shape[0]
             if input_size == 1:
                 input_memlets[conn] = dace.Memlet(data=input_node.data, subset="0")
-            elif local_size is not None and input_size != local_size:
-                raise ValueError(f"Invalid node {node}")
+            elif input_size != local_size:
+                raise ValueError(
+                    f"Argument to map node with local size {input_size}, expected {local_size}."
+                )
             else:
+                assert input_expr.local_offset
                 input_memlets[conn] = dace.Memlet(data=input_node.data, subset=map_index)
-                local_size = input_size
 
             input_nodes[input_node.data] = input_node
 
-        if local_size is None:
-            # corner case where map is applied to 1-element lists
-            assert len(input_nodes) >= 1
-            local_size = 1
+        result, _ = self.sdfg.add_temp_transient((local_size,), dc_dtype)
+        result_node = self.state.add_access(result)
+
+        skip_value_connectivities = {
+            offset: offset_provider
+            for offset, offset_provider in local_connectivities.items()
+            if offset_provider.has_skip_values
+        }
+
+        if len(skip_value_connectivities) == 0:
+            result_offset = local_offset
+        else:
+            # In case one or more of input expressions contain skip values, we use
+            # the connectivity-based offset provider as mask for map computation.
+            # Therefore, the result of map computation will also contain skip values.
+            # GT4Py guarantees that the skip values are placed in the same positions
+            # for all input expressions.
+
+            result_offset, offset_provider = next(iter(skip_value_connectivities.items()))
+
+            connectivity = dace_utils.connectivity_identifier(result_offset)
+            connectivity_desc = self.sdfg.arrays[connectivity]
+            connectivity_desc.transient = False
 
-        out, _ = self.sdfg.add_temp_transient((local_size,), dtype)
-        out_node = self.state.add_access(out)
+            origin_map_index = dace_gtir_utils.get_map_variable(offset_provider.origin_axis)
+
+            connectivity_slice = self._construct_local_view(
+                MemletExpr(
+                    self.state.add_access(connectivity),
+                    sbs.Range.from_string(f"{origin_map_index}, 0:{offset_provider.max_neighbors}"),
+                )
+            )
+
+            if self.reduce_identity is None:
+                raise ValueError(
+                    f"Found local offset '{result_offset}' with skip values, but 'reduce_identity' is not set."
+                )
+            assert self.reduce_identity.dc_dtype == dc_dtype
+            input_memlets["__neighbor_idx"] = dace.Memlet(
+                data=connectivity_slice.dc_node.data, subset=map_index
+            )
+            input_nodes[connectivity_slice.dc_node.data] = connectivity_slice.dc_node
+            tasklet_expression += f" if __neighbor_idx != {gtx_common._DEFAULT_SKIP_VALUE} else {dc_dtype}({self.reduce_identity.value})"
 
         self._add_mapped_tasklet(
             name="map",
@@ -644,15 +731,15 @@ class LambdaToDataflow(eve.NodeVisitor):
             inputs=input_memlets,
             input_nodes=input_nodes,
             outputs={
-                output_connector: dace.Memlet(data=out, subset=map_index),
+                output_connector: dace.Memlet(data=result, subset=map_index),
             },
-            output_nodes={out: out_node},
+            output_nodes={result: result_node},
             external_edges=True,
         )
 
-        return DataExpr(out_node, dtype)
+        return ValueExpr(result_node, dc_dtype, result_offset)
 
-    def _visit_reduce(self, node: gtir.FunCall) -> DataExpr:
+    def _visit_reduce(self, node: gtir.FunCall) -> ValueExpr:
         assert isinstance(node.type, ts.ScalarType)
         op_name, reduce_init, reduce_identity = get_reduce_params(node)
 
@@ -672,8 +759,8 @@ class LambdaToDataflow(eve.NodeVisitor):
             # ensure that we leave the visitor in the same state as we entered
             self.reduce_identity = prev_reduce_identity
 
-        assert isinstance(input_expr, MemletExpr | DataExpr)
-        input_desc = input_expr.node.desc(self.sdfg)
+        assert isinstance(input_expr, MemletExpr | ValueExpr)
+        input_desc = input_expr.dc_node.desc(self.sdfg)
         assert isinstance(input_desc, dace.data.Array)
 
         if len(input_desc.shape) > 1:
@@ -691,19 +778,19 @@ class LambdaToDataflow(eve.NodeVisitor):
 
         if isinstance(input_expr, MemletExpr):
             self._add_input_data_edge(
-                input_expr.node,
+                input_expr.dc_node,
                 input_expr.subset,
                 reduce_node,
             )
         else:
             self.state.add_nedge(
-                input_expr.node,
+                input_expr.dc_node,
                 reduce_node,
-                dace.Memlet.from_array(input_expr.node.data, input_desc),
+                dace.Memlet.from_array(input_expr.dc_node.data, input_desc),
             )
 
         temp_name = self.sdfg.temp_data_name()
-        self.sdfg.add_scalar(temp_name, reduce_identity.dtype, transient=True)
+        self.sdfg.add_scalar(temp_name, reduce_identity.dc_dtype, transient=True)
         temp_node = self.state.add_access(temp_name)
 
         self.state.add_nedge(
@@ -711,7 +798,7 @@ class LambdaToDataflow(eve.NodeVisitor):
             temp_node,
             dace.Memlet(data=temp_name, subset="0"),
         )
-        return DataExpr(temp_node, node.type)
+        return ValueExpr(temp_node, node.type)
 
     def _split_shift_args(
         self, args: list[gtir.Expr]
@@ -742,18 +829,18 @@ class LambdaToDataflow(eve.NodeVisitor):
         return offset_provider_arg, offset_value_arg, it
 
     def _make_cartesian_shift(
-        self, it: IteratorExpr, offset_dim: gtx_common.Dimension, offset_expr: ValueExpr
+        self, it: IteratorExpr, offset_dim: gtx_common.Dimension, offset_expr: DataExpr
     ) -> IteratorExpr:
         """Implements cartesian shift along one dimension."""
         assert offset_dim in it.dimensions
-        new_index: SymbolExpr | DataExpr
+        new_index: SymbolExpr | ValueExpr
         assert offset_dim in it.indices
         index_expr = it.indices[offset_dim]
         if isinstance(index_expr, SymbolExpr) and isinstance(offset_expr, SymbolExpr):
             # purely symbolic expression which can be interpreted at compile time
             new_index = SymbolExpr(
                 dace.symbolic.pystr_to_symbolic(index_expr.value) + offset_expr.value,
-                index_expr.dtype,
+                index_expr.dc_dtype,
             )
         else:
             # the offset needs to be calculated by means of a tasklet (i.e. dynamic offset)
@@ -782,27 +869,27 @@ class LambdaToDataflow(eve.NodeVisitor):
             for input_expr, input_connector in [(index_expr, "index"), (offset_expr, "offset")]:
                 if isinstance(input_expr, MemletExpr):
                     self._add_input_data_edge(
-                        input_expr.node,
+                        input_expr.dc_node,
                         input_expr.subset,
                         dynamic_offset_tasklet,
                         input_connector,
                     )
-                elif isinstance(input_expr, DataExpr):
+                elif isinstance(input_expr, ValueExpr):
                     self._add_edge(
-                        input_expr.node,
+                        input_expr.dc_node,
                         None,
                         dynamic_offset_tasklet,
                         input_connector,
-                        dace.Memlet(data=input_expr.node.data, subset="0"),
+                        dace.Memlet(data=input_expr.dc_node.data, subset="0"),
                     )
 
             if isinstance(index_expr, SymbolExpr):
-                dtype = index_expr.dtype
+                dc_dtype = index_expr.dc_dtype
             else:
-                dtype = index_expr.node.desc(self.sdfg).dtype
+                dc_dtype = index_expr.dc_node.desc(self.sdfg).dtype
 
             new_index = self._construct_tasklet_result(
-                dtype, dynamic_offset_tasklet, new_index_connector
+                dc_dtype, dynamic_offset_tasklet, new_index_connector
             )
 
         # a new iterator with a shifted index along one dimension
@@ -814,10 +901,10 @@ class LambdaToDataflow(eve.NodeVisitor):
 
     def _make_dynamic_neighbor_offset(
         self,
-        offset_expr: MemletExpr | DataExpr,
+        offset_expr: MemletExpr | ValueExpr,
         offset_table_node: dace.nodes.AccessNode,
         origin_index: SymbolExpr,
-    ) -> DataExpr:
+    ) -> ValueExpr:
         """
         Implements access to neighbor connectivity table by means of a tasklet node.
 
@@ -839,29 +926,29 @@ class LambdaToDataflow(eve.NodeVisitor):
         )
         if isinstance(offset_expr, MemletExpr):
             self._add_input_data_edge(
-                offset_expr.node,
+                offset_expr.dc_node,
                 offset_expr.subset,
                 tasklet_node,
                 "offset",
             )
         else:
             self._add_edge(
-                offset_expr.node,
+                offset_expr.dc_node,
                 None,
                 tasklet_node,
                 "offset",
-                dace.Memlet(data=offset_expr.node.data, subset="0"),
+                dace.Memlet(data=offset_expr.dc_node.data, subset="0"),
             )
 
-        dtype = offset_table_node.desc(self.sdfg).dtype
-        return self._construct_tasklet_result(dtype, tasklet_node, new_index_connector)
+        dc_dtype = offset_table_node.desc(self.sdfg).dtype
+        return self._construct_tasklet_result(dc_dtype, tasklet_node, new_index_connector)
 
     def _make_unstructured_shift(
         self,
         it: IteratorExpr,
         connectivity: gtx_common.Connectivity,
         offset_table_node: dace.nodes.AccessNode,
-        offset_expr: ValueExpr,
+        offset_expr: DataExpr,
     ) -> IteratorExpr:
         """Implements shift in unstructured domain by means of a neighbor table."""
         assert connectivity.neighbor_axis in it.dimensions
@@ -928,18 +1015,18 @@ class LambdaToDataflow(eve.NodeVisitor):
                 it, offset_provider, offset_table_node, offset_expr
             )
 
-    def _visit_generic_builtin(self, node: gtir.FunCall) -> DataExpr:
+    def _visit_generic_builtin(self, node: gtir.FunCall) -> ValueExpr:
         """
         Generic handler called by `visit_FunCall()` when it encounters
         a builtin function that does not match any other specific handler.
         """
         node_internals = []
-        node_connections: dict[str, MemletExpr | DataExpr] = {}
+        node_connections: dict[str, MemletExpr | ValueExpr] = {}
         for i, arg in enumerate(node.args):
             arg_expr = self.visit(arg)
-            if isinstance(arg_expr, MemletExpr | DataExpr):
+            if isinstance(arg_expr, MemletExpr | ValueExpr):
                 # the argument value is the result of a tasklet node or direct field access
-                connector = f"__inp_{i}"
+                connector = f"__arg{i}"
                 node_connections[connector] = arg_expr
                 node_internals.append(connector)
             else:
@@ -961,17 +1048,17 @@ class LambdaToDataflow(eve.NodeVisitor):
         )
 
         for connector, arg_expr in node_connections.items():
-            if isinstance(arg_expr, DataExpr):
+            if isinstance(arg_expr, ValueExpr):
                 self._add_edge(
-                    arg_expr.node,
+                    arg_expr.dc_node,
                     None,
                     tasklet_node,
                     connector,
-                    dace.Memlet(data=arg_expr.node.data, subset="0"),
+                    dace.Memlet(data=arg_expr.dc_node.data, subset="0"),
                 )
             else:
                 self._add_input_data_edge(
-                    arg_expr.node,
+                    arg_expr.dc_node,
                     arg_expr.subset,
                     tasklet_node,
                     connector,
@@ -988,18 +1075,18 @@ class LambdaToDataflow(eve.NodeVisitor):
             # Therefore we handle `ListType` as a single-element array with shape (1,)
             # that will be accessed in a map expression on a local domain.
             assert isinstance(node.type.element_type, ts.ScalarType)
-            dtype = dace_utils.as_dace_type(node.type.element_type)
+            dc_dtype = dace_utils.as_dace_type(node.type.element_type)
             # In order to ease the lowring of the parent expression on local dimension,
             # we represent the scalar value as a single-element 1D array.
             use_array = True
         else:
             assert isinstance(node.type, ts.ScalarType)
-            dtype = dace_utils.as_dace_type(node.type)
+            dc_dtype = dace_utils.as_dace_type(node.type)
             use_array = False
 
-        return self._construct_tasklet_result(dtype, tasklet_node, "result", use_array=use_array)
+        return self._construct_tasklet_result(dc_dtype, tasklet_node, "result", use_array=use_array)
 
-    def visit_FunCall(self, node: gtir.FunCall) -> IteratorExpr | ValueExpr:
+    def visit_FunCall(self, node: gtir.FunCall) -> IteratorExpr | DataExpr:
         if cpm.is_call_to(node, "deref"):
             return self._visit_deref(node)
 
@@ -1026,16 +1113,16 @@ class LambdaToDataflow(eve.NodeVisitor):
     ) -> tuple[list[DataflowInputEdge], DataflowOutputEdge]:
         for p, arg in zip(node.params, args, strict=True):
             self.symbol_map[str(p.id)] = arg
-        output_expr: ValueExpr = self.visit(node.expr)
-        if isinstance(output_expr, DataExpr):
+        output_expr: DataExpr = self.visit(node.expr)
+        if isinstance(output_expr, ValueExpr):
             return self.input_edges, DataflowOutputEdge(self.state, output_expr)
 
         if isinstance(output_expr, MemletExpr):
             # special case where the field operator is simply copying data from source to destination node
-            output_dtype = output_expr.node.desc(self.sdfg).dtype
+            output_dtype = output_expr.dc_node.desc(self.sdfg).dtype
             tasklet_node = self._add_tasklet("copy", {"__inp"}, {"__out"}, "__out = __inp")
             self._add_input_data_edge(
-                output_expr.node,
+                output_expr.dc_node,
                 output_expr.subset,
                 tasklet_node,
                 "__inp",
@@ -1043,15 +1130,15 @@ class LambdaToDataflow(eve.NodeVisitor):
         else:
             assert isinstance(output_expr, SymbolExpr)
             # even simpler case, where a constant value is written to destination node
-            output_dtype = output_expr.dtype
+            output_dtype = output_expr.dc_dtype
             tasklet_node = self._add_tasklet("write", {}, {"__out"}, f"__out = {output_expr.value}")
 
         output_expr = self._construct_tasklet_result(output_dtype, tasklet_node, "__out")
         return self.input_edges, DataflowOutputEdge(self.state, output_expr)
 
     def visit_Literal(self, node: gtir.Literal) -> SymbolExpr:
-        dtype = dace_utils.as_dace_type(node.type)
-        return SymbolExpr(node.value, dtype)
+        dc_dtype = dace_utils.as_dace_type(node.type)
+        return SymbolExpr(node.value, dc_dtype)
 
     def visit_SymRef(self, node: gtir.SymRef) -> IteratorExpr | MemletExpr | SymbolExpr:
         param = str(node.id)
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_sdfg.py b/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_sdfg.py
index 3697609d..4627293c 100644
--- a/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_sdfg.py
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/gtir_sdfg.py
@@ -17,6 +17,7 @@ from __future__ import annotations
 import abc
 import dataclasses
 import itertools
+import operator
 from typing import Any, Dict, Iterable, List, Optional, Protocol, Sequence, Set, Tuple, Union
 
 import dace
@@ -26,7 +27,7 @@ from gt4py.eve import concepts
 from gt4py.next import common as gtx_common, utils as gtx_utils
 from gt4py.next.iterator import ir as gtir
 from gt4py.next.iterator.ir_utils import common_pattern_matcher as cpm
-from gt4py.next.iterator.transforms import prune_casts as ir_prune_casts
+from gt4py.next.iterator.transforms import prune_casts as ir_prune_casts, symbol_ref_utils
 from gt4py.next.iterator.type_system import inference as gtir_type_inference
 from gt4py.next.program_processors.runners.dace_common import utility as dace_utils
 from gt4py.next.program_processors.runners.dace_fieldview import (
@@ -106,6 +107,38 @@ class SDFGBuilder(DataflowBuilder, Protocol):
         ...
 
 
+def _collect_symbols_in_domain_expressions(
+    ir: gtir.Node, ir_params: Sequence[gtir.Sym]
+) -> set[str]:
+    """
+    Collect symbols accessed in domain expressions that also appear in the paremeter list.
+
+    This function is used to identify all parameters that are accessed in domain
+    expressions. They have to be passed to the SDFG call as DaCe symbols (instead
+    of scalars) such that they can be used as bounds in map ranges.
+
+    Args:
+        ir: GTIR node to be traversed and where to search for domain expressions.
+        ir_params: List of parameters to search for in domain expressions.
+
+    Returns:
+        A set of names corresponding to the parameters found in domain expressions.
+    """
+    params = {str(sym.id) for sym in ir_params}
+    return set(
+        eve.walk_values(ir)
+        .filter(lambda node: cpm.is_call_to(node, ("cartesian_domain", "unstructured_domain")))
+        .map(
+            lambda domain: eve.walk_values(domain)
+            .if_isinstance(gtir.SymRef)
+            .map(lambda symref: str(symref.id))
+            .filter(lambda sym: sym in params)
+            .to_list()
+        )
+        .reduce(operator.add, init=[])
+    )
+
+
 @dataclasses.dataclass(frozen=True)
 class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
     """Provides translation capability from a GTIR program to a DaCe SDFG.
@@ -153,18 +186,18 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
         Returns:
             Two lists of symbols, one for the shape and the other for the strides of the array.
         """
-        dtype = dace.int32
+        dc_dtype = gtir_builtin_translators.INDEX_DTYPE
         neighbor_tables = dace_utils.filter_connectivities(self.offset_provider)
         shape = [
             (
                 neighbor_tables[dim.value].max_neighbors
                 if dim.kind == gtx_common.DimensionKind.LOCAL
-                else dace.symbol(dace_utils.field_size_symbol_name(name, i), dtype)
+                else dace.symbol(dace_utils.field_size_symbol_name(name, i), dc_dtype)
             )
             for i, dim in enumerate(dims)
         ]
         strides = [
-            dace.symbol(dace_utils.field_stride_symbol_name(name, i), dtype)
+            dace.symbol(dace_utils.field_stride_symbol_name(name, i), dc_dtype)
             for i in range(len(dims))
         ]
         return shape, strides
@@ -172,60 +205,79 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
     def _add_storage(
         self,
         sdfg: dace.SDFG,
+        symbolic_arguments: set[str],
         name: str,
-        symbol_type: ts.DataType,
+        gt_type: ts.DataType,
         transient: bool = True,
-        is_tuple_member: bool = False,
     ) -> list[tuple[str, ts.DataType]]:
         """
-        Add storage for data containers used in the SDFG. For fields, it allocates dace arrays,
-        while scalars are stored as SDFG symbols.
+        Add storage in the SDFG for a given GT4Py data symbol.
 
-        The fields used as temporary arrays, when `transient = True`, are allocated and exist
-        only within the SDFG; when `transient = False`, the fields have to be allocated outside
-        and have to be passed as array arguments to the SDFG.
+        GT4Py fields are allocated as DaCe arrays. GT4Py scalars are represented
+        as DaCe scalar objects in the SDFG; the exception are the symbols passed as
+        `symbolic_arguments`, e.g. symbols used in domain expressions, and those used
+        for symbolic array shape and strides.
+
+        The fields used as temporary arrays, when `transient = True`, are allocated
+        and exist only within the SDFG; when `transient = False`, the fields have
+        to be allocated outside and have to be passed as arguments to the SDFG call.
+
+        Args:
+            sdfg: The SDFG where storage needs to be allocated.
+            symbolic_arguments: Set of GT4Py scalars that must be represented as SDFG symbols.
+            name: Symbol Name to be allocated.
+            gt_type: GT4Py symbol type.
+            transient: True when the data symbol has to be allocated as internal storage.
 
         Returns:
-            List of data containers or symbols allocated as storage. This is a list, not a single value,
-            because in case of tuples we flat the tuple fields (eventually nested) and allocate storage
-            for each tuple element.
+            List of tuples '(data_name, gt_type)' where 'data_name' is the name of
+            the data container used as storage in the SDFG and 'gt_type' is the
+            corresponding GT4Py type. In case the storage has to be allocated for
+            a tuple symbol the list contains a flattened version of the tuple,
+            otherwise the list will contain a single entry.
         """
-        if isinstance(symbol_type, ts.TupleType):
+        if isinstance(gt_type, ts.TupleType):
             tuple_fields = []
             for tname, tsymbol_type in dace_gtir_utils.get_tuple_fields(
-                name, symbol_type, flatten=True
+                name, gt_type, flatten=True
             ):
                 tuple_fields.extend(
-                    self._add_storage(sdfg, tname, tsymbol_type, transient, is_tuple_member=True)
+                    self._add_storage(sdfg, symbolic_arguments, tname, tsymbol_type, transient)
                 )
             return tuple_fields
 
-        elif isinstance(symbol_type, ts.FieldType):
-            dtype = dace_utils.as_dace_type(symbol_type.dtype)
+        elif isinstance(gt_type, ts.FieldType):
+            dc_dtype = dace_utils.as_dace_type(gt_type.dtype)
             # use symbolic shape, which allows to invoke the program with fields of different size;
             # and symbolic strides, which enables decoupling the memory layout from generated code.
-            sym_shape, sym_strides = self._make_array_shape_and_strides(name, symbol_type.dims)
-            sdfg.add_array(name, sym_shape, dtype, strides=sym_strides, transient=transient)
-
-            return [(name, symbol_type)]
-
-        elif isinstance(symbol_type, ts.ScalarType):
-            dtype = dace_utils.as_dace_type(symbol_type)
-            # Scalar arguments passed to the program are represented as symbols in DaCe SDFG;
-            # the exception are members of tuple arguments, that are represented as scalar containers.
-            # The field size is sometimes passed as scalar argument to the program, so we have to
-            # check if the shape symbol was already allocated by `_make_array_shape_and_strides`.
-            # We assume that the scalar argument for field size always follows the field argument.
-            if is_tuple_member:
-                sdfg.add_scalar(name, dtype, transient=transient)
-            elif name in sdfg.symbols:
-                assert sdfg.symbols[name].dtype == dtype
+            sym_shape, sym_strides = self._make_array_shape_and_strides(name, gt_type.dims)
+            sdfg.add_array(name, sym_shape, dc_dtype, strides=sym_strides, transient=transient)
+
+            return [(name, gt_type)]
+
+        elif isinstance(gt_type, ts.ScalarType):
+            dc_dtype = dace_utils.as_dace_type(gt_type)
+            if name in symbolic_arguments:
+                sdfg.add_symbol(name, dc_dtype)
+            elif dace_utils.is_field_symbol(name):
+                # Sometimes, when the field domain is implicitly derived from the
+                # field domain, the gt4py lowering adds the field size as a scalar
+                # argument to the program IR. Suppose a field '__sym', then gt4py
+                # will add '__sym_size_0'.
+                # Therefore, here we check whether the shape symbol was already
+                # created by `_make_array_shape_and_strides`, when allocating
+                # storage for field arguments. We assume that the scalar argument
+                # for field size, if present, always follows the field argument.
+                if name in sdfg.symbols:
+                    assert sdfg.symbols[name].dc_dtype == dc_dtype
+                else:
+                    sdfg.add_symbol(name, dc_dtype)
             else:
-                sdfg.add_symbol(name, dtype)
+                sdfg.add_scalar(name, dc_dtype, transient=transient)
 
-            return [(name, symbol_type)]
+            return [(name, gt_type)]
 
-        raise RuntimeError(f"Data type '{type(symbol_type)}' not supported.")
+        raise RuntimeError(f"Data type '{type(gt_type)}' not supported.")
 
     def _add_storage_for_temporary(self, temp_decl: gtir.Temporary) -> dict[str, str]:
         """
@@ -237,7 +289,7 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
 
     def _visit_expression(
         self, node: gtir.Expr, sdfg: dace.SDFG, head_state: dace.SDFGState, use_temp: bool = True
-    ) -> list[gtir_builtin_translators.Field]:
+    ) -> list[gtir_builtin_translators.FieldopData]:
         """
         Specialized visit method for fieldview expressions.
 
@@ -246,10 +298,6 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
 
         Returns:
             A list of array nodes containing the result fields.
-
-        TODO: Do we need to return the GT4Py `FieldType`/`ScalarType`? It is needed
-        in case the transient arrays containing the expression result are not guaranteed
-        to have the same memory layout as the target array.
         """
         result = self.visit(node, sdfg=sdfg, head_state=head_state, reduce_identity=None)
 
@@ -259,29 +307,47 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
         assert len(sink_states) == 1
         assert sink_states[0] == head_state
 
-        def make_temps(field: gtir_builtin_translators.Field) -> gtir_builtin_translators.Field:
-            desc = sdfg.arrays[field.data_node.data]
+        def make_temps(
+            field: gtir_builtin_translators.FieldopData,
+        ) -> gtir_builtin_translators.FieldopData:
+            desc = sdfg.arrays[field.dc_node.data]
             if desc.transient or not use_temp:
                 return field
             else:
                 temp, _ = sdfg.add_temp_transient_like(desc)
                 temp_node = head_state.add_access(temp)
                 head_state.add_nedge(
-                    field.data_node, temp_node, sdfg.make_array_memlet(field.data_node.data)
+                    field.dc_node, temp_node, sdfg.make_array_memlet(field.dc_node.data)
+                )
+                return gtir_builtin_translators.FieldopData(
+                    temp_node, field.gt_dtype, field.local_offset
                 )
-                return gtir_builtin_translators.Field(temp_node, field.data_type)
 
         temp_result = gtx_utils.tree_map(make_temps)(result)
         return list(gtx_utils.flatten_nested_tuple((temp_result,)))
 
-    def _add_sdfg_params(self, sdfg: dace.SDFG, node_params: Sequence[gtir.Sym]) -> list[str]:
-        """Helper function to add storage for node parameters and connectivity tables."""
+    def _add_sdfg_params(
+        self,
+        sdfg: dace.SDFG,
+        node_params: Sequence[gtir.Sym],
+        symbolic_arguments: set[str],
+    ) -> list[str]:
+        """
+        Helper function to add storage for node parameters and connectivity tables.
+
+        GT4Py field arguments will be translated to `dace.data.Array` objects.
+        GT4Py scalar arguments will be translated to `dace.data.Scalar` objects,
+        except when they are listed in 'symbolic_arguments', in which case they
+        will be represented in the SDFG as DaCe symbols.
+        """
         # add non-transient arrays and/or SDFG symbols for the program arguments
         sdfg_args = []
         for param in node_params:
             pname = str(param.id)
             assert isinstance(param.type, (ts.DataType))
-            sdfg_args += self._add_storage(sdfg, pname, param.type, transient=False)
+            sdfg_args += self._add_storage(
+                sdfg, symbolic_arguments, pname, param.type, transient=False
+            )
             self.global_symbols[pname] = param.type
 
         # add SDFG storage for connectivity tables
@@ -290,7 +356,7 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
         ).items():
             scalar_kind = tt.get_scalar_kind(offset_provider.index_type)
             local_dim = gtx_common.Dimension(offset, kind=gtx_common.DimensionKind.LOCAL)
-            type_ = ts.FieldType(
+            gt_type = ts.FieldType(
                 [offset_provider.origin_axis, local_dim], ts.ScalarType(scalar_kind)
             )
             # We store all connectivity tables as transient arrays here; later, while building
@@ -298,7 +364,9 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
             # the tables that are actually used. This way, we avoid adding SDFG arguments for
             # the connectivity tables that are not used. The remaining unused transient arrays
             # are removed by the dace simplify pass.
-            self._add_storage(sdfg, dace_utils.connectivity_identifier(offset), type_)
+            self._add_storage(
+                sdfg, symbolic_arguments, dace_utils.connectivity_identifier(offset), gt_type
+            )
 
         # the list of all sdfg arguments (aka non-transient arrays) which include tuple-element fields
         return [arg_name for arg_name, _ in sdfg_args]
@@ -330,7 +398,8 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
         else:
             head_state = entry_state
 
-        sdfg_arg_names = self._add_sdfg_params(sdfg, node.params)
+        domain_symbols = _collect_symbols_in_domain_expressions(node, node.params)
+        sdfg_arg_names = self._add_sdfg_params(sdfg, node.params, symbolic_arguments=domain_symbols)
 
         # visit one statement at a time and expand the SDFG from the current head state
         for i, stmt in enumerate(node.body):
@@ -384,35 +453,35 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
 
         target_state: Optional[dace.SDFGState] = None
         for temp, target in zip(temp_fields, target_fields, strict=True):
-            target_desc = sdfg.arrays[target.data_node.data]
+            target_desc = sdfg.arrays[target.dc_node.data]
             assert not target_desc.transient
 
-            if isinstance(target.data_type, ts.FieldType):
+            if isinstance(target.gt_dtype, ts.FieldType):
                 subset = ",".join(
-                    f"{domain[dim][0]}:{domain[dim][1]}" for dim in target.data_type.dims
+                    f"{domain[dim][0]}:{domain[dim][1]}" for dim in target.gt_dtype.dims
                 )
             else:
                 assert len(domain) == 0
                 subset = "0"
 
-            if target.data_node.data in state_input_data:
+            if target.dc_node.data in state_input_data:
                 # if inout argument, write the result in separate next state
                 # this is needed to avoid undefined behavior for expressions like: X, Y = X + 1, X
                 if not target_state:
                     target_state = sdfg.add_state_after(state, f"post_{state.label}")
                 # create new access nodes in the target state
                 target_state.add_nedge(
-                    target_state.add_access(temp.data_node.data),
-                    target_state.add_access(target.data_node.data),
-                    dace.Memlet(data=target.data_node.data, subset=subset, other_subset=subset),
+                    target_state.add_access(temp.dc_node.data),
+                    target_state.add_access(target.dc_node.data),
+                    dace.Memlet(data=target.dc_node.data, subset=subset, other_subset=subset),
                 )
                 # remove isolated access node
-                state.remove_node(target.data_node)
+                state.remove_node(target.dc_node)
             else:
                 state.add_nedge(
-                    temp.data_node,
-                    target.data_node,
-                    dace.Memlet(data=target.data_node.data, subset=subset, other_subset=subset),
+                    temp.dc_node,
+                    target.dc_node,
+                    dace.Memlet(data=target.dc_node.data, subset=subset, other_subset=subset),
                 )
 
         return target_state or state
@@ -490,8 +559,11 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
         ]
 
         # inherit symbols from parent scope but eventually override with local symbols
-        lambda_symbols = self.global_symbols | {
-            pname: dace_gtir_utils.get_tuple_type(arg) if isinstance(arg, tuple) else arg.data_type
+        lambda_symbols = {
+            sym: self.global_symbols[sym]
+            for sym in symbol_ref_utils.collect_symbol_refs(node.expr, self.global_symbols.keys())
+        } | {
+            pname: dace_gtir_utils.get_tuple_type(arg) if isinstance(arg, tuple) else arg.gt_dtype
             for pname, arg in lambda_args_mapping
         }
 
@@ -501,11 +573,12 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
         nstate = nsdfg.add_state("lambda")
 
         # add sdfg storage for the symbols that need to be passed as input parameters
+        lambda_params = [
+            gtir.Sym(id=p_name, type=p_type) for p_name, p_type in lambda_symbols.items()
+        ]
+        lambda_domain_symbols = _collect_symbols_in_domain_expressions(node.expr, lambda_params)
         lambda_translator._add_sdfg_params(
-            nsdfg,
-            node_params=[
-                gtir.Sym(id=p_name, type=p_type) for p_name, p_type in lambda_symbols.items()
-            ],
+            nsdfg, node_params=lambda_params, symbolic_arguments=lambda_domain_symbols
         )
 
         lambda_result = lambda_translator.visit(
@@ -518,7 +591,7 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
         def _flatten_tuples(
             name: str,
             arg: gtir_builtin_translators.FieldopResult,
-        ) -> list[tuple[str, gtir_builtin_translators.Field]]:
+        ) -> list[tuple[str, gtir_builtin_translators.FieldopData]]:
             if isinstance(arg, tuple):
                 tuple_type = dace_gtir_utils.get_tuple_type(arg)
                 tuple_field_names = [
@@ -548,7 +621,7 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
                 continue
             datadesc: Optional[dace.dtypes.Data] = None
             if nsdfg_dataname in lambda_arg_nodes:
-                src_node = lambda_arg_nodes[nsdfg_dataname].data_node
+                src_node = lambda_arg_nodes[nsdfg_dataname].dc_node
                 dataname = src_node.data
                 datadesc = src_node.desc(sdfg)
             else:
@@ -573,18 +646,20 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
 
         # Process lambda outputs
         #
-        lambda_output_nodes: Iterable[gtir_builtin_translators.Field] = (
+        lambda_output_data: Iterable[gtir_builtin_translators.FieldopData] = (
             gtx_utils.flatten_nested_tuple(lambda_result)
         )
         # sanity check on isolated nodes
         assert all(
-            nstate.degree(x.data_node) == 0
-            for x in lambda_output_nodes
-            if x.data_node.data in input_memlets
+            nstate.degree(output_data.dc_node) == 0
+            for output_data in lambda_output_data
+            if output_data.dc_node.data in input_memlets
         )
         # keep only non-isolated output nodes
         lambda_outputs = {
-            x.data_node.data for x in lambda_output_nodes if x.data_node.data not in input_memlets
+            output_data.dc_node.data
+            for output_data in lambda_output_data
+            if output_data.dc_node.data not in input_memlets
         }
 
         if lambda_outputs:
@@ -599,36 +674,40 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
 
             for connector, memlet in input_memlets.items():
                 if connector in lambda_arg_nodes:
-                    src_node = lambda_arg_nodes[connector].data_node
+                    src_node = lambda_arg_nodes[connector].dc_node
                 else:
                     src_node = head_state.add_access(memlet.data)
 
                 head_state.add_edge(src_node, None, nsdfg_node, connector, memlet)
 
         def make_temps(
-            x: gtir_builtin_translators.Field,
-        ) -> gtir_builtin_translators.Field:
-            if x.data_node.data in lambda_outputs:
-                connector = x.data_node.data
-                desc = x.data_node.desc(nsdfg)
+            output_data: gtir_builtin_translators.FieldopData,
+        ) -> gtir_builtin_translators.FieldopData:
+            if output_data.dc_node.data in lambda_outputs:
+                connector = output_data.dc_node.data
+                desc = output_data.dc_node.desc(nsdfg)
                 # make lambda result non-transient and map it to external temporary
                 desc.transient = False
                 # isolated access node will make validation fail
-                if nstate.degree(x.data_node) == 0:
-                    nstate.remove_node(x.data_node)
+                if nstate.degree(output_data.dc_node) == 0:
+                    nstate.remove_node(output_data.dc_node)
                 temp, _ = sdfg.add_temp_transient_like(desc)
                 dst_node = head_state.add_access(temp)
                 head_state.add_edge(
                     nsdfg_node, connector, dst_node, None, sdfg.make_array_memlet(temp)
                 )
-                return gtir_builtin_translators.Field(dst_node, x.data_type)
-            elif x.data_node.data in lambda_arg_nodes:
-                nstate.remove_node(x.data_node)
-                return lambda_arg_nodes[x.data_node.data]
+                return gtir_builtin_translators.FieldopData(
+                    dst_node, output_data.gt_dtype, output_data.local_offset
+                )
+            elif output_data.dc_node.data in lambda_arg_nodes:
+                nstate.remove_node(output_data.dc_node)
+                return lambda_arg_nodes[output_data.dc_node.data]
             else:
-                nstate.remove_node(x.data_node)
-                data_node = head_state.add_access(x.data_node.data)
-                return gtir_builtin_translators.Field(data_node, x.data_type)
+                nstate.remove_node(output_data.dc_node)
+                data_node = head_state.add_access(output_data.dc_node.data)
+                return gtir_builtin_translators.FieldopData(
+                    data_node, output_data.gt_dtype, output_data.local_offset
+                )
 
         return gtx_utils.tree_map(make_temps)(lambda_result)
 
@@ -665,7 +744,7 @@ def build_sdfg_from_gtir(
     The lowering to SDFG requires that the program node is type-annotated, therefore this function
     runs type ineference as first step.
 
-    Arguments:
+    Args:
         ir: The GTIR program node to be lowered to SDFG
         offset_provider: The definitions of offset providers used by the program node
 
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/utility.py b/src/gt4py/next/program_processors/runners/dace_fieldview/utility.py
index b5c447a1..355eaac9 100644
--- a/src/gt4py/next/program_processors/runners/dace_fieldview/utility.py
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/utility.py
@@ -15,12 +15,16 @@ from gt4py.next import common as gtx_common
 from gt4py.next.type_system import type_specifications as ts
 
 
-def get_map_variable(dim: gtx_common.Dimension) -> str:
+def get_map_variable(dim: gtx_common.Dimension | str) -> str:
     """
     Format map variable name based on the naming convention for application-specific SDFG transformations.
     """
+    if not isinstance(dim, gtx_common.Dimension):
+        if len(dim) != 0:
+            dim = gtx_common.Dimension(dim, gtx_common.DimensionKind.LOCAL)
+        else:
+            raise ValueError("Dimension name cannot be empty.")
     suffix = "dim" if dim.kind == gtx_common.DimensionKind.LOCAL else ""
-    # TODO(edopao): raise exception if dim.value is empty
     return f"i_{dim.value}_gtx_{dim.kind}{suffix}"
 
 
@@ -60,5 +64,5 @@ def get_tuple_type(data: tuple[Any, ...]) -> ts.TupleType:
     Compute the `ts.TupleType` corresponding to the structure of a tuple of data nodes.
     """
     return ts.TupleType(
-        types=[get_tuple_type(d) if isinstance(d, tuple) else d.data_type for d in data]
+        types=[get_tuple_type(d) if isinstance(d, tuple) else d.gt_dtype for d in data]
     )
diff --git a/src/gt4py/next/program_processors/runners/dace_fieldview/workflow.py b/src/gt4py/next/program_processors/runners/dace_fieldview/workflow.py
index ffc33a9f..f2953eb0 100644
--- a/src/gt4py/next/program_processors/runners/dace_fieldview/workflow.py
+++ b/src/gt4py/next/program_processors/runners/dace_fieldview/workflow.py
@@ -17,7 +17,7 @@ import factory
 
 from gt4py._core import definitions as core_defs
 from gt4py.next import common, config
-from gt4py.next.iterator import ir as itir
+from gt4py.next.iterator import ir as itir, transforms as itir_transforms
 from gt4py.next.otf import languages, recipes, stages, step_types, workflow
 from gt4py.next.otf.binding import interface
 from gt4py.next.otf.languages import LanguageSettings
@@ -46,10 +46,8 @@ class DaCeTranslator(
         offset_provider: common.OffsetProvider,
         column_axis: Optional[common.Dimension],
     ) -> dace.SDFG:
-        # TODO(edopao): Call IR transformations and domain inference, finally lower IR to SDFG
-        raise NotImplementedError
-
-        return gtir_sdfg.build_sdfg_from_gtir(program=ir, offset_provider=offset_provider)
+        ir = itir_transforms.apply_fieldview_transforms(ir, offset_provider=offset_provider)
+        return gtir_sdfg.build_sdfg_from_gtir(ir=ir, offset_provider=offset_provider)
 
     def __call__(
         self, inp: stages.CompilableProgram
diff --git a/src/gt4py/next/program_processors/runners/roundtrip.py b/src/gt4py/next/program_processors/runners/roundtrip.py
index 93e6d09c..57785ceb 100644
--- a/src/gt4py/next/program_processors/runners/roundtrip.py
+++ b/src/gt4py/next/program_processors/runners/roundtrip.py
@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import dataclasses
+import functools
 import importlib.util
 import pathlib
 import tempfile
@@ -20,7 +21,7 @@ from typing import Any, Optional
 from gt4py.eve import codegen
 from gt4py.eve.codegen import FormatTemplate as as_fmt, MakoTemplate as as_mako
 from gt4py.next import allocators as next_allocators, backend as next_backend, common, config
-from gt4py.next.ffront import foast_to_gtir, past_to_itir
+from gt4py.next.ffront import foast_to_gtir, foast_to_past, past_to_itir
 from gt4py.next.iterator import ir as itir, transforms as itir_transforms
 from gt4py.next.otf import stages, workflow
 from gt4py.next.type_system import type_specifications as ts
@@ -90,11 +91,11 @@ _FENCIL_CACHE: dict[int, Callable] = {}
 
 
 def fencil_generator(
-    ir: itir.Node,
+    ir: itir.Program | itir.FencilDefinition,
     debug: bool,
-    lift_mode: itir_transforms.LiftMode,
     use_embedded: bool,
     offset_provider: dict[str, common.Connectivity | common.Dimension],
+    transforms: itir_transforms.ITIRTransform,
 ) -> stages.CompiledProgram:
     """
     Generate a directly executable fencil from an ITIR node.
@@ -102,7 +103,6 @@ def fencil_generator(
     Arguments:
         ir: The iterator IR (ITIR) node.
         debug: Keep module source containing fencil implementation.
-        lift_mode: Change the way lifted function calls are evaluated.
         use_embedded: Directly use builtins from embedded backend instead of
                       generic dispatcher. Gives faster performance and is easier
                       to debug.
@@ -110,15 +110,13 @@ def fencil_generator(
     """
     # TODO(tehrengruber): just a temporary solution until we have a proper generic
     #  caching mechanism
-    cache_key = hash((ir, lift_mode, debug, use_embedded, tuple(offset_provider.items())))
+    cache_key = hash((ir, transforms, debug, use_embedded, tuple(offset_provider.items())))
     if cache_key in _FENCIL_CACHE:
         if debug:
             print(f"Using cached fencil for key {cache_key}")
         return typing.cast(stages.CompiledProgram, _FENCIL_CACHE[cache_key])
 
-    ir = itir_transforms.apply_common_transforms(
-        ir, lift_mode=lift_mode, offset_provider=offset_provider
-    )
+    ir = transforms(ir, offset_provider=offset_provider)
 
     program = EmbeddedDSL.apply(ir)
 
@@ -187,9 +185,9 @@ def fencil_generator(
 @dataclasses.dataclass(frozen=True)
 class Roundtrip(workflow.Workflow[stages.CompilableProgram, stages.CompiledProgram]):
     debug: Optional[bool] = None
-    lift_mode: itir_transforms.LiftMode = itir_transforms.LiftMode.FORCE_INLINE
     use_embedded: bool = True
     dispatch_backend: Optional[next_backend.Backend] = None
+    transforms: itir_transforms.ITIRTransform = itir_transforms.apply_common_transforms  # type: ignore[assignment] # TODO(havogt): cleanup interface of `apply_common_transforms`
 
     def __call__(self, inp: stages.CompilableProgram) -> stages.CompiledProgram:
         debug = config.DEBUG if self.debug is None else self.debug
@@ -198,8 +196,8 @@ class Roundtrip(workflow.Workflow[stages.CompilableProgram, stages.CompiledProgr
             inp.data,
             offset_provider=inp.args.offset_provider,
             debug=debug,
-            lift_mode=self.lift_mode,
             use_embedded=self.use_embedded,
+            transforms=self.transforms,
         )
 
         def decorated_fencil(
@@ -224,28 +222,38 @@ class Roundtrip(workflow.Workflow[stages.CompilableProgram, stages.CompiledProgr
         return decorated_fencil
 
 
-executor = Roundtrip()
-executor_with_temporaries = Roundtrip(lift_mode=itir_transforms.LiftMode.USE_TEMPORARIES)
-
 default = next_backend.Backend(
     name="roundtrip",
-    executor=executor,
+    executor=Roundtrip(
+        transforms=functools.partial(
+            itir_transforms.apply_common_transforms, lift_mode=itir_transforms.LiftMode.FORCE_INLINE
+        )
+    ),
     allocator=next_allocators.StandardCPUFieldBufferAllocator(),
     transforms=next_backend.DEFAULT_TRANSFORMS,
 )
 with_temporaries = next_backend.Backend(
     name="roundtrip_with_temporaries",
-    executor=executor_with_temporaries,
+    executor=Roundtrip(
+        transforms=functools.partial(
+            itir_transforms.apply_common_transforms,
+            lift_mode=itir_transforms.LiftMode.USE_TEMPORARIES,
+        )
+    ),
     allocator=next_allocators.StandardCPUFieldBufferAllocator(),
     transforms=next_backend.DEFAULT_TRANSFORMS,
 )
 
+
 gtir = next_backend.Backend(
     name="roundtrip_gtir",
-    executor=executor,
+    executor=Roundtrip(transforms=itir_transforms.apply_fieldview_transforms),  # type: ignore[arg-type] # on purpose doesn't support `FencilDefintion` will resolve itself later...
     allocator=next_allocators.StandardCPUFieldBufferAllocator(),
     transforms=next_backend.Transforms(
         past_to_itir=past_to_itir.past_to_itir_factory(to_gtir=True),
         foast_to_itir=foast_to_gtir.adapted_foast_to_gtir_factory(cached=True),
+        field_view_op_to_prog=foast_to_past.operator_to_program_factory(
+            foast_to_itir_step=foast_to_gtir.adapted_foast_to_gtir_factory()
+        ),
     ),
 )


Test patch:
diff --git a/tests/next_tests/unit_tests/ffront_tests/test_foast_to_gtir.py b/tests/next_tests/unit_tests/ffront_tests/test_foast_to_gtir.py
index 09f18246..4a1a7cba 100644
--- a/tests/next_tests/unit_tests/ffront_tests/test_foast_to_gtir.py
+++ b/tests/next_tests/unit_tests/ffront_tests/test_foast_to_gtir.py
@@ -915,7 +915,7 @@ def test_broadcast():
     lowered = FieldOperatorLowering.apply(parsed)
 
     assert lowered.id == "foo"
-    assert lowered.expr == im.ref("inp")
+    assert lowered.expr == im.as_fieldop("deref")(im.ref("inp"))
 
 
 def test_scalar_broadcast():
diff --git a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/test_gtir_to_sdfg.py b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/test_gtir_to_sdfg.py
index 728b4b02..2c1c04ce 100644
--- a/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/test_gtir_to_sdfg.py
+++ b/tests/next_tests/unit_tests/program_processor_tests/runners_tests/dace_tests/test_gtir_to_sdfg.py
@@ -819,9 +819,7 @@ def test_gtir_cartesian_shift_left():
 
         sdfg = dace_backend.build_sdfg_from_gtir(testee, CARTESIAN_OFFSETS)
 
-        FSYMBOLS_tmp = FSYMBOLS.copy()
-        FSYMBOLS_tmp["__x_offset_stride_0"] = 1
-        sdfg(a, a_offset, b, **FSYMBOLS_tmp)
+        sdfg(a, a_offset, b, **FSYMBOLS, __x_offset_size_0=N, __x_offset_stride_0=1)
         assert np.allclose(a[OFFSET:] + DELTA, b[:-OFFSET])
 
 
@@ -914,7 +912,7 @@ def test_gtir_cartesian_shift_right():
 
         sdfg = dace_backend.build_sdfg_from_gtir(testee, CARTESIAN_OFFSETS)
 
-        sdfg(a, a_offset, b, **FSYMBOLS, __x_offset_stride_0=1)
+        sdfg(a, a_offset, b, **FSYMBOLS, __x_offset_size_0=N, __x_offset_stride_0=1)
         assert np.allclose(a[:-OFFSET] + DELTA, b[OFFSET:])
 
 
@@ -1072,7 +1070,9 @@ def test_gtir_connectivity_shift():
             __ev_field_size_1=SIMPLE_MESH.num_vertices,
             __ev_field_stride_0=SIMPLE_MESH.num_vertices,
             __ev_field_stride_1=1,
+            __c2e_offset_size_0=SIMPLE_MESH.num_cells,
             __c2e_offset_stride_0=1,
+            __e2v_offset_size_0=SIMPLE_MESH.num_edges,
             __e2v_offset_stride_0=1,
         )
         assert np.allclose(ce, ref)
@@ -1137,8 +1137,6 @@ def test_gtir_connectivity_shift_chain():
 
 
 def test_gtir_neighbors_as_input():
-    # FIXME[#1582](edopao): Enable testcase when type inference is working
-    pytest.skip("Field of lists not fully supported by GTIR type inference")
     init_value = np.random.rand()
     vertex_domain = im.domain(gtx_common.GridType.UNSTRUCTURED, ranges={Vertex: (0, "nvertices")})
     testee = gtir.Program(
@@ -1146,24 +1144,28 @@ def test_gtir_neighbors_as_input():
         function_definitions=[],
         params=[
             gtir.Sym(id="v2e_field", type=V2E_FTYPE),
-            gtir.Sym(id="vertex", type=EFTYPE),
+            gtir.Sym(id="edges", type=EFTYPE),
+            gtir.Sym(id="vertices", type=VFTYPE),
             gtir.Sym(id="nvertices", type=SIZE_TYPE),
         ],
         declarations=[],
         body=[
             gtir.SetAt(
-                expr=im.call(
-                    im.call("as_fieldop")(
-                        im.lambda_("it")(
-                            im.call(im.call("reduce")("plus", im.literal_from_value(init_value)))(
-                                "it"
-                            )
-                        ),
-                        vertex_domain,
+                expr=im.as_fieldop(
+                    im.lambda_("it")(
+                        im.call(im.call("reduce")("plus", im.literal_from_value(init_value)))(
+                            im.deref("it")
+                        )
+                    ),
+                    vertex_domain,
+                )(
+                    im.op_as_fieldop(im.map_("plus"), vertex_domain)(
+                        "v2e_field",
+                        im.as_fieldop_neighbors("V2E", "edges", vertex_domain),
                     )
-                )("v2e_field"),
+                ),
                 domain=vertex_domain,
-                target=gtir.SymRef(id="vertex"),
+                target=gtir.SymRef(id="vertices"),
             )
         ],
     )
@@ -1174,16 +1176,19 @@ def test_gtir_neighbors_as_input():
     assert isinstance(connectivity_V2E, gtx_common.NeighborTable)
 
     v2e_field = np.random.rand(SIMPLE_MESH.num_vertices, connectivity_V2E.max_neighbors)
+    e = np.random.rand(SIMPLE_MESH.num_edges)
     v = np.empty(SIMPLE_MESH.num_vertices, dtype=v2e_field.dtype)
 
     v_ref = [
-        functools.reduce(lambda x, y: x + y, v2e_neighbors, init_value)
-        for v2e_neighbors in v2e_field
+        functools.reduce(lambda x, y: x + y, v2e_values + e[v2e_neighbors], init_value)
+        for v2e_neighbors, v2e_values in zip(connectivity_V2E.table, v2e_field, strict=True)
     ]
 
     sdfg(
         v2e_field,
+        e,
         v,
+        connectivity_V2E=connectivity_V2E.table,
         **FSYMBOLS,
         **make_mesh_symbols(SIMPLE_MESH),
         __v2e_field_size_0=SIMPLE_MESH.num_vertices,
@@ -1386,13 +1391,48 @@ def test_gtir_reduce_dot_product():
     connectivity_V2E = SIMPLE_MESH_OFFSET_PROVIDER["V2E"]
     assert isinstance(connectivity_V2E, gtx_common.NeighborTable)
 
+    # create mesh with skip values
+    connectivity_V2E_skip = copy.deepcopy(connectivity_V2E)
+    connectivity_V2E_skip.has_skip_values = True
+    connectivity_V2E_skip.table = np.asarray(
+        [
+            [x if i != skip_idx else gtx_common._DEFAULT_SKIP_VALUE for i, x in enumerate(row)]
+            for skip_idx, row in zip(
+                np.random.randint(0, connectivity_V2E.max_neighbors, size=SIMPLE_MESH.num_vertices),
+                connectivity_V2E.table,
+                strict=True,
+            )
+        ],
+        dtype=connectivity_V2E.table.dtype,
+    )
+    # safety check that the connectivity table actually contains skip values
+    assert len(np.where(connectivity_V2E.table == gtx_common._DEFAULT_SKIP_VALUE)) != 0
+
+    offset_provider = SIMPLE_MESH_OFFSET_PROVIDER | {
+        "V2E_skip": connectivity_V2E_skip,
+    }
+
+    V2E_SKIP_SYMBOLS = dict(
+        __connectivity_V2E_skip_size_0=SIMPLE_MESH.num_vertices,
+        __connectivity_V2E_skip_size_1=connectivity_V2E_skip.max_neighbors,
+        __connectivity_V2E_skip_stride_0=connectivity_V2E_skip.max_neighbors,
+        __connectivity_V2E_skip_stride_1=1,
+    )
+
     e = np.random.rand(SIMPLE_MESH.num_edges)
     v = np.empty(SIMPLE_MESH.num_vertices, dtype=e.dtype)
     v_ref = [
         functools.reduce(
-            lambda x, y: x + y, (e[v2e_neighbors] * e[v2e_neighbors]) + 1.0, init_value
+            lambda x, y: x + y,
+            map(
+                lambda x: 0.0 if x[1] == gtx_common._DEFAULT_SKIP_VALUE else x[0],
+                zip((e[v2e_neighbors] * e[v2e_skip_neighbors]) + 1.0, v2e_skip_neighbors),
+            ),
+            init_value,
+        )
+        for v2e_neighbors, v2e_skip_neighbors in zip(
+            connectivity_V2E.table, connectivity_V2E_skip.table
         )
-        for v2e_neighbors in connectivity_V2E.table
     ]
 
     stencil_inlined = im.call(
@@ -1402,7 +1442,7 @@ def test_gtir_reduce_dot_product():
                     im.map_("plus")(
                         im.map_("multiplies")(
                             im.neighbors("V2E", "it"),
-                            im.neighbors("V2E", "it"),
+                            im.neighbors("V2E_skip", "it"),
                         ),
                         im.call("make_const_list")(1.0),
                     )
@@ -1425,7 +1465,7 @@ def test_gtir_reduce_dot_product():
         im.op_as_fieldop(im.map_("plus"), vertex_domain)(
             im.op_as_fieldop(im.map_("multiplies"), vertex_domain)(
                 im.as_fieldop_neighbors("V2E", "edges", vertex_domain),
-                im.as_fieldop_neighbors("V2E", "edges", vertex_domain),
+                im.as_fieldop_neighbors("V2E_skip", "edges", vertex_domain),
             ),
             im.op_as_fieldop("make_const_list", vertex_domain)(1.0),
         )
@@ -1450,14 +1490,16 @@ def test_gtir_reduce_dot_product():
             ],
         )
 
-        sdfg = dace_backend.build_sdfg_from_gtir(testee, SIMPLE_MESH_OFFSET_PROVIDER)
+        sdfg = dace_backend.build_sdfg_from_gtir(testee, offset_provider)
 
         sdfg(
             e,
             v,
             connectivity_V2E=connectivity_V2E.table,
+            connectivity_V2E_skip=connectivity_V2E_skip.table,
             **FSYMBOLS,
             **make_mesh_symbols(SIMPLE_MESH),
+            **V2E_SKIP_SYMBOLS,
         )
         assert np.allclose(v, v_ref)
 
@@ -1592,6 +1634,41 @@ def test_gtir_let_lambda():
     assert np.allclose(b, ref)
 
 
+def test_gtir_let_lambda_scalar_expression():
+    domain = im.domain(gtx_common.GridType.CARTESIAN, ranges={IDim: (0, "size")})
+    testee = gtir.Program(
+        id="let_lambda_scalar_expression",
+        function_definitions=[],
+        params=[
+            gtir.Sym(id="a", type=IFTYPE.dtype),
+            gtir.Sym(id="b", type=IFTYPE.dtype),
+            gtir.Sym(id="x", type=IFTYPE),
+            gtir.Sym(id="y", type=IFTYPE),
+            gtir.Sym(id="size", type=SIZE_TYPE),
+        ],
+        declarations=[],
+        body=[
+            gtir.SetAt(
+                expr=im.let("tmp", im.multiplies_("a", "b"))(
+                    im.op_as_fieldop("multiplies", domain)("x", im.multiplies_("tmp", "tmp"))
+                ),
+                domain=domain,
+                target=gtir.SymRef(id="y"),
+            )
+        ],
+    )
+
+    a = np.random.rand()
+    b = np.random.rand()
+    c = np.random.rand(N)
+    d = np.empty_like(c)
+
+    sdfg = dace_backend.build_sdfg_from_gtir(testee, {})
+
+    sdfg(a, b, c, d, **FSYMBOLS)
+    assert np.allclose(d, (a * a * b * b * c))
+
+
 def test_gtir_let_lambda_with_connectivity():
     C2E_neighbor_idx = 1
     C2V_neighbor_idx = 2
@@ -1757,9 +1834,9 @@ def test_gtir_if_scalars():
         body=[
             gtir.SetAt(
                 expr=im.let("f", im.tuple_get(0, "x"))(
-                    im.let("y", im.tuple_get(1, "x"))(
-                        im.let("y_0", im.tuple_get(0, "y"))(
-                            im.let("y_1", im.tuple_get(1, "y"))(
+                    im.let("g", im.tuple_get(1, "x"))(
+                        im.let("y_0", im.tuple_get(0, "g"))(
+                            im.let("y_1", im.tuple_get(1, "g"))(
                                 im.op_as_fieldop("plus", domain)(
                                     "f",
                                     im.if_(