🔬 Algorithm Deep Dive

Mathematical Definition

Pseudo-Code

function initialize_query_state(query: Query) → QueryState:
    Σ ← empty QueryState
    
    // Step 1: Resolve FROM schema
    F ← resolve_schema(query.from_clause())
    
    // Step 2: Load table and extract IDs
    T_F ← database.get_table(F)
    N_F ← extract_ids(T_F)  // O(|T_F|)
    
    // Step 3: Initialize state
    Σ.ids[F] ← N_F
    Σ.tables[F] ← T_F
    Σ.aliases[F.alias] ← F.schema
    Σ.connections ← ∅
    Σ.incoming ← ∅
    
    // Invariant check
    assert(|Σ.ids[F]| = |T_F|)
    assert(Σ.connections.empty())
    
    return Σ

Query: FROM u:users

Initial State Σ₀:
  ids = {
    u: {0, 1, 2, 3, 4}  // All 5 users
  }
  tables = {
    u: Arrow::Table(5 rows × 3 columns)
        ┌────┬───────┬─────┐
        │ id │ name  │ age │
        ├────┼───────┼─────┤
        │ 0  │ alex  │ 25  │
        │ 1  │ bob   │ 31  │
        │ 2  │ jeff  │ 33  │
        │ 3  │ sam   │ 21  │
        │ 4  │ matt  │ 40  │
        └────┴───────┴─────┘
  }
  connections = ∅
  aliases = { u → users }

Mathematical Formulation

Join Type Semantics

Pseudo-Code with Complexity Analysis

