Maximum Flow¶

Algorithm introduction¶

Maximum flow computes the largest feasible flow value from source to sink under edge capacity constraints.

Compute the maximum flow in a directed graph whose node primary key is id and whose edge capacity property is weight. Parallel directed edges are aggregated by (src_id, dst_id) before each residual search.

For more information, see Maximum Flow.

Data model¶

To implement the Maximum Flow algorithm in GQL, the following data model is used to create a graph type:

Entities are represented by nodes of type Node, with properties id and rating.
Relationships are represented by edges of type Link, with property weight.

For the complete definition of the graph type, see Prepare graph data.

Procedure implementation ¶

To compute Maximum Flow results, a procedure is created.

For more information about the procedure, see Procedure implementation. The following section describes the key components of the procedure and how it works.

The procedure takes the following input parameters:

source_id: INT64. ID of the flow source node.
sink_id: INT64. ID of the flow sink node.

Additionally, the following variables are defined within the procedure:

Variable name	Variable type	Description
`frontier`	Primitive value variable	Active set for BFS augmenting-path search.
`has_path`	Primitive value variable	Whether an augmenting path to the sink was found.
`inf`	Primitive value variable	Infinity sentinel for unbounded capacity.
`total_flow`	Primitive value variable	Total accumulated flow from source to sink.
`augment_count`	Primitive value variable	Number of augmenting path iterations completed.
`augment_cap`	Primitive value variable	Bottleneck capacity of the current augmenting path.
`walk_id`	Primitive value variable	Node ID used during augmenting path reconstruction.
`prev_id`	Primitive value variable	Previous node in the augmenting path during reconstruction.
`edge_dir`	Primitive value variable	Direction flag (+1 forward, -1 backward) for flow update.
`bfs_active`	Primitive value variable	Whether BFS expanded at least one new node in this step.
`sink_found`	Primitive value variable	Whether the sink node was reached in this BFS pass.
`sink_state`	Global SumAgg aggregator value variable	Best path state recorded at the sink node.
`back_state`	Global SumAgg aggregator value variable	State used during augmenting path reconstruction.
`out_cap`	Node-attached SumAgg aggregator value variable	Per-node map of {neighbor_id → remaining capacity}.
`out_flow`	Node-attached SumAgg aggregator value variable	Per-node map of {neighbor_id → current flow}.
`seen`	Node-attached SumAgg aggregator value variable	Per-node flag: node was visited in the current BFS pass.
`cand`	Node-attached SumAgg aggregator value variable	Per-node best path candidate from the current BFS expansion.
`state`	Node-attached SumAgg aggregator value variable	Per-node current best augmenting path state.

The following steps outline the logic of the procedure:

Initialize residual capacities out_cap from edge weights and flows out_flow to 0.
Repeatedly run BFS from source to find an augmenting path to the sink (sink_found); track path state in cand and state.
Walk back from sink to source, compute the bottleneck capacity (augment_cap), and update out_cap and out_flow along the path.
Accumulate augment_cap into total_flow; stop when no augmenting path exists. Return the total maximum flow.

Steps¶

Run the following code blocks in order in the NebulaGraph Studio CONSOLE. For first-time setup, follow Prepare algorithm sample data to check where to run commands and how the directory is mounted.

1. Prepare graph data ¶

Create a graph type that matches your data model.

'''
CREATE GRAPH TYPE IF NOT EXISTS maximum_flow_graph_type AS {
    NODE Node (LABEL Node {id INT PRIMARY KEY, rating INT}),
    EDGE Link (Node)-[LABEL Link {weight INT, MULTIEDGE KEY()}]->(Node)
}
'''

Download these CSV files and save them with their original names in examples/algorithms/maximum_flow/ under your deployment directory:

maximum_flow_node.csv: Node data.
maximum_flow_edge.csv: Edge data.

Preview the CSV files

maximum_flow_node.csv

id,rating
1,0
2,2
3,1
4,0
5,0
6,2
7,1
8,2
9,0
10,1
11,0
12,2
13,1
14,1
15,0
16,2
17,0
18,1
19,2
20,0
21,2
22,1
23,0
24,1

maximum_flow_edge.csv

