Procedural BFS Examples¶

This page shows queries with variable-length paths (VLP) transpiled using the procedural BFS rendering mode.

Procedural BFS uses SQL scripting (BEGIN...END, WHILE) instead of WITH RECURSIVE CTEs. This enables a global visited set that prevents re-visiting nodes (shortest-path semantics).

Two materialization strategies are shown:

Databricks (temp_tables): Uses CREATE TEMPORARY TABLE + INSERT INTO. Fixed table names, O(1) visited reads per level.
PySpark 4.2 (numbered_views): Uses EXECUTE IMMEDIATE + numbered views. Dynamic names, UNION chain for visited.

Quick Start with GraphContext¶

The simplest way to use procedural BFS is via GraphContext:

PySpark: enable SQL scripting

Procedural BFS generates BEGIN...END blocks with DECLARE, WHILE, etc. PySpark requires SQL scripting to be enabled:

Python

spark.conf.set("spark.sql.scripting.enabled", "true")

Databricks has SQL scripting enabled by default.

Python

from gsql2rsql import GraphContext

graph = GraphContext(
    spark=spark,
    nodes_table="catalog.schema.nodes",
    edges_table="catalog.schema.edges",
)

# Procedural BFS for Databricks (default)
sql = graph.transpile(
    """
    MATCH (root {node_id: 'Alice'})
    MATCH p = (root)-[*1..4]-()
    UNWIND relationships(p) AS r
    RETURN r
    """,
    vlp_rendering_mode="procedural",
    materialization_strategy="temp_tables",
)

# Procedural BFS for PySpark 4.2+
sql_pyspark = graph.transpile(
    """
    MATCH (root {node_id: 'Alice'})
    MATCH p = (root)-[*1..4]-()
    UNWIND relationships(p) AS r
    RETURN r
    """,
    vlp_rendering_mode="procedural",
    materialization_strategy="numbered_views",
)

Edge collection support

Procedural BFS supports UNWIND relationships(path) AS r to access edge properties. Each BFS result row is one edge, so UNWIND produces one row per traversed edge. nodes(path) is not supported (requires full path reconstruction). Use vlp_rendering_mode='cte' if you need it.

1. Variable-length path traversal (1 to 3 hops)¶

Source: Features

OpenCypher Query

Cypher

MATCH (p:Person)-[:KNOWS*1..3]->(f:Person)
WHERE p.name = 'Alice'
RETURN DISTINCT f.name AS reachable

Procedural BFS — Databricks (temp_tables)

SQL

BEGIN
  DECLARE current_depth_1 INT DEFAULT 0;
  DECLARE rows_in_frontier_1 BIGINT DEFAULT 1;

  DROP TEMPORARY TABLE IF EXISTS bfs_visited_1;
  DROP TEMPORARY TABLE IF EXISTS bfs_frontier_1;
  DROP TEMPORARY TABLE IF EXISTS bfs_result_1;
  DROP TEMPORARY TABLE IF EXISTS bfs_frontier_1_init;
  DROP TEMPORARY TABLE IF EXISTS bfs_edges_keyed_1;
  CREATE TEMPORARY TABLE bfs_visited_1 (node STRING);
  CREATE TEMPORARY TABLE bfs_frontier_1 AS
  SELECT n.id AS node
  FROM catalog.demo.Person n
  WHERE (n.name) = ('Alice');
  INSERT INTO bfs_visited_1
  SELECT node FROM bfs_frontier_1;
  CREATE TEMPORARY TABLE bfs_result_1 (_row_id BIGINT, _next_node STRING, _bfs_depth INT);
  CREATE TEMPORARY TABLE bfs_frontier_1_init AS
  SELECT node FROM bfs_frontier_1;
  SELECT IF(COUNT(*) > 1, RAISE_ERROR('gsql2rsql procedural BFS is single-source but the start filter matched multiple nodes. Use vlp_rendering_mode="cte" for multi-source traversals.'), TRUE) FROM bfs_frontier_1_init;
  CREATE TEMPORARY TABLE bfs_edges_keyed_1 AS
  SELECT e.*, MONOTONICALLY_INCREASING_ID() AS _row_id
  FROM (
  SELECT e.person_id, e.friend_id, e.since, e.strength
  FROM catalog.demo.Knows e
  ) e;

  WHILE rows_in_frontier_1 > 0 AND current_depth_1 < 3 DO
    SET current_depth_1 = current_depth_1 + 1;

    DROP TEMPORARY TABLE IF EXISTS bfs_edges_1;
    CREATE TEMPORARY TABLE bfs_edges_1 AS
    SELECT e._row_id, e.friend_id AS _next_node
  FROM bfs_edges_keyed_1 e
  INNER JOIN bfs_frontier_1 f ON e.person_id = f.node
  WHERE e.friend_id IS NOT NULL AND NOT EXISTS (SELECT 1 FROM bfs_visited_1 v WHERE v.node = e.friend_id);

    SET rows_in_frontier_1 = (SELECT COUNT(1) FROM bfs_edges_1);

    IF rows_in_frontier_1 > 0 THEN
      INSERT INTO bfs_visited_1
      SELECT DISTINCT _next_node FROM bfs_edges_1;
      DROP TEMPORARY TABLE bfs_frontier_1;
      CREATE TEMPORARY TABLE bfs_frontier_1 AS
      SELECT DISTINCT _next_node AS node FROM bfs_edges_1;
      INSERT INTO bfs_result_1
      SELECT *, current_depth_1 AS _bfs_depth FROM bfs_edges_1;
    END IF;
  END WHILE;

  CREATE OR REPLACE TEMPORARY VIEW paths_1 AS
  SELECT f0.node AS start_node, r._next_node AS end_node, r._bfs_depth AS depth,
         e.person_id, e.friend_id, e.since, e.strength
  FROM bfs_result_1 r
  JOIN bfs_edges_keyed_1 e ON e._row_id = r._row_id
  CROSS JOIN bfs_frontier_1_init f0;

  SELECT 
   FIRST(_gsql2rsql_f_name) AS reachable
