Support longest path

Author: FlorentinD

graphdatascience/plugin_v2_endpoints.py

@@ -35,6 +35,7 @@
 from graphdatascience.procedure_surface.api.node_embedding.hashgnn_endpoints import HashGNNEndpoints
 from graphdatascience.procedure_surface.api.node_embedding.node2vec_endpoints import Node2VecEndpoints
 from graphdatascience.procedure_surface.api.pathfinding.all_shortest_path_endpoints import AllShortestPathEndpoints
+from graphdatascience.procedure_surface.api.pathfinding.dag_endpoints import DagEndpoints
 from graphdatascience.procedure_surface.api.pathfinding.k_spanning_tree_endpoints import KSpanningTreeEndpoints
 from graphdatascience.procedure_surface.api.pathfinding.max_flow_endpoints import MaxFlowEndpoints
 from graphdatascience.procedure_surface.api.pathfinding.prize_steiner_tree_endpoints import PrizeSteinerTreeEndpoints
@@ -103,6 +104,7 @@
 from graphdatascience.procedure_surface.cypher.pathfinding.all_shortest_path_cypher_endpoints import (
     AllShortestPathCypherEndpoints,
 )
+from graphdatascience.procedure_surface.cypher.pathfinding.dag_cypher_endpoints import DagCypherEndpoints
 from graphdatascience.procedure_surface.cypher.pathfinding.k_spanning_tree_cypher_endpoints import (
     KSpanningTreeCypherEndpoints,
 )
@@ -217,6 +219,13 @@ def closeness_centrality(self) -> ClosenessEndpoints:
         """
         return ClosenessCypherEndpoints(self._db_client)
 
+    @property
+    def dag(self) -> DagEndpoints:
+        """
+        Return endpoints for Directed Acyclic Graph (DAG) algorithms.
+        """
+        return DagCypherEndpoints(self._db_client)
+
     @property
     def degree_centrality(self) -> DegreeEndpoints:
         """

graphdatascience/procedure_surface/api/pathfinding/dag_endpoints.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from graphdatascience.procedure_surface.api.pathfinding.longest_path_endpoints import LongestPathEndpoints
+
+
+class DagEndpoints(ABC):
+    """
+    Container for Directed Acyclic Graph (DAG) algorithm endpoints.
+    """
+
+    @property
+    @abstractmethod
+    def longest_path(self) -> LongestPathEndpoints:
+        """Access to Longest Path algorithm endpoints for DAGs."""
+        ...

graphdatascience/procedure_surface/api/pathfinding/longest_path_endpoints.py

@@ -0,0 +1,53 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+
+from pandas import DataFrame
+
+from graphdatascience.procedure_surface.api.catalog.graph_api import GraphV2
+from graphdatascience.procedure_surface.api.default_values import ALL_LABELS, ALL_TYPES
+
+
+class LongestPathEndpoints(ABC):
+    @abstractmethod
+    def stream(
+        self,
+        G: GraphV2,
+        relationship_weight_property: str | None = None,
+        relationship_types: list[str] = ALL_TYPES,
+        node_labels: list[str] = ALL_LABELS,
+        sudo: bool = False,
+        log_progress: bool = True,
+        username: str | None = None,
+        concurrency: int | None = None,
+        job_id: str | None = None,
+    ) -> DataFrame:
+        """
+        Runs the longest path algorithm in a DAG and returns the result as a DataFrame.
+
+        Parameters
+        ----------
+        G
+           Graph object to use
+        relationship_weight_property
+            Name of the property to be used as weights.
+        relationship_types
+            Filter the graph using the given relationship types. Relationships with any of the given types will be included.
+        node_labels
+            Filter the graph using the given node labels. Nodes with any of the given labels will be included.
+        sudo
+            Disable the memory guard.
+        log_progress
+            Display progress logging.
+        username
+            As an administrator, impersonate a different user for accessing their graphs.
+        concurrency
+            Number of concurrent threads to use.
+        job_id
+            Identifier for the computation.
+
+        Returns
+        -------
+        DataFrame
+            The longest path results as a DataFrame with columns for index, sourceNode, targetNode, totalCost, nodeIds, costs.
+        """

