Source: https://dev-docs.nebula-graph.io/v5.3.2/analytic/algorithms/maximum_flow/
Documentation version: v5.3.2

# 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](https://en.wikipedia.org/wiki/Maximum_flow_problem).

## 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](https://dev-docs.nebula-graph.io/v5.3.2/analytic/algorithms/maximum_flow/#1_prepare_graph_data).

## Procedure implementation[]()

To compute Maximum Flow results, a [procedure](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/procedures/create/) is created.

For more information about the procedure, see [Procedure implementation](https://dev-docs.nebula-graph.io/v5.3.2/analytic/algorithms/maximum_flow/#procedure_implement). 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](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Active set for BFS augmenting-path search.
 | `has_path` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Whether an augmenting path to the sink was found.
 | `inf` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Infinity sentinel for unbounded capacity.
 | `total_flow` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Total accumulated flow from source to sink.
 | `augment_count` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Number of augmenting path iterations completed.
 | `augment_cap` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Bottleneck capacity of the current augmenting path.
 | `walk_id` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Node ID used during augmenting path reconstruction.
 | `prev_id` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Previous node in the augmenting path during reconstruction.
 | `edge_dir` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Direction flag (+1 forward, -1 backward) for flow update.
 | `bfs_active` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Whether BFS expanded at least one new node in this step.
 | `sink_found` | [Primitive value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/value-variable/) | Whether the sink node was reached in this BFS pass.
 | `sink_state` | Global [SumAgg aggregator value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/agg/sum_agg/) | Best path state recorded at the sink node.
 | `back_state` | Global [SumAgg aggregator value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/agg/sum_agg/) | State used during augmenting path reconstruction.
 | `out_cap` | Node-attached [SumAgg aggregator value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/agg/sum_agg/) | Per-node map of {neighbor_id → remaining capacity}.
 | `out_flow` | Node-attached [SumAgg aggregator value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/agg/sum_agg/) | Per-node map of {neighbor_id → current flow}.
 | `seen` | Node-attached [SumAgg aggregator value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/agg/sum_agg/) | Per-node flag: node was visited in the current BFS pass.
 | `cand` | Node-attached [SumAgg aggregator value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/agg/sum_agg/) | Per-node best path candidate from the current BFS expansion.
 | `state` | Node-attached [SumAgg aggregator value variable](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/variable-definition/agg/sum_agg/) | Per-node current best augmenting path state.

The following steps outline the logic of the procedure:

1. Initialize residual capacities `out_cap` from edge weights and flows `out_flow` to 0.
2. Repeatedly run BFS from source to find an augmenting path to the sink (`sink_found`); track path state in `cand` and `state`.
3. Walk back from sink to source, compute the bottleneck capacity (`augment_cap`), and update `out_cap` and `out_flow` along the path.
4. 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](https://dev-docs.nebula-graph.io/v5.3.2/analytic/algorithms/prepare-data/) to check where to run commands and how the directory is mounted.

### 1. Prepare graph data[]()

1. 

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)
}
'''

````

2. 

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

  - [maximum_flow_node.csv](https://dev-docs.nebula-graph.io/v5.3.2/analytic/algorithms/files/maximum_flow/maximum_flow_node.csv): Node data.
  - [maximum_flow_edge.csv](https://dev-docs.nebula-graph.io/v5.3.2/analytic/algorithms/files/maximum_flow/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

````

3. 

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](https://dev-docs.nebula-graph.io/v5.3.2/analytic/algorithms/prepare-data/#troubleshooting) before continuing.

### 2. Create a procedure[]()

Create a procedure to implement the Maximum Flow algorithm. For more information about the procedure, see [Procedure implementation](https://dev-docs.nebula-graph.io/v5.3.2/analytic/algorithms/maximum_flow/#procedure_implement). 

````
'''
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](https://dev-docs.nebula-graph.io/v5.3.2/database-gql-reference/in-out/export/). Match the exported fields to the returned columns and target properties.