Apply a transform function to each edge and its associated source and
        target vertices in parallel. Each edge is visited once and in parallel.
        Modification to vertex data is protected by lock. The effect on the
        returned SGraph is equivalent to the following pseudocode:

        >>> PARALLEL FOR (source, edge, target) AS triple in G:
        ...     LOCK (triple.source, triple.target)
        ...     (source, edge, target) = triple_apply_fn(triple)
        ...     UNLOCK (triple.source, triple.target)
        ... END PARALLEL FOR

        Parameters
        ----------
        triple_apply_fn : function : (dict, dict, dict) -> (dict, dict, dict)
            The function to apply to each triple of (source_vertex, edge,
            target_vertex). This function must take as input a tuple of
            (source_data, edge_data, target_data) and return a tuple of
            (new_source_data, new_edge_data, new_target_data). All variables in
            the both tuples must be of dict type.
            This can also be a toolkit extension function which is compiled
            as a native shared library using SDK.

        mutated_fields : list[str] | str
            Fields that ``triple_apply_fn`` will mutate. Note: columns that are
            actually mutated by the triple apply function but not specified in
            ``mutated_fields`` will have undetermined effects.

        input_fields : list[str] | str, optional
            Fields that ``triple_apply_fn`` will have access to.
            The default is ``None``, which grants access to all fields.
            ``mutated_fields`` will always be included in ``input_fields``.

        Returns
        -------
        out : SGraph
            A new SGraph with updated vertex and edge data. Only fields
            specified in the ``mutated_fields`` parameter are updated.

        Notes
        -----
        - ``triple_apply`` does not currently support creating new fields in the
          lambda function.

        Examples
        --------
        Import turicreate and set up the graph.

        >>> edges = turicreate.SFrame({'source': range(9), 'dest': range(1, 10)})
        >>> g = turicreate.SGraph()
        >>> g = g.add_edges(edges, src_field='source', dst_field='dest')
        >>> g.vertices['degree'] = 0

        Define the function to apply to each (source_node, edge, target_node)
        triple.

        >>> def degree_count_fn (src, edge, dst):
                src['degree'] += 1
                dst['degree'] += 1
                return (src, edge, dst)

        Apply the function to the SGraph.

        >>> g = g.triple_apply(degree_count_fn, mutated_fields=['degree'])


        Using native toolkit extension function:

        .. code-block:: c++

            #include <turicreate/sdk/toolkit_function_macros.hpp>
            #include <vector>

            using namespace turi;
            std::vector<variant_type> connected_components_parameterized(
              std::map<std::string, flexible_type>& src,
              std::map<std::string, flexible_type>& edge,
              std::map<std::string, flexible_type>& dst,
              std::string column) {
                if (src[column] < dst[column]) dst[column] = src[column];
                else src[column] = dst[column];
                return {to_variant(src), to_variant(edge), to_variant(dst)};
            }

            BEGIN_FUNCTION_REGISTRATION
            REGISTER_FUNCTION(connected_components_parameterized, "src", "edge", "dst", "column");
            END_FUNCTION_REGISTRATION

        compiled into example.so

        >>> from example import connected_components_parameterized as cc
        >>> e = tc.SFrame({'__src_id':[1,2,3,4,5], '__dst_id':[3,1,2,5,4]})
        >>> g = tc.SGraph().add_edges(e)
        >>> g.vertices['cid'] = g.vertices['__id']
        >>> for i in range(2):
        ...     g = g.triple_apply(lambda src, edge, dst: cc(src, edge, dst, 'cid'), ['cid'], ['cid'])
        >>> g.vertices['cid']
        dtype: int
        Rows: 5
        [4, 1, 1, 1, 4]