graphdatascience/procedure_surface/arrow/pathfinding/dag_arrow_endpoints.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from graphdatascience.arrow_client.authenticated_flight_client import AuthenticatedArrowClient
+from graphdatascience.procedure_surface.api.pathfinding.dag_endpoints import DagEndpoints
+from graphdatascience.procedure_surface.api.pathfinding.longest_path_endpoints import LongestPathEndpoints
+from graphdatascience.procedure_surface.arrow.pathfinding.longest_path_arrow_endpoints import (
+    LongestPathArrowEndpoints,
+)
+
+
+class DagArrowEndpoints(DagEndpoints):
+    def __init__(
+        self,
+        arrow_client: AuthenticatedArrowClient,
+        show_progress: bool = False,
+    ):
+        self._longest_path = LongestPathArrowEndpoints(arrow_client, show_progress)
+
+    @property
+    def longest_path(self) -> LongestPathEndpoints:
+        return self._longest_path

graphdatascience/procedure_surface/arrow/pathfinding/longest_path_arrow_endpoints.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from pandas import DataFrame
+
+from graphdatascience.arrow_client.authenticated_flight_client import AuthenticatedArrowClient
+from graphdatascience.procedure_surface.api.catalog.graph_api import GraphV2
+from graphdatascience.procedure_surface.api.default_values import ALL_LABELS, ALL_TYPES
+from graphdatascience.procedure_surface.api.pathfinding.longest_path_endpoints import LongestPathEndpoints
+from graphdatascience.procedure_surface.arrow.relationship_endpoints_helper import RelationshipEndpointsHelper
+from graphdatascience.procedure_surface.arrow.stream_result_mapper import map_shortest_path_stream_result
+
+
+class LongestPathArrowEndpoints(LongestPathEndpoints):
+    def __init__(
+        self,
+        arrow_client: AuthenticatedArrowClient,
+        show_progress: bool = False,
+    ):
+        self._endpoints_helper = RelationshipEndpointsHelper(
+            arrow_client, write_back_client=None, show_progress=show_progress
+        )
+
+    def stream(
+        self,
+        G: GraphV2,
+        relationship_weight_property: str | None = None,
+        relationship_types: list[str] = ALL_TYPES,
+        node_labels: list[str] = ALL_LABELS,
+        sudo: bool = False,
+        log_progress: bool = True,
+        username: str | None = None,
+        concurrency: int | None = None,
+        job_id: str | None = None,
+    ) -> DataFrame:
+        config = self._endpoints_helper.create_base_config(
+            G,
+            relationshipWeightProperty=relationship_weight_property,
+            relationshipTypes=relationship_types,
+            nodeLabels=node_labels,
+            sudo=sudo,
+            logProgress=log_progress,
+            username=username,
+            concurrency=concurrency,
+            jobId=job_id,
+        )
+
+        result = self._endpoints_helper.run_job_and_stream("v2/pathfinding.longestPath", G, config)
+        map_shortest_path_stream_result(result)
+
+        return result

graphdatascience/procedure_surface/cypher/pathfinding/dag_cypher_endpoints.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from graphdatascience.procedure_surface.api.pathfinding.dag_endpoints import DagEndpoints
+from graphdatascience.procedure_surface.api.pathfinding.longest_path_endpoints import LongestPathEndpoints
+from graphdatascience.procedure_surface.cypher.pathfinding.longest_path_cypher_endpoints import (
+    LongestPathCypherEndpoints,
+)
+from graphdatascience.query_runner.query_runner import QueryRunner
+
+
+class DagCypherEndpoints(DagEndpoints):
+    def __init__(self, query_runner: QueryRunner):
+        self._query_runner = query_runner
+        self._longest_path = LongestPathCypherEndpoints(query_runner)
+
+    @property
+    def longest_path(self) -> LongestPathEndpoints:
+        return self._longest_path

