SSSP results are inconsistent with expectations.

[MikuSugar]

I am trying to implement the PIE algorithm in Python, following this document(https://nbviewer.org/github/alibaba/GraphScope/blob/main/tutorials/06_writing_your_own_algorithms.ipynb). However, all nodes in the resulting file have the initial value of 1000000000.0, including the starting node. Below is my Python code:

import time

import graphscope
from graphscope.dataset import load_p2p_network
from graphscope.framework.context import VertexDataContextDAGNode

# assume we mount `~/test_data` to `/testingdata` in pods, and `ogbn_mag_small` within.
k8s_volumes = {
    "data": {
        "type": "hostPath",
        "field": {
            "path": "/host/test_data/",
            "type": "Directory"
        },
        "mounts": {
            "mountPath": "/testingdata"
        }
    }
}

from graphscope.analytical.udf.decorators import pie
from graphscope.framework.app import AppAssets


@pie(vd_type="double", md_type="double")
class SSSP_PIE(AppAssets):
    @staticmethod
    def Init(frag, context):
        v_label_num = frag.vertex_label_num()
        for v_label_id in range(v_label_num):
            nodes = frag.nodes(v_label_id)
            context.init_value(
                nodes, v_label_id, 1000000000.0, PIEAggregateType.kMinAggregate
            )
            context.register_sync_buffer(v_label_id, MessageStrategy.kSyncOnOuterVertex)

    @staticmethod
    def PEval(frag, context):
        src = int(context.get_config(b"src"))
        graphscope.declare(graphscope.Vertex, source)
        native_source = False
        v_label_num = frag.vertex_label_num()
        for v_label_id in range(v_label_num):
            if frag.get_inner_node(v_label_id, src, source):
                native_source = True
                break
        if native_source:
            context.set_node_value(source, 0)
        else:
            return
        e_label_num = frag.edge_label_num()
        for e_label_id in range(e_label_num):
            edges = frag.get_outgoing_edges(source, e_label_id)
            for e in edges:
                dst = e.neighbor()
                distv = e.get_int(2)
                if context.get_node_value(dst) > distv:
                    context.set_node_value(dst, distv)

    @staticmethod
    def IncEval(frag, context):
        v_label_num = frag.vertex_label_num()
        e_label_num = frag.edge_label_num()
        for v_label_id in range(v_label_num):
            iv = frag.inner_nodes(v_label_id)
            for v in iv:
                v_dist = context.get_node_value(v)
                for e_label_id in range(e_label_num):
                    es = frag.get_outgoing_edges(v, e_label_id)
                    for e in es:
                        u = e.neighbor()
                        u_dist = v_dist + e.get_int(2)
                        if context.get_node_value(u) > u_dist:
                            context.set_node_value(u, u_dist)


if __name__ == '__main__':
    graphscope.set_option(show_log=True)
    sess = graphscope.session(k8s_volumes=k8s_volumes,
                              num_workers=1,
                              k8s_engine_cpu=8,
                              k8s_engine_mem="8Gi",
                              vineyard_shared_mem="3Gi", )
    #
    g = load_p2p_network(sess, "/testingdata/p2p_network/")

    # g = graph
    print(g.schema)

    my_app2 = SSSP_PIE()
    ret2: graphscope.framework.context.Context = my_app2(g, src=6)
    ret2.output("file:////testingdata/res/pie" + str(int(time.time())),
                {"node": "v:host.id", "result": "r:host"})

[MikuSugar]

I found that the 'dist' attribute is the third one by observing the g.schema. I changed the code from e.get_int(2) to e.get_int(3), but the result is still inconsistent with my expectation.

oid_type: LONG
vid_type: ULONG
type: VERTEX
Label: host
Properties: Property(0, weight, LONG), Property(1, id, LONG)

type: EDGE
Label: connect
Properties: Property(0, eid, LONG), Property(1, src_label_id, LONG), Property(2, dst_label_id, LONG), Property(3, dist, LONG)
Relations: [Relation(source='host', destination='host')]

[sighingnow]

@MikuSugar can you share some example data and expected outputs (as minimal as possible) that can reproduce the issue?

[MikuSugar]

@sighingnow
The data is the load_p2p_network shown in the sample diagram.
The expected result is consistent with calling the built-in sssp algorithm.

pg2 = g.project(vertices={"host": ["id"]}, edges={"connect": ["dist"]})
res: VertexDataContextDAGNode = graphscope.sssp(pg2, src=6)
res.output("file:////testingdata/res/native" + str(int(time.time())), {"node": "v.id", "result": "r"})

My starting point is 6, and the result for at least this point should be 0, but it is also 1000000000.0. I think there is a problem with the PEval function.

[MikuSugar]

Data files: https://github.com/MikuSugar/data/blob/master/p2p_network.tar.gz

[MikuSugar]

My version is 0.21.0.

[MikuSugar]

I tried using version 0.20.0, but the result is the same. I don't know where the problem is. 😣

[sighingnow]

@lidongze0629 can you take a look on this issue?

[lidongze0629]

@MikuSugar Thanks for reporting this problem. I will take a look.

[MikuSugar]

@lidongze0629 Thanks 🙏

[lidongze0629]

@MikuSugar This problem caused by missing the generate_eid=False params in load_p2p_network function.

generate_eid: (bool, optional): Generate id for each edge as a property when set to True. Defaults to True.

>>> from graphscope.dataset import load_p2p_network
>>> g = load_p2p_network()
>>> # A new column named 'eid' is generated as the first property, and 'e.get_int(2)' in 'IncEval' get a wrong index of 'dist' property
>>> print(g.schema)

type: EDGE
Label: connect
Properties: Property(0, eid, LONG), Property(1, src_label_id, LONG), Property(2, dst_label_id, LONG), Property(3, dist, LONG)
Relations: [Relation(source='host', destination='host')]

Here is the solution for you now within v0.21.0 version.

1. Just change your SSSP code to get the correct column index of 'dist' property

- u_dist = v_dist + e.get_int(2)
+ u_dist = v_dist + e.get_int(3) 

@pie(vd_type="double", md_type="double")
class SSSP_PIE(AppAssets):
    @staticmethod
    def Init(frag, context):
        v_label_num = frag.vertex_label_num()
        for v_label_id in range(v_label_num):
            nodes = frag.nodes(v_label_id)
            context.init_value(
                nodes, v_label_id, 1000000000.0, PIEAggregateType.kMinAggregate
            )
            context.register_sync_buffer(v_label_id, MessageStrategy.kSyncOnOuterVertex)

    @staticmethod
    def PEval(frag, context):
        src = context.get_config(b"src")
        graphscope.declare(graphscope.Vertex, source)
        native_source = False
        v_label_num = frag.vertex_label_num()
        for v_label_id in range(v_label_num):
            if frag.get_inner_node(v_label_id, src, source):
                native_source = True
                break
        if native_source:
            context.set_node_value(source, 0)
        else:
            return
        e_label_num = frag.edge_label_num()
        for e_label_id in range(e_label_num):
            edges = frag.get_outgoing_edges(source, e_label_id)
            for e in edges:
                dst = e.neighbor()
                distv = e.get_int(3)
                if context.get_node_value(dst) > distv:
                    context.set_node_value(dst, distv)

    @staticmethod
    def IncEval(frag, context):
        print(context.superstep(), flush=True)
        v_label_num = frag.vertex_label_num()
        e_label_num = frag.edge_label_num()
        for v_label_id in range(v_label_num):
            iv = frag.inner_nodes(v_label_id)
            for v in iv:
                v_dist = context.get_node_value(v)
                for e_label_id in range(e_label_num):
                    es = frag.get_outgoing_edges(v, e_label_id)
                    for e in es:
                        u = e.neighbor()
                        u_dist = v_dist + e.get_int(3)
                        if context.get_node_value(u) > u_dist:
                            context.set_node_value(u, u_dist)

[MikuSugar]

@lidongze0629 I have previously tried to change e.get_int(2), but it didn't work #2735 (comment),
#2735 (reply in thread)

[lidongze0629]

@MikuSugar

Here is my test code:

import graphscope
from graphscope.dataset import load_p2p_network

from graphscope.analytical.udf.decorators import pie
from graphscope.framework.app import AppAssets

@pie(vd_type="double", md_type="double")
class SSSP_PIE(AppAssets):
    @staticmethod
    def Init(frag, context):
        v_label_num = frag.vertex_label_num()
        for v_label_id in range(v_label_num):
            nodes = frag.nodes(v_label_id)
            context.init_value(
                nodes, v_label_id, 1000000000.0, PIEAggregateType.kMinAggregate
            )
            context.register_sync_buffer(v_label_id, MessageStrategy.kSyncOnOuterVertex)

    @staticmethod
    def PEval(frag, context):
        src = context.get_config(b"src")
        graphscope.declare(graphscope.Vertex, source)
        native_source = False
        v_label_num = frag.vertex_label_num()
        for v_label_id in range(v_label_num):
            if frag.get_inner_node(v_label_id, src, source):
                native_source = True
                break
        if native_source:
            context.set_node_value(source, 0)
        else:
            return
        e_label_num = frag.edge_label_num()
        for e_label_id in range(e_label_num):
            edges = frag.get_outgoing_edges(source, e_label_id)
            for e in edges:
                dst = e.neighbor()
                distv = e.get_int(3)
                if context.get_node_value(dst) > distv:
                    context.set_node_value(dst, distv)

    @staticmethod
    def IncEval(frag, context):
        print(context.superstep(), flush=True)
        v_label_num = frag.vertex_label_num()
        e_label_num = frag.edge_label_num()
        for v_label_id in range(v_label_num):
            iv = frag.inner_nodes(v_label_id)
            for v in iv:
                v_dist = context.get_node_value(v)
                for e_label_id in range(e_label_num):
                    es = frag.get_outgoing_edges(v, e_label_id)
                    for e in es:
                        u = e.neighbor()
                        u_dist = v_dist + e.get_int(3)
                        if context.get_node_value(u) > u_dist:
                            context.set_node_value(u, u_dist)


if __name__ == '__main__':
    sess = graphscope.session(cluster_type="hosts")
    graphscope.set_option(show_log=True)
    g = load_p2p_network(sess, directed=True)

    my_app2 = SSSP_PIE()
    ret2 = my_app2(g, src=6)
    r = ret2.to_dataframe({"node": "v:host.id", "result": "r:host"}).sort_values(by=["node"])
    print(r)

[lidongze0629]

After a further test, the result is incorrect if the session param num_workers=1, other values(I test =2, =4) ​​are correct.

[MikuSugar]

After a further test, the result is incorrect if the session param num_workers=1, other values(I test =2, =4) ​​are correct.

@lidongze0629 Thank you. Why doesn't num_workers=1 work? Is it because PIE does not support single node or is there a bug? 🤔️

[lidongze0629]

I think maybe something wrong with it, and I will further take a look at this.

[lidongze0629]

alibaba/libgrape-lite#137

Fixes the num_workers=1 problem in alibaba/libgrape-lite#137. It will work in the next release.

[lidongze0629]

Update the docs: #2762