src_id,dst_id,weight
1,2,4
1,3,2
1,4,7
1,5,9
2,3,1
2,4,6
2,5,5
2,6,2
3,4,3
3,5,4
3,6,8
4,6,2
4,7,6
4,8,5
5,6,4
5,8,7
5,9,6
6,8,1
6,9,5
6,10,6
7,9,2
7,10,9
8,9,3
8,11,6
8,12,4
9,10,4
9,12,7
9,13,2
10,12,1
10,13,8
10,14,4
11,12,2
11,14,5
11,15,7
12,14,3
12,15,6
12,16,5
13,15,2
13,16,7
14,15,4
14,17,8
14,18,2
15,16,1
15,18,5
15,19,3
16,18,3
16,19,6
17,18,2
17,20,7
17,21,5
18,19,4
18,21,9
18,22,6
19,21,1
19,22,5
20,21,3
20,23,6
20,24,8
21,22,2
21,23,4
21,24,7
22,24,4
23,24,1

Create a temporary graph based on the graph type and load the data files into the temporary graph.

CREATE TEMPORARY GRAPH IF NOT EXISTS #maximum_flow_graph TYPED maximum_flow_graph_type

'''
USE #maximum_flow_graph IMPORT INTO GRAPH {
      NODE (v@Node {id: id, rating: rating}) FROM DATAFILE {
          PATH: "file:///examples/algorithms/maximum_flow/maximum_flow_node.csv",
          FORMAT: "CSV",
          DELIMITER: ','
      },
      EDGE (id: src_id)-[e@Link {weight: weight}]->(id: dst_id) FROM DATAFILE {
          PATH: "file:///examples/algorithms/maximum_flow/maximum_flow_edge.csv",
          FORMAT: "CSV",
          DELIMITER: ','
      }
  }
'''

After the import succeeds, run this query in the console to check the node count:

'''
USE #maximum_flow_graph
MATCH (n)
RETURN count(n) AS node_count
'''

The expected node_count is 24. If it differs, check the files and import results before continuing.

2. Create a procedure ¶

Create a procedure to implement the Maximum Flow algorithm. For more information about the procedure, see Procedure implementation.