FROM (
  SELECT
     sink.id AS _gsql2rsql_f_id
    ,sink.name AS _gsql2rsql_f_name
    ,sink.age AS _gsql2rsql_f_age
    ,sink.nickname AS _gsql2rsql_f_nickname
    ,sink.salary AS _gsql2rsql_f_salary
    ,sink.active AS _gsql2rsql_f_active
    ,p.start_node
    ,p.end_node
    ,p.depth
  FROM paths_1 p
  JOIN catalog.demo.Person sink
    ON sink.id = p.end_node
  WHERE p.depth >= 1 AND p.depth <= 3
) AS _proj
GROUP BY TO_JSON(NAMED_STRUCT('_', _gsql2rsql_f_name));
END

Procedural BFS — PySpark 4.2 (numbered_views)

SQL

BEGIN
  DECLARE bfs_depth_1 INT DEFAULT 0;
  DECLARE bfs_frontier_count_1 BIGINT DEFAULT 1;
  DECLARE bfs_union_sql_1 STRING DEFAULT '';

  CREATE OR REPLACE TEMPORARY VIEW bfs_frontier_1_0 AS
  SELECT n.id AS node
  FROM catalog.demo.Person n
  WHERE (n.name) = ('Alice');
  SELECT IF(COUNT(*) > 1, RAISE_ERROR('gsql2rsql procedural BFS is single-source but the start filter matched multiple nodes. Use vlp_rendering_mode="cte" for multi-source traversals.'), TRUE) FROM bfs_frontier_1_0;

  CREATE OR REPLACE TEMPORARY VIEW bfs_visited_1_0 AS
  SELECT node FROM bfs_frontier_1_0;

  WHILE bfs_frontier_count_1 > 0 AND bfs_depth_1 < 3 DO
    SET bfs_depth_1 = bfs_depth_1 + 1;

    EXECUTE IMMEDIATE
      'CREATE OR REPLACE TEMPORARY VIEW bfs_edges_1_' || CAST(bfs_depth_1 AS STRING) || ' AS
       SELECT e.person_id, e.friend_id, e.since, e.strength, e.friend_id AS _next_node, ' || CAST(bfs_depth_1 AS STRING) || ' AS _bfs_depth FROM catalog.demo.Knows e INNER JOIN bfs_frontier_1_' || CAST(bfs_depth_1 - 1 AS STRING) || ' f ON e.person_id = f.node WHERE e.friend_id IS NOT NULL AND NOT EXISTS (SELECT 1 FROM bfs_visited_1_' || CAST(bfs_depth_1 - 1 AS STRING) || ' v WHERE v.node = e.friend_id)';

    EXECUTE IMMEDIATE
      'SELECT COUNT(1) FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING) INTO bfs_frontier_count_1;

    IF bfs_frontier_count_1 > 0 THEN

      EXECUTE IMMEDIATE
        'CREATE OR REPLACE TEMPORARY VIEW bfs_visited_1_' || CAST(bfs_depth_1 AS STRING) || ' AS
         SELECT node FROM bfs_visited_1_' || CAST(bfs_depth_1 - 1 AS STRING) || '
         UNION
         SELECT DISTINCT _next_node AS node
         FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING);

      EXECUTE IMMEDIATE
        'CREATE OR REPLACE TEMPORARY VIEW bfs_frontier_1_' || CAST(bfs_depth_1 AS STRING) || ' AS
         SELECT DISTINCT _next_node AS node
         FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING);

      IF bfs_depth_1 >= 1 THEN
        IF bfs_union_sql_1 = '' THEN
          SET bfs_union_sql_1 =
            'SELECT person_id, friend_id, since, strength, _next_node AS end_node, _bfs_depth AS depth FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING);
        ELSE
          SET bfs_union_sql_1 = bfs_union_sql_1
            || ' UNION ALL SELECT person_id, friend_id, since, strength, _next_node AS end_node, _bfs_depth AS depth FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING);
        END IF;
      END IF;

    END IF;
  END WHILE;

  IF bfs_union_sql_1 != '' THEN
    EXECUTE IMMEDIATE
      'CREATE OR REPLACE TEMPORARY VIEW paths_1 AS
       SELECT f0.node AS start_node, r.end_node, r.depth,
              r.person_id, r.friend_id, r.since, r.strength
       FROM (' || bfs_union_sql_1 || ') r
       CROSS JOIN bfs_frontier_1_0 f0';
  ELSE
    CREATE OR REPLACE TEMPORARY VIEW paths_1 AS
    SELECT
      CAST(NULL AS STRING) AS start_node,
      CAST(NULL AS STRING) AS end_node,
      CAST(NULL AS INT) AS depth,
      CAST(NULL AS STRING) AS person_id,
      CAST(NULL AS STRING) AS friend_id,
      CAST(NULL AS STRING) AS since,
      CAST(NULL AS STRING) AS strength
    WHERE 1 = 0;
  END IF;

  SELECT 
   FIRST(_gsql2rsql_f_name) AS reachable