graphdatascience/procedure_surface/cypher/pathfinding/longest_path_cypher_endpoints.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from pandas import DataFrame
+
+from graphdatascience.call_parameters import CallParameters
+from graphdatascience.procedure_surface.api.catalog.graph_api import GraphV2
+from graphdatascience.procedure_surface.api.default_values import ALL_LABELS, ALL_TYPES
+from graphdatascience.procedure_surface.api.pathfinding.longest_path_endpoints import LongestPathEndpoints
+from graphdatascience.procedure_surface.utils.config_converter import ConfigConverter
+from graphdatascience.query_runner.query_runner import QueryRunner
+
+
+class LongestPathCypherEndpoints(LongestPathEndpoints):
+    def __init__(self, query_runner: QueryRunner):
+        self._query_runner = query_runner
+
+    def stream(
+        self,
+        G: GraphV2,
+        relationship_weight_property: str | None = None,
+        relationship_types: list[str] = ALL_TYPES,
+        node_labels: list[str] = ALL_LABELS,
+        sudo: bool = False,
+        log_progress: bool = True,
+        username: str | None = None,
+        concurrency: int | None = None,
+        job_id: str | None = None,
+    ) -> DataFrame:
+        config = ConfigConverter.convert_to_gds_config(
+            relationshipWeightProperty=relationship_weight_property,
+            relationshipTypes=relationship_types,
+            nodeLabels=node_labels,
+            sudo=sudo,
+            logProgress=log_progress,
+            username=username,
+            concurrency=concurrency,
+            jobId=job_id,
+        )
+        params = CallParameters(graph_name=G.name(), config=config)
+        params.ensure_job_id_in_config()
+
+        return self._query_runner.call_procedure(
+            "gds.dag.longestPath.stream",
+            params=params,
+            logging=log_progress,
+            yields=["index", "sourceNode", "targetNode", "totalCost", "nodeIds", "costs"],  # skip path column
+        )

graphdatascience/session/session_v2_endpoints.py

@@ -35,6 +35,7 @@
 from graphdatascience.procedure_surface.api.node_embedding.hashgnn_endpoints import HashGNNEndpoints
 from graphdatascience.procedure_surface.api.node_embedding.node2vec_endpoints import Node2VecEndpoints
 from graphdatascience.procedure_surface.api.pathfinding.all_shortest_path_endpoints import AllShortestPathEndpoints
+from graphdatascience.procedure_surface.api.pathfinding.dag_endpoints import DagEndpoints
 from graphdatascience.procedure_surface.api.pathfinding.k_spanning_tree_endpoints import KSpanningTreeEndpoints
 from graphdatascience.procedure_surface.api.pathfinding.max_flow_endpoints import MaxFlowEndpoints
 from graphdatascience.procedure_surface.api.pathfinding.prize_steiner_tree_endpoints import PrizeSteinerTreeEndpoints
@@ -102,6 +103,7 @@
 from graphdatascience.procedure_surface.arrow.pathfinding.all_shortest_path_arrow_endpoints import (
     AllShortestPathArrowEndpoints,
 )
+from graphdatascience.procedure_surface.arrow.pathfinding.dag_arrow_endpoints import DagArrowEndpoints
 from graphdatascience.procedure_surface.arrow.pathfinding.k_spanning_tree_arrow_endpoints import (
     KSpanningTreeArrowEndpoints,
 )
@@ -230,6 +232,13 @@ def closeness_centrality(self) -> ClosenessEndpoints:
         """
         return ClosenessArrowEndpoints(self._arrow_client, self._write_back_client, show_progress=self._show_progress)
 
+    @property
+    def dag(self) -> DagEndpoints:
+        """
+        Return endpoints for Directed Acyclic Graph (DAG) algorithms.
+        """
+        return DagArrowEndpoints(self._arrow_client, show_progress=self._show_progress)
+
     @property
     def degree_centrality(self) -> DegreeEndpoints:
         """

graphdatascience/tests/integrationV2/procedure_surface/arrow/pathfinding/test_longest_path_arrow_endpoints.py