'''
CREATE OR REPLACE PROCEDURE maximum_flow(source_id INT64, sink_id INT64)
RETURNS (source_id INT64, sink_id INT64, max_flow DOUBLE, augmentation_count INT64)
COMMENT "Compute the maximum flow in a directed graph whose vertex primary key is id and whose edge capacity property is weight. Parallel directed edges are aggregated by (src_id, dst_id) before each residual search."
AS {
  VALUE frontier ACTIVE_SET

  VALUE has_path BOOLEAN = true
  VALUE inf DOUBLE = 1000000000000000000.0
  VALUE total_flow DOUBLE = 0.0
  VALUE augment_count INT64 = 0
  VALUE augment_cap DOUBLE = 0.0
  VALUE walk_id INT64 = 0
  VALUE prev_id INT64 = 0
  VALUE edge_dir INT64 = 0

  VALUE bfs_active OrAgg = false
  VALUE sink_found OrAgg = false
  VALUE sink_state TopKAgg<1, path_cap DOUBLE DESC, parent_id INT64 ASC, dir INT64 DESC>
  VALUE back_state TopKAgg<1, path_cap DOUBLE DESC, parent_id INT64 ASC, dir INT64 DESC>

  NODE VALUE out_cap MapAgg<INT64, SumAgg<DOUBLE>>
  NODE VALUE out_flow MapAgg<INT64, SumAgg<DOUBLE>>
  NODE VALUE seen OrAgg = false
  NODE VALUE cand TopKAgg<1, path_cap DOUBLE DESC, parent_id INT64 ASC, dir INT64 DESC>
  NODE VALUE state TopKAgg<1, path_cap DOUBLE DESC, parent_id INT64 ASC, dir INT64 DESC>

  MATCH (s)-[e]->(t)
  PER PATH {
    SET s.@out_cap += TUPLE(t.id, CAST(e.weight AS DOUBLE))
    SET s.@out_flow += TUPLE(t.id, 0.0)
  }

  IF source_id = sink_id THEN {
    SET has_path = false
  }

  WHILE has_path = true THEN {
    MATCH (n)
    PER NODE (n) {
      SET n.@seen = false
      SET n.@cand.clear()
      SET n.@state.clear()
    }

    MATCH (s)
    WHERE s.id = source_id
    PER NODE (s) {
      SET s.@seen = true
      SET s.@state += RECORD{path_cap: inf, parent_id: source_id, dir: 1}
    }
    FINALLY {
      SET frontier = s
    }

    SET @bfs_active = true
    SET @sink_found = false

    WHILE @bfs_active = true AND @sink_found = false THEN {
      SET @bfs_active = false

      MATCH (n)
      WHERE n.@seen = false
      PER NODE (n) {
        SET n.@cand.clear()
      }

      MATCH (u)-[e]->(v)
      WHERE u IN frontier
        AND v.@seen = false
        AND u.@out_cap.get(v.id) - u.@out_flow.get(v.id) > 0.0
      PER PATH {
        VALUE residual DOUBLE = 0.0
        VALUE candidate_cap DOUBLE = 0.0

        SET residual = u.@out_cap.get(v.id) - u.@out_flow.get(v.id)

        IF u.@state.min().path_cap < residual THEN {
          SET candidate_cap = u.@state.min().path_cap
        }
        ELSE {
          SET candidate_cap = residual
        }

        SET v.@cand += RECORD{path_cap: candidate_cap, parent_id: u.id, dir: 1}
      }

      MATCH (u)<-[e]-(p)
      WHERE u IN frontier
        AND p.@seen = false
        AND p.@out_flow.get(u.id) > 0.0
      PER PATH {
        VALUE residual DOUBLE = 0.0
        VALUE candidate_cap DOUBLE = 0.0

        SET residual = p.@out_flow.get(u.id)

        IF u.@state.min().path_cap < residual THEN {
          SET candidate_cap = u.@state.min().path_cap
        }
        ELSE {
          SET candidate_cap = residual
        }

        SET p.@cand += RECORD{path_cap: candidate_cap, parent_id: u.id, dir: -1}
      }

      MATCH (n)
      WHERE n.@seen = false AND n.@cand.size() > 0
      PER NODE (n) {
        SET n.@seen = true
        SET n.@state += n.@cand.min()
        SET @bfs_active += true

        IF n.id = sink_id THEN {
          SET @sink_found += true
        }
      }
      FINALLY {
        SET frontier = n
      }
    }

    IF @sink_found = false THEN {
      SET has_path = false
    }
    ELSE {
      SET @sink_state.clear()

      MATCH (n)
      WHERE n.id = sink_id
      PER NODE (n) {
        SET @sink_state += n.@state.min()
      }

      SET augment_cap = @sink_state.min().path_cap
      SET total_flow = total_flow + augment_cap
      SET augment_count = augment_count + 1
      SET walk_id = sink_id

      WHILE walk_id <> source_id THEN {
        SET @back_state.clear()

        MATCH (n)
        WHERE n.id = walk_id
        PER NODE (n) {
          SET @back_state += n.@state.min()
        }

        SET prev_id = @back_state.min().parent_id
        SET edge_dir = @back_state.min().dir

        IF edge_dir = 1 THEN {
          MATCH (p)
          WHERE p.id = prev_id
          PER NODE (p) {
            SET p.@out_flow += TUPLE(walk_id, augment_cap)
          }
        }
        ELSE {
          MATCH (p)
          WHERE p.id = walk_id
          PER NODE (p) {
            SET p.@out_flow += TUPLE(prev_id, -augment_cap)
          }
        }

        SET walk_id = prev_id
      }
    }
  }

  RETURN
    source_id AS source_id,
    sink_id AS sink_id,
    total_flow AS max_flow,
    augment_count AS augmentation_count
}
'''

3. Call your procedure ¶

Use node 1 as the source and node 24 as the sink to compute the maximum flow and augmentation count.

USE #maximum_flow_graph CALL maximum_flow(1, 24) RETURN source_id, sink_id, max_flow, augmentation_count

Returns:

+-----------+---------+----------+--------------------+
| source_id | sink_id | max_flow | augmentation_count |
+-----------+---------+----------+--------------------+
| 1         | 24      | 19.0     | 6                  |
+-----------+---------+----------+--------------------+

Save results to a file or graph

To save the returned results as CSV files or write them to a graph, see EXPORT. Match the exported fields to the returned columns and target properties.

Maximum Flow¶

Algorithm introduction¶

Data model¶

Procedure implementation¶

Steps¶

1. Prepare graph data¶

2. Create a procedure¶

3. Call your procedure¶

Procedure implementation ¶

1. Prepare graph data ¶

2. Create a procedure ¶

3. Call your procedure ¶