FROM (
  SELECT
     sink.id AS _gsql2rsql_f_id
    ,sink.name AS _gsql2rsql_f_name
    ,sink.age AS _gsql2rsql_f_age
    ,sink.nickname AS _gsql2rsql_f_nickname
    ,sink.salary AS _gsql2rsql_f_salary
    ,sink.active AS _gsql2rsql_f_active
    ,p.start_node
    ,p.end_node
    ,p.depth
  FROM paths_1 p
  JOIN catalog.demo.Person sink
    ON sink.id = p.end_node
  WHERE p.depth >= 1 AND p.depth <= 3
) AS _proj
GROUP BY TO_JSON(NAMED_STRUCT('_', _gsql2rsql_f_name));
END

2. Variable-length with source AND sink filter pushdown¶

Source: Features

OpenCypher Query

Cypher

MATCH path = (a:Person)-[:KNOWS*2..4]->(b:Person)
WHERE a.age > 30 AND b.age > 50
RETURN a.id, b.id

Procedural BFS — Databricks (temp_tables)

SQL

BEGIN
  DECLARE current_depth_1 INT DEFAULT 0;
  DECLARE rows_in_frontier_1 BIGINT DEFAULT 1;

  DROP TEMPORARY TABLE IF EXISTS bfs_visited_1;
  DROP TEMPORARY TABLE IF EXISTS bfs_frontier_1;
  DROP TEMPORARY TABLE IF EXISTS bfs_result_1;
  DROP TEMPORARY TABLE IF EXISTS bfs_frontier_1_init;
  DROP TEMPORARY TABLE IF EXISTS bfs_edges_keyed_1;
  CREATE TEMPORARY TABLE bfs_visited_1 (node STRING);
  CREATE TEMPORARY TABLE bfs_frontier_1 AS
  SELECT n.id AS node
  FROM catalog.demo.Person n
  WHERE (n.age) > (30);
  INSERT INTO bfs_visited_1
  SELECT node FROM bfs_frontier_1;
  CREATE TEMPORARY TABLE bfs_result_1 (_row_id BIGINT, _next_node STRING, _bfs_depth INT);
  CREATE TEMPORARY TABLE bfs_frontier_1_init AS
  SELECT node FROM bfs_frontier_1;
  SELECT IF(COUNT(*) > 1, RAISE_ERROR('gsql2rsql procedural BFS is single-source but the start filter matched multiple nodes. Use vlp_rendering_mode="cte" for multi-source traversals.'), TRUE) FROM bfs_frontier_1_init;
  CREATE TEMPORARY TABLE bfs_edges_keyed_1 AS
  SELECT e.*, MONOTONICALLY_INCREASING_ID() AS _row_id
  FROM (
  SELECT e.person_id, e.friend_id, e.since, e.strength
  FROM catalog.demo.Knows e
  ) e;

  WHILE rows_in_frontier_1 > 0 AND current_depth_1 < 4 DO
    SET current_depth_1 = current_depth_1 + 1;

    DROP TEMPORARY TABLE IF EXISTS bfs_edges_1;
    CREATE TEMPORARY TABLE bfs_edges_1 AS
    SELECT e._row_id, e.friend_id AS _next_node
  FROM bfs_edges_keyed_1 e
  INNER JOIN bfs_frontier_1 f ON e.person_id = f.node
  WHERE e.friend_id IS NOT NULL AND NOT EXISTS (SELECT 1 FROM bfs_visited_1 v WHERE v.node = e.friend_id);

    SET rows_in_frontier_1 = (SELECT COUNT(1) FROM bfs_edges_1);

    IF rows_in_frontier_1 > 0 THEN
      INSERT INTO bfs_visited_1
      SELECT DISTINCT _next_node FROM bfs_edges_1;
      DROP TEMPORARY TABLE bfs_frontier_1;
      CREATE TEMPORARY TABLE bfs_frontier_1 AS
      SELECT DISTINCT _next_node AS node FROM bfs_edges_1;
      IF current_depth_1 >= 2 THEN
        INSERT INTO bfs_result_1
        SELECT *, current_depth_1 AS _bfs_depth FROM bfs_edges_1;
      END IF;
    END IF;
  END WHILE;

  CREATE OR REPLACE TEMPORARY VIEW paths_1 AS
  SELECT f0.node AS start_node, r._next_node AS end_node, r._bfs_depth AS depth,
         e.person_id, e.friend_id, e.since, e.strength,
         ARRAY(NAMED_STRUCT('person_id', e.person_id, 'friend_id', e.friend_id, 'since', e.since, 'strength', e.strength)) AS path_edges
  FROM bfs_result_1 r
  JOIN bfs_edges_keyed_1 e ON e._row_id = r._row_id
  CROSS JOIN bfs_frontier_1_init f0;

  SELECT 
   _gsql2rsql_a_id AS a_id
  ,_gsql2rsql_b_id AS b_id
FROM (
  SELECT
     sink.id AS _gsql2rsql_b_id
    ,sink.name AS _gsql2rsql_b_name
    ,sink.age AS _gsql2rsql_b_age
    ,sink.nickname AS _gsql2rsql_b_nickname
    ,sink.salary AS _gsql2rsql_b_salary
    ,sink.active AS _gsql2rsql_b_active
    ,source.id AS _gsql2rsql_a_id
    ,source.name AS _gsql2rsql_a_name
    ,source.age AS _gsql2rsql_a_age
    ,source.nickname AS _gsql2rsql_a_nickname
    ,source.salary AS _gsql2rsql_a_salary
    ,source.active AS _gsql2rsql_a_active
    ,p.start_node
    ,p.end_node
    ,p.depth
    ,p.path_edges AS _gsql2rsql_path_edges
  FROM paths_1 p
  JOIN catalog.demo.Person sink
    ON sink.id = p.end_node
  JOIN catalog.demo.Person source
    ON source.id = p.start_node
  WHERE p.depth >= 2 AND p.depth <= 4 AND (sink.age) > (50)
) AS _proj;
END

Procedural BFS — PySpark 4.2 (numbered_views)

SQL

BEGIN
  DECLARE bfs_depth_1 INT DEFAULT 0;
  DECLARE bfs_frontier_count_1 BIGINT DEFAULT 1;
  DECLARE bfs_union_sql_1 STRING DEFAULT '';

  CREATE OR REPLACE TEMPORARY VIEW bfs_frontier_1_0 AS
  SELECT n.id AS node
  FROM catalog.demo.Person n
  WHERE (n.age) > (30);
  SELECT IF(COUNT(*) > 1, RAISE_ERROR('gsql2rsql procedural BFS is single-source but the start filter matched multiple nodes. Use vlp_rendering_mode="cte" for multi-source traversals.'), TRUE) FROM bfs_frontier_1_0;

  CREATE OR REPLACE TEMPORARY VIEW bfs_visited_1_0 AS
  SELECT node FROM bfs_frontier_1_0;

  WHILE bfs_frontier_count_1 > 0 AND bfs_depth_1 < 4 DO
    SET bfs_depth_1 = bfs_depth_1 + 1;

    EXECUTE IMMEDIATE
      'CREATE OR REPLACE TEMPORARY VIEW bfs_edges_1_' || CAST(bfs_depth_1 AS STRING) || ' AS
       SELECT e.person_id, e.friend_id, e.since, e.strength, e.friend_id AS _next_node, ' || CAST(bfs_depth_1 AS STRING) || ' AS _bfs_depth FROM catalog.demo.Knows e INNER JOIN bfs_frontier_1_' || CAST(bfs_depth_1 - 1 AS STRING) || ' f ON e.person_id = f.node WHERE e.friend_id IS NOT NULL AND NOT EXISTS (SELECT 1 FROM bfs_visited_1_' || CAST(bfs_depth_1 - 1 AS STRING) || ' v WHERE v.node = e.friend_id)';

    EXECUTE IMMEDIATE
      'SELECT COUNT(1) FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING) INTO bfs_frontier_count_1;

    IF bfs_frontier_count_1 > 0 THEN

      EXECUTE IMMEDIATE
        'CREATE OR REPLACE TEMPORARY VIEW bfs_visited_1_' || CAST(bfs_depth_1 AS STRING) || ' AS
         SELECT node FROM bfs_visited_1_' || CAST(bfs_depth_1 - 1 AS STRING) || '
         UNION
         SELECT DISTINCT _next_node AS node
         FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING);

      EXECUTE IMMEDIATE
        'CREATE OR REPLACE TEMPORARY VIEW bfs_frontier_1_' || CAST(bfs_depth_1 AS STRING) || ' AS
         SELECT DISTINCT _next_node AS node
         FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING);

      IF bfs_depth_1 >= 2 THEN
        IF bfs_union_sql_1 = '' THEN
          SET bfs_union_sql_1 =
            'SELECT person_id, friend_id, since, strength, _next_node AS end_node, _bfs_depth AS depth FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING);
        ELSE
          SET bfs_union_sql_1 = bfs_union_sql_1
            || ' UNION ALL SELECT person_id, friend_id, since, strength, _next_node AS end_node, _bfs_depth AS depth FROM bfs_edges_1_' || CAST(bfs_depth_1 AS STRING);
        END IF;
      END IF;

    END IF;
  END WHILE;

  IF bfs_union_sql_1 != '' THEN
    EXECUTE IMMEDIATE
      'CREATE OR REPLACE TEMPORARY VIEW paths_1 AS
       SELECT f0.node AS start_node, r.end_node, r.depth,
              r.person_id, r.friend_id, r.since, r.strength,
              ARRAY(NAMED_STRUCT(''person_id'', r.person_id, ''friend_id'', r.friend_id, ''since'', r.since, ''strength'', r.strength)) AS path_edges
       FROM (' || bfs_union_sql_1 || ') r
       CROSS JOIN bfs_frontier_1_0 f0';
  ELSE
    CREATE OR REPLACE TEMPORARY VIEW paths_1 AS
    SELECT
      CAST(NULL AS STRING) AS start_node,
      CAST(NULL AS STRING) AS end_node,
      CAST(NULL AS INT) AS depth,
      CAST(NULL AS STRING) AS person_id,
      CAST(NULL AS STRING) AS friend_id,
      CAST(NULL AS STRING) AS since,
      CAST(NULL AS STRING) AS strength,
      CAST(NULL AS ARRAY<STRUCT<person_id: STRING, friend_id: STRING, since: STRING, strength: STRING>>) AS path_edges
    WHERE 1 = 0;
  END IF;

  SELECT 
   _gsql2rsql_a_id AS a_id
  ,_gsql2rsql_b_id AS b_id
FROM (
  SELECT
     sink.id AS _gsql2rsql_b_id
    ,sink.name AS _gsql2rsql_b_name
    ,sink.age AS _gsql2rsql_b_age
    ,sink.nickname AS _gsql2rsql_b_nickname
    ,sink.salary AS _gsql2rsql_b_salary
    ,sink.active AS _gsql2rsql_b_active
    ,source.id AS _gsql2rsql_a_id
    ,source.name AS _gsql2rsql_a_name
    ,source.age AS _gsql2rsql_a_age
    ,source.nickname AS _gsql2rsql_a_nickname
    ,source.salary AS _gsql2rsql_a_salary
    ,source.active AS _gsql2rsql_a_active
    ,p.start_node
    ,p.end_node
    ,p.depth
    ,p.path_edges AS _gsql2rsql_path_edges
  FROM paths_1 p
  JOIN catalog.demo.Person sink
    ON sink.id = p.end_node
  JOIN catalog.demo.Person source
    ON source.id = p.start_node
  WHERE p.depth >= 2 AND p.depth <= 4 AND (sink.age) > (50)
) AS _proj;
END