def triple_apply(self, triple_apply_fn, mutated_fields, input_fields=None):
        '''
        Apply a transform function to each edge and its associated source and
        target vertices in parallel. Each edge is visited once and in parallel.
        Modification to vertex data is protected by lock. The effect on the
        returned SGraph is equivalent to the following pseudocode:

        >>> PARALLEL FOR (source, edge, target) AS triple in G:
        ...     LOCK (triple.source, triple.target)
        ...     (source, edge, target) = triple_apply_fn(triple)
        ...     UNLOCK (triple.source, triple.target)
        ... END PARALLEL FOR

        Parameters
        ----------
        triple_apply_fn : function : (dict, dict, dict) -> (dict, dict, dict)
            The function to apply to each triple of (source_vertex, edge,
            target_vertex). This function must take as input a tuple of
            (source_data, edge_data, target_data) and return a tuple of
            (new_source_data, new_edge_data, new_target_data). All variables in
            the both tuples must be of dict type.
            This can also be a toolkit extension function which is compiled
            as a native shared library using SDK.

        mutated_fields : list[str] | str
            Fields that ``triple_apply_fn`` will mutate. Note: columns that are
            actually mutated by the triple apply function but not specified in
            ``mutated_fields`` will have undetermined effects.

        input_fields : list[str] | str, optional
            Fields that ``triple_apply_fn`` will have access to.
            The default is ``None``, which grants access to all fields.
            ``mutated_fields`` will always be included in ``input_fields``.

        Returns
        -------
        out : SGraph
            A new SGraph with updated vertex and edge data. Only fields
            specified in the ``mutated_fields`` parameter are updated.

        Notes
        -----
        - ``triple_apply`` does not currently support creating new fields in the
          lambda function.

        Examples
        --------
        Import turicreate and set up the graph.

        >>> edges = turicreate.SFrame({'source': range(9), 'dest': range(1, 10)})
        >>> g = turicreate.SGraph()
        >>> g = g.add_edges(edges, src_field='source', dst_field='dest')
        >>> g.vertices['degree'] = 0

        Define the function to apply to each (source_node, edge, target_node)
        triple.

        >>> def degree_count_fn (src, edge, dst):
                src['degree'] += 1
                dst['degree'] += 1
                return (src, edge, dst)

        Apply the function to the SGraph.

        >>> g = g.triple_apply(degree_count_fn, mutated_fields=['degree'])


        Using native toolkit extension function:

        .. code-block:: c++

            #include <turicreate/sdk/toolkit_function_macros.hpp>
            #include <vector>

            using namespace turi;
            std::vector<variant_type> connected_components_parameterized(
              std::map<std::string, flexible_type>& src,
              std::map<std::string, flexible_type>& edge,
              std::map<std::string, flexible_type>& dst,
              std::string column) {
                if (src[column] < dst[column]) dst[column] = src[column];
                else src[column] = dst[column];
                return {to_variant(src), to_variant(edge), to_variant(dst)};
            }

            BEGIN_FUNCTION_REGISTRATION
            REGISTER_FUNCTION(connected_components_parameterized, "src", "edge", "dst", "column");
            END_FUNCTION_REGISTRATION

        compiled into example.so

        >>> from example import connected_components_parameterized as cc
        >>> e = tc.SFrame({'__src_id':[1,2,3,4,5], '__dst_id':[3,1,2,5,4]})
        >>> g = tc.SGraph().add_edges(e)
        >>> g.vertices['cid'] = g.vertices['__id']
        >>> for i in range(2):
        ...     g = g.triple_apply(lambda src, edge, dst: cc(src, edge, dst, 'cid'), ['cid'], ['cid'])
        >>> g.vertices['cid']
        dtype: int
        Rows: 5
        [4, 1, 1, 1, 4]
        '''

        assert inspect.isfunction(triple_apply_fn), "Input must be a function"
        if not (type(mutated_fields) is list or type(mutated_fields) is str):
            raise TypeError('mutated_fields must be str or list of str')
        if not (input_fields is None or type(input_fields) is list or type(input_fields) is str):
            raise TypeError('input_fields must be str or list of str')
        if type(mutated_fields) == str:
            mutated_fields = [mutated_fields]
        if len(mutated_fields) is 0:
            raise ValueError('mutated_fields cannot be empty')
        for f in ['__id', '__src_id', '__dst_id']:
            if f in mutated_fields:
                raise ValueError('mutated_fields cannot contain %s' % f)

        all_fields = self.get_fields()
        if not set(mutated_fields).issubset(set(all_fields)):
            extra_fields = list(set(mutated_fields).difference(set(all_fields)))
            raise ValueError('graph does not contain fields: %s' % str(extra_fields))

        # select input fields
        if input_fields is None:
            input_fields = self.get_fields()
        elif type(input_fields) is str:
            input_fields = [input_fields]

        # make input fields a superset of mutated_fields
        input_fields_set = set(input_fields + mutated_fields)
        input_fields = [x for x in self.get_fields() if x in input_fields_set]
        g = self.select_fields(input_fields)

        nativefn = None
        try:
            from .. import extensions
            nativefn = extensions._build_native_function_call(triple_apply_fn)
        except:
            # failure are fine. we just fall out into the next few phases
            pass
        if nativefn is not None:
            with cython_context():
                return SGraph(_proxy=g.__proxy__.lambda_triple_apply_native(nativefn, mutated_fields))
        else:
            with cython_context():
                return SGraph(_proxy=g.__proxy__.lambda_triple_apply(triple_apply_fn, mutated_fields))