@@ -0,0 +1,49 @@
+from typing import Generator
+
+import pytest
+
+from graphdatascience.arrow_client.authenticated_flight_client import AuthenticatedArrowClient
+from graphdatascience.procedure_surface.api.catalog.graph_api import GraphV2
+from graphdatascience.procedure_surface.arrow.pathfinding.longest_path_arrow_endpoints import LongestPathArrowEndpoints
+from graphdatascience.tests.integrationV2.procedure_surface.arrow.graph_creation_helper import create_graph
+
+# Create a DAG (Directed Acyclic Graph) for testing longest path
+dag_graph = """
+        CREATE
+            (a: Node {id: 0}),
+            (b: Node {id: 1}),
+            (c: Node {id: 2}),
+            (d: Node {id: 3}),
+            (e: Node {id: 4}),
+            (f: Node {id: 5}),
+            (a)-[:LINK {cost: 1.0}]->(b),
+            (a)-[:LINK {cost: 2.0}]->(c),
+            (b)-[:LINK {cost: 3.0}]->(d),
+            (c)-[:LINK {cost: 2.0}]->(d),
+            (d)-[:LINK {cost: 1.0}]->(e),
+            (c)-[:LINK {cost: 5.0}]->(f),
+            (f)-[:LINK {cost: 1.0}]->(e)
+        """
+
+
+@pytest.fixture
+def sample_dag(arrow_client: AuthenticatedArrowClient) -> Generator[GraphV2, None, None]:
+    with create_graph(arrow_client, "dag", dag_graph) as G:
+        yield G
+
+
+@pytest.fixture
+def longest_path_endpoints(
+    arrow_client: AuthenticatedArrowClient,
+) -> Generator[LongestPathArrowEndpoints, None, None]:
+    yield LongestPathArrowEndpoints(arrow_client)
+
+
+def test_longest_path_stream(longest_path_endpoints: LongestPathArrowEndpoints, sample_dag: GraphV2) -> None:
+    result_df = longest_path_endpoints.stream(
+        G=sample_dag,
+        relationship_weight_property="cost",
+    )
+
+    assert len(result_df) == 6
+    assert {"index", "sourceNode", "targetNode", "totalCost", "nodeIds", "costs"} == set(result_df.columns)

graphdatascience/tests/integrationV2/procedure_surface/cypher/pathfinding/test_longest_path_cypher_endpoints.py

@@ -0,0 +1,59 @@
+from typing import Generator
+
+import pytest
+
+from graphdatascience.procedure_surface.api.catalog.graph_api import GraphV2
+from graphdatascience.procedure_surface.cypher.pathfinding.longest_path_cypher_endpoints import (
+    LongestPathCypherEndpoints,
+)
+from graphdatascience.query_runner.query_runner import QueryRunner
+from graphdatascience.tests.integrationV2.procedure_surface.cypher.cypher_graph_helper import create_graph
+
+# Create a DAG (Directed Acyclic Graph) for testing longest path
+dag_graph = """
+        CREATE
+            (a: Node {id: 0}),
+            (b: Node {id: 1}),
+            (c: Node {id: 2}),
+            (d: Node {id: 3}),
+            (e: Node {id: 4}),
+            (f: Node {id: 5}),
+            (a)-[:LINK {cost: 1.0}]->(b),
+            (a)-[:LINK {cost: 2.0}]->(c),
+            (b)-[:LINK {cost: 3.0}]->(d),
+            (c)-[:LINK {cost: 2.0}]->(d),
+            (d)-[:LINK {cost: 1.0}]->(e),
+            (c)-[:LINK {cost: 5.0}]->(f),
+            (f)-[:LINK {cost: 1.0}]->(e)
+        """
+
+
+@pytest.fixture
+def sample_dag(query_runner: QueryRunner) -> Generator[GraphV2, None, None]:
+    projection_query = """
+        MATCH (source)-[r]->(target)
+        WITH gds.graph.project('dag', source, target, {
+            sourceNodeProperties: properties(source),
+            targetNodeProperties: properties(target),
+            relationshipProperties: properties(r)
+        }) AS G
+        RETURN G
+    """
+
+    with create_graph(query_runner, "dag", dag_graph, projection_query) as G:
+        yield G
+
+
+@pytest.fixture
+def longest_path_endpoints(query_runner: QueryRunner) -> Generator[LongestPathCypherEndpoints, None, None]:
+    yield LongestPathCypherEndpoints(query_runner)
+
+
+def test_longest_path_stream(longest_path_endpoints: LongestPathCypherEndpoints, sample_dag: GraphV2) -> None:
+    result_df = longest_path_endpoints.stream(
+        G=sample_dag,
+        relationship_weight_property="cost",
+    )
+
+    assert len(result_df) == 6
+    assert {"index", "sourceNode", "targetNode", "totalCost", "nodeIds", "costs"} == set(result_df.columns)