function process_traverse(traverse: Traverse, Σ: QueryState) → QueryState:
    // Resolve schemas
    s_schema ← resolve_schema(traverse.source, Σ)
    d_schema ← resolve_schema(traverse.dest, Σ)
    
    // Initialize tracking sets
    M_s ← ∅  // Matched sources
    M_d ← ∅  // Matched destinations
    U_s ← ∅  // Unmatched sources
    
    // Phase 2.1: Traverse edges  [Complexity: O(|Σ.ids[s]| × E_avg)]
    for each u in Σ.ids[s]:
        edges ← edge_store.get_outgoing_edges(u, traverse.edge_type)  // O(E_avg)
        
        had_match ← false
        for each edge in edges:
            v ← edge.target_id
            
            // Check if target exists and passes filters
            node ← get_node(d_schema, v)  // O(1) hash lookup
            if node = null or not apply_filters(node, WHERE_clauses):
                continue
            
            // Valid edge found
            had_match ← true
            M_s.insert(u)
            M_d.insert(v)
            
            // Record connection  [Complexity: O(1)]
            conn ← GraphConnection(s, u, traverse.edge_type, d, v)
            Σ.connections[s][u].append(conn)
            Σ.incoming[v].append(conn)
        
        if not had_match:
            U_s.insert(u)
    
    // Phase 2.2: Apply join logic  [Complexity: O(|M_s| + |M_d|)]
    Σ' ← apply_join_logic(traverse.join_type, M_s, M_d, U_s, Σ)
    
    // Phase 2.3: Build target table  [Complexity: O(|Σ'.ids[d]| × F)]
    target_nodes ← [get_node(d_schema, v) for v in Σ'.ids[d]]
    Σ'.tables[d] ← create_table_from_nodes(d_schema, target_nodes)
    
    // Invariant checks
    assert(all(u in Σ'.ids[s] for (u,v) in Σ'.connections))
    assert(all(v in Σ'.ids[d] for (u,v) in Σ'.connections))
    
    return Σ'

Query: FROM u:users TRAVERSE u-[friend]->f:users INNER

State Before (Σ₀):
  ids[u] = {0, 1, 2, 3, 4}

Edge Traversal:
  u:0 → edges = [(0, friend, 1)] → M_s.add(0), M_d.add(1)
  u:1 → edges = [] → U_s.add(1)
  u:2 → edges = [] → U_s.add(2)
  u:3 → edges = [] → U_s.add(3)
  u:4 → edges = [] → U_s.add(4)

Results:
  M_s = {0}       // Only alex has a friend
  M_d = {1}       // Only bob is a friend
  U_s = {1,2,3,4} // 4 users with no friends

INNER JOIN Logic:
  Remove U_s from ids[u]
  ids[u]' = M_s = {0}
  ids[f]' = M_d = {1}

State After (Σ₁):
  ids = {
    u: {0},     // Only alex (has friend)
    f: {1}      // Only bob (is friend)
  }
  connections = {
    u:0 → [(0, friend, 1)]
  }
  incoming = {
    1: [(u:0, friend, f:1)]
  }

Self-Join vs Cross-Schema: Critical Distinction

Scenario: users:0 -[works-at]-> companies:0

Per-Schema IDs:
  users:     {0, 1, 2, 3, 4}
  companies: {0, 1, 2}

FULL JOIN (WRONG approach - using M_s):
  M_s = {0}  // User 0 (alex) works at a company
  Unmatched = companies - M_s = {0,1,2} - {0} = {1,2}
  ❌ WRONG! Company:0 (IBM) is excluded due to ID collision with User:0

FULL JOIN (CORRECT approach - using M_d):
  M_d = {0}  // Company 0 (IBM) is matched
  Unmatched = companies - M_d = {0,1,2} - {0} = {1,2}
  ✅ CORRECT! All companies properly included

Mathematical Foundation

BFS Algorithm with State Tracking

function populate_rows_bfs(start_id: int, start_schema: Schema, 
                          output_schema: Schema, Σ: QueryState) → List[Row]:
    
    // Initialize state
    Q ← Queue()  // BFS queue
    R ← []       // Result rows
    G_visited ← Set()  // Global visited (across all starting nodes)
    row_counter ← 0
    
    // Create initial row and queue item
    r₀ ← create_empty_row(output_schema)
    q₀ ← QueueItem(start_id, start_schema, depth=0, row=r₀, path=[], path_visited=∅)
    Q.enqueue(q₀)
    
    // BFS Loop  [Complexity: O(|paths| × P_avg) where P_avg = avg path length]
    while not Q.empty():
        item ← Q.dequeue()
        
        // Step 1: Mark as visited (cycle prevention)
        packed_id ← hash(item.schema, item.node_id)
        if packed_id in item.path_visited:
            continue  // Skip if already in current path (cycle)
        
        G_visited.insert(packed_id)
        item.path_visited.insert(packed_id)
        
        // Step 2: Fill row with current node's fields
        node ← get_node(item.schema, item.node_id)
        for each field in item.schema.fields:
            field_idx ← output_schema.field_index(item.schema.alias + "." + field.name)
            item.row[field_idx] ← node.get_value(field)
        
        item.path.append((item.schema, item.node_id))
        
        // Step 3: Get outgoing connections, grouped by target schema
        grouped_conns ← group_by_target_schema(Σ.connections[item.schema][item.node_id])
        
        // Remove connections to already-visited nodes in this path
        for schema, conns in grouped_conns:
            filtered_conns ← [c for c in conns if hash(c.target, c.target_id) ∉ item.path_visited]
            grouped_conns[schema] = filtered_conns
        
        // Step 4: Branch or complete
        if grouped_conns.empty():
            // ✅ Complete path - add to results
            r ← clone(item.row)
            r.id ← row_counter++
            r.path ← item.path
            R.append(r)
        else:
            // 🌿 Branch: Continue traversal
            for schema, conns in grouped_conns:
                if |conns| = 1:
                    // Single connection: reuse row (optimization)
                    conn ← conns[0]
                    q_next ← QueueItem(
                        node_id=conn.target_id,
                        schema=conn.target,
                        depth=item.depth + 1,
                        row=item.row,  // Reuse row
                        path=item.path.copy(),
                        path_visited=item.path_visited.copy()
                    )
                    Q.enqueue(q_next)
                else:
                    // Multiple connections: clone row for each branch
                    for conn in conns:
                        r_new ← clone(item.row)
                        q_next ← QueueItem(
                            node_id=conn.target_id,
                            schema=conn.target,
                            depth=item.depth + 1,
                            row=r_new,  // New row
                            path=item.path.copy(),
                            path_visited=item.path_visited.copy()
                        )
                        Q.enqueue(q_next)
    
    // Step 5: Merge rows (handle multi-path scenarios)
    tree ← RowNode(root)
    for r in R:
        tree.insert_row(r)
    
    R_merged ← tree.merge_rows()
    
    return R_merged

Interactive State Visualization

Query: u:users -[friend]-> f:users -[works-at]-> c:companies

Data:
  u:0 (alex) -[friend]-> u:1 (bob) -[works-at]-> c:1 (google)

═══════════════════════════════════════════════════════════════

Initial State:
  Q = [QueueItem(u:0, depth=0, row={}, path=[], visited=∅)]
  R = []
  G_visited = ∅

Iteration 1: Process u:0 (alex)
  ├─ Dequeue: QueueItem(u:0, depth=0, ...)
  ├─ Mark visited: G_visited = {u:0}
  ├─ Fill row: row[u.id]=0, row[u.name]="alex", row[u.age]=25
  ├─ Get connections: u:0 → {f:1}
  ├─ Not complete (has connections)
  └─ Enqueue: QueueItem(f:1, depth=1, row={u.id:0, ...}, path=[(u,0)], visited={u:0})

  Q = [QueueItem(f:1, depth=1, ...)]
  R = []
  G_visited = {u:0}

Iteration 2: Process f:1 (bob)
  ├─ Dequeue: QueueItem(f:1, depth=1, ...)
  ├─ Mark visited: G_visited = {u:0, f:1}
  ├─ Fill row: row[f.id]=1, row[f.name]="bob", row[f.age]=31
  ├─ Get connections: f:1 → {c:1}
  ├─ Not complete (has connections)
  └─ Enqueue: QueueItem(c:1, depth=2, row={u.id:0, ..., f.id:1, ...}, path=[(u,0), (f,1)], visited={u:0, f:1})

  Q = [QueueItem(c:1, depth=2, ...)]
  R = []
  G_visited = {u:0, f:1}

Iteration 3: Process c:1 (google)
  ├─ Dequeue: QueueItem(c:1, depth=2, ...)
  ├─ Mark visited: G_visited = {u:0, f:1, c:1}
  ├─ Fill row: row[c.id]=1, row[c.name]="google", row[c.size]=3000
  ├─ Get connections: c:1 → ∅ (no outgoing)
  ├─ ✅ Complete path!
  └─ Add to results: R.append({u.id:0, u.name:"alex", ..., f.id:1, ..., c.id:1, ...})

  Q = []
  R = [Row(id=0, path=[(u,0), (f,1), (c,1)])]
  G_visited = {u:0, f:1, c:1}

BFS Complete!
  Return R = [Row{u.id:0, u.name:"alex", u.age:25, f.id:1, f.name:"bob", f.age:31, c.id:1, c.name:"google", c.size:3000}]

Complexity Analysis

Mathematical Model

Tree Construction Algorithm

struct RowNode:
    path_segment: (schema, node_id)
    row: Row
    children: Map[PathSegment, RowNode]

function insert_row(tree: RowNode, row: Row) → void:
    """Insert a row into the tree based on its path."""
    current ← tree
    
    for segment in row.path:
        if segment not in current.children:
            // Create new tree node
            current.children[segment] ← RowNode(segment, empty_row, {})
        current ← current.children[segment]
    
    // At leaf: merge row data
    current.row ← merge_rows(current.row, row)

function merge_rows(r1: Row, r2: Row) → Row:
    """Merge two rows field-by-field."""
    r_merged ← clone(r1)
    
    for i in 0 to |r2.fields|:
        if r1.fields[i] is NULL and r2.fields[i] is not NULL:
            r_merged.fields[i] ← r2.fields[i]
        elif r1.fields[i] is not NULL and r2.fields[i] is NULL:
            r_merged.fields[i] ← r1.fields[i]
        elif r1.fields[i] = r2.fields[i]:
            r_merged.fields[i] ← r1.fields[i]
        else:
            // Conflict: prefer first row
            r_merged.fields[i] ← r1.fields[i]
    
    return r_merged

function flatten_tree(tree: RowNode) → List[Row]:
    """Convert tree back to flat list of rows."""
    result ← []
    
    function dfs(node: RowNode, accumulated_row: Row):
        // Merge current node's data
        current_row ← merge_rows(accumulated_row, node.row)
        
        if node.children.empty():
            // Leaf: complete row
            result.append(current_row)
        else:
            // Internal: recurse to children
            for child in node.children.values():
                dfs(child, current_row)
    
    dfs(tree, empty_row)
    return result

Scenario: One user with TWO friends

Data:
  u:0 (alex) -[friend]-> u:1 (bob)
  u:0 (alex) -[friend]-> u:2 (jeff)

BFS generates 2 rows:
  Row₁: {u.id:0, u.name:"alex", u.age:25, f.id:1, f.name:"bob", f.age:31}
  Row₂: {u.id:0, u.name:"alex", u.age:25, f.id:2, f.name:"jeff", f.age:33}

Tree Construction:
  
  root
   └─ u:0 (alex)           [u.id:0, u.name:"alex", u.age:25]
       ├─ f:1 (bob)        [f.id:1, f.name:"bob", f.age:31]
       └─ f:2 (jeff)       [f.id:2, f.name:"jeff", f.age:33]

Flattened Result (2 rows):
  ┌────┬───────┬─────┬────┬───────┬─────┐
  │ u.id │ u.name │ u.age │ f.id │ f.name │ f.age │
  ├────┼───────┼─────┼────┼───────┼─────┤
  │ 0   │ alex  │ 25  │ 1  │ bob   │ 31  │
  │ 0   │ alex  │ 25  │ 2  │ jeff  │ 33  │
  └────┴───────┴─────┴────┴───────┴─────┘

Note: alex's data (u.*) is duplicated in both rows.

Overall Complexity

Optimization Strategies

// Instead of allocating new connections:
for each edge:
    conn = new GraphConnection(...)  // ❌ Allocation overhead

// Use object pool:
for each edge:
    conn = connection_pool.get()     // ✅ Reuse existing
    conn.update(source, target, ...)
    state.connections.append(conn)