Repository: GridTools/gt4py
Pull request #1694 (2024-10-17 19:12:43)

feat[next][dace]: GTIR-to-DaCe lowering of map-reduce with skip values
This PR extends the solution for map-reduce provided in #1683 with the support for connectivity tables with skip values.

The field definition is extended with a `local_offset` attribute that stores the offset provider used to build the values in the local dimension. In case the local dimension is built by the `neighbors` expression, the `local_offset` corresponds to the offset provider used to access the neighbor dimension. Since this information is carried along the data itself, whenever the data is accessed it is also possible to access the corresponding offset provider and check whether the neighbor index is valid or if there is a skip value. For local dimensions already present in the program argument, this information is retreived from the field domain (enabled in new test case).

The data is accessed in the `map_` and `reduce` expressions. Here it is now possible to check for skip values. Therefore, the main objective of this PR is the lowering of map-reduce with skip values.

A secondary objective is to pave the road to simplify the lowering logic, by getting rid of the `reduce_identity` value. The current approach is propagate the `reduce_identity` value while visiting the arguments to `reduce` expressions. By introducing `local_offset`, the argument visitor will return the information needed to implement `reduce` of local values in presence of skip values. This solution is available as a draft PR, see [this diff](https://github.com/edopao/gt4py/compare/dace-gtir-map_reduce_skip...edopao:gt4py:dace-gtir-reduce_identity).

edopao: cscs-ci run default
edopao: cscs-ci run default

Patch:
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 3cd2b17b..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,7 +495,19 @@ def _get_data_nodes(
 ) -> FieldopResult:
     if isinstance(sym_type, ts.FieldType):
         sym_node = state.add_access(sym_name)
-        return Field(sym_node, sym_type)
+        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(
@@ -487,7 +515,7 @@ def _get_data_nodes(
             )
         else:
             sym_node = state.add_access(sym_name)
-        return Field(sym_node, sym_type)
+        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(
@@ -543,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(
@@ -586,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]
 
@@ -633,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"__arg{i}"
-            args.append(arg.data_node)
+            args.append(arg.dc_node)
             connectors.append(param)
             scalar_expr_args.append(gtir.SymRef(id=param))
         else:
@@ -678,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 fdd92b57..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,16 +1015,16 @@ 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"__arg{i}"
                 node_connections[connector] = arg_expr
@@ -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 31e561a1..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
@@ -186,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
@@ -247,18 +247,18 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
             return tuple_fields
 
         elif isinstance(gt_type, ts.FieldType):
-            dtype = dace_utils.as_dace_type(gt_type.dtype)
+            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, gt_type.dims)
-            sdfg.add_array(name, sym_shape, dtype, strides=sym_strides, transient=transient)
+            sdfg.add_array(name, sym_shape, dc_dtype, strides=sym_strides, transient=transient)
 
             return [(name, gt_type)]
 
         elif isinstance(gt_type, ts.ScalarType):
-            dtype = dace_utils.as_dace_type(gt_type)
+            dc_dtype = dace_utils.as_dace_type(gt_type)
             if name in symbolic_arguments:
-                sdfg.add_symbol(name, dtype)
+                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
@@ -269,11 +269,11 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
                 # 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].dtype == dtype
+                    assert sdfg.symbols[name].dc_dtype == dc_dtype
                 else:
-                    sdfg.add_symbol(name, dtype)
+                    sdfg.add_symbol(name, dc_dtype)
             else:
-                sdfg.add_scalar(name, dtype, transient=transient)
+                sdfg.add_scalar(name, dc_dtype, transient=transient)
 
             return [(name, gt_type)]
 
@@ -289,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.
 
@@ -307,17 +307,21 @@ 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,)))
@@ -449,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
@@ -559,7 +563,7 @@ class GTIRToSDFG(eve.NodeVisitor, SDFGBuilder):
             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.data_type
+            pname: dace_gtir_utils.get_tuple_type(arg) if isinstance(arg, tuple) else arg.gt_dtype
             for pname, arg in lambda_args_mapping
         }
 
@@ -587,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 = [
@@ -617,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:
@@ -642,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:
@@ -668,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)
 
@@ -734,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]
     )


Test patch:
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 230ff695..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
@@ -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)