⚡️ Speed up function find_last_node by 26,750%

[codeflash-ai]

📄 26,750% (267.50x) speedup for find_last_node in src/algorithms/graph.py

⏱️ Runtime : 102 milliseconds → 381 microseconds (best of 250 runs)

📝 Explanation and details

The optimized code achieves a 267x speedup by eliminating a nested loop antipattern that caused O(n*m) complexity, reducing it to O(n+m).

Key Optimization:

The original code checked all(e["source"] != n["id"] for e in edges) for every node, resulting in nested iteration: for each of the n nodes, it scanned all m edges. This created O(n*m) time complexity.

The optimized version pre-computes a set of all source IDs with source_ids = {e["source"] for e in edges}, then uses fast O(1) set membership testing with n["id"] not in source_ids. This reduces complexity to O(m) for set creation + O(n) for the lookup, giving O(n+m) overall.

Why This Is Faster:

1. Set lookups are O(1) vs. the O(m) linear scan through all edges for each node
2. Single-pass edge iteration instead of re-scanning edges for every node
3. Hash-based membership testing is dramatically faster than repeated comparisons

Performance Characteristics:

• Small graphs (≤10 nodes/edges): 20-95% faster - modest gains since overhead is low anyway
• Large linear chains (1000 nodes): 330x faster - the nested loop penalty becomes severe
• Large cycles: 3000-33000% faster - every node previously required full edge traversal
• Empty/tiny inputs: Slightly slower (9-28%) due to set construction overhead, but negligible absolute time (<1μs difference)

The optimization is especially valuable when find_last_node is called repeatedly on moderately-sized graphs, as the speedup scales quadratically with graph size. Since finding terminal nodes is a common graph traversal pattern (e.g., identifying workflow endpoints, DAG sinks), this optimization would benefit any hot path performing graph analysis.

✅ Correctness verification report:

Test	Status
⚙️ Existing Unit Tests	🔘 None Found
🌀 Generated Regression Tests	✅ 38 Passed
⏪ Replay Tests	🔘 None Found
🔎 Concolic Coverage Tests	🔘 None Found
📊 Tests Coverage	100.0%

🌀 Click to see Generated Regression Tests

import pytest
from src.algorithms.graph import find_last_node

# unit tests

# ----------- BASIC TEST CASES -----------


def test_single_node_no_edges():
    # One node, no edges: should return the node itself
    nodes = [{"id": 1}]
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.25μs -> 1.00μs (25.0% faster)


def test_two_nodes_one_edge():
    # Two nodes, one edge from 1 to 2: last node is 2
    nodes = [{"id": 1}, {"id": 2}]
    edges = [{"source": 1, "target": 2}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.83μs -> 1.17μs (57.2% faster)


def test_three_nodes_linear_chain():
    # 1->2->3, last node is 3
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
    edges = [{"source": 1, "target": 2}, {"source": 2, "target": 3}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 2.21μs -> 1.21μs (82.9% faster)


def test_branching_graph():
    # 1->2, 1->3, both 2 and 3 are last nodes (no outgoing edges)
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
    edges = [{"source": 1, "target": 2}, {"source": 1, "target": 3}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.88μs -> 1.25μs (50.0% faster)


def test_multiple_last_nodes_returns_first():
    # 1->2, 1->3, both 2 and 3 are last, function should return first in nodes order
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
    edges = [{"source": 1, "target": 2}, {"source": 1, "target": 3}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.83μs -> 1.21μs (51.6% faster)


# ----------- EDGE TEST CASES -----------


def test_no_nodes():
    # No nodes at all: should return None
    nodes = []
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 792ns -> 917ns (13.6% slower)


def test_no_edges_multiple_nodes():
    # Multiple nodes, no edges: first node should be returned
    nodes = [{"id": "a"}, {"id": "b"}, {"id": "c"}]
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.25μs -> 1.00μs (25.0% faster)


def test_cycle_graph():
    # 1->2, 2->3, 3->1 (cycle): no node without outgoing edges, should return None
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
    edges = [
        {"source": 1, "target": 2},
        {"source": 2, "target": 3},
        {"source": 3, "target": 1},
    ]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 2.21μs -> 1.33μs (65.6% faster)


def test_disconnected_nodes():
    # Some nodes not connected by any edge
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
    edges = [{"source": 1, "target": 2}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.79μs -> 1.17μs (53.6% faster)


def test_node_with_multiple_outgoing_edges():
    # 1->2, 1->3, 2->4, 3->4: only node 4 is last node
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}, {"id": 4}]
    edges = [
        {"source": 1, "target": 2},
        {"source": 1, "target": 3},
        {"source": 2, "target": 4},
        {"source": 3, "target": 4},
    ]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 2.67μs -> 1.38μs (94.0% faster)


def test_node_with_non_integer_ids():
    # Node IDs are strings
    nodes = [{"id": "start"}, {"id": "end"}]
    edges = [{"source": "start", "target": "end"}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.96μs -> 1.25μs (56.6% faster)


def test_edges_with_nonexistent_nodes():
    # Edges refer to node IDs not in nodes list
    nodes = [{"id": 1}]
    edges = [{"source": 2, "target": 1}]  # source does not exist in nodes
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.42μs -> 1.08μs (30.7% faster)


def test_duplicate_nodes():
    # Duplicate nodes in list, only the first should be returned
    nodes = [{"id": 1}, {"id": 2}, {"id": 2}]
    edges = [{"source": 1, "target": 2}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.79μs -> 1.12μs (59.3% faster)


def test_empty_edges_with_empty_nodes():
    # Both lists empty
    nodes = []
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 791ns -> 875ns (9.60% slower)


# ----------- LARGE SCALE TEST CASES -----------


def test_large_linear_chain():
    # 1000 nodes in a chain: 1->2->3->...->1000, last node is 1000
    N = 1000
    nodes = [{"id": i} for i in range(1, N + 1)]
    edges = [{"source": i, "target": i + 1} for i in range(1, N)]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 18.6ms -> 57.1μs (32507% faster)


def test_large_branching_graph():
    # 500 nodes, each node i (1..499) points to i+1 and i+2, last nodes are 500 and 501 (but only 500 exists)
    N = 500
    nodes = [{"id": i} for i in range(1, N + 1)]
    edges = []
    for i in range(1, N - 1):
        edges.append({"source": i, "target": i + 1})
        edges.append({"source": i, "target": i + 2})
    # last two nodes (N, N-1) have no outgoing edges, should return N-1 as it appears first
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 9.16ms -> 38.3μs (23806% faster)


def test_large_graph_with_disconnected_last_node():
    # 999 nodes connected in a chain, one extra node disconnected
    N = 999
    nodes = [{"id": i} for i in range(1, N + 2)]  # 1..1000
    edges = [{"source": i, "target": i + 1} for i in range(1, N)]
    # node 1000 is disconnected
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 18.7ms -> 56.7μs (32951% faster)


def test_large_cycle_graph():
    # 100 nodes in a cycle: 1->2->...->100->1, should return None
    N = 100
    nodes = [{"id": i} for i in range(1, N + 1)]
    edges = [{"source": i, "target": i % N + 1} for i in range(1, N + 1)]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 206μs -> 6.58μs (3031% faster)


def test_large_graph_all_isolated():
    # 1000 nodes, no edges: should return first node
    N = 1000
    nodes = [{"id": i} for i in range(N)]
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.38μs -> 1.04μs (32.0% faster)


# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

from __future__ import annotations

# imports
import pytest  # used for our unit tests
from src.algorithms.graph import find_last_node

# unit tests

# --------- Basic Test Cases ---------


def test_single_node_no_edges():
    # Only one node, no edges; should return that node
    nodes = [{"id": 1, "label": "A"}]
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.21μs -> 1.00μs (20.8% faster)


def test_two_nodes_one_edge():
    # Two nodes, one edge from first to second; last node is the one with no outgoing edge
    nodes = [{"id": 1, "label": "A"}, {"id": 2, "label": "B"}]
    edges = [{"source": 1, "target": 2}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.75μs -> 1.12μs (55.6% faster)


def test_three_nodes_linear_chain():
    # Three nodes in a chain: 1 -> 2 -> 3; last node is 3
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
    edges = [{"source": 1, "target": 2}, {"source": 2, "target": 3}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 2.17μs -> 1.21μs (79.2% faster)


def test_multiple_possible_last_nodes():
    # Two nodes, no edges; both are last nodes, should return the first found
    nodes = [{"id": 1}, {"id": 2}]
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.17μs -> 958ns (21.8% faster)


def test_last_node_with_multiple_incoming_edges():
    # Three nodes, edges: 1->3, 2->3; node 3 has no outgoing, 1 and 2 do
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
    edges = [{"source": 1, "target": 3}, {"source": 2, "target": 3}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 2.21μs -> 1.25μs (76.6% faster)


# --------- Edge Test Cases ---------


def test_empty_nodes_and_edges():
    # No nodes, no edges; should return None
    nodes = []
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 625ns -> 875ns (28.6% slower)


def test_nodes_but_edges_to_nonexistent_nodes():
    # Nodes exist, but edges point to non-existent nodes
    nodes = [{"id": 1}, {"id": 2}]
    edges = [{"source": 3, "target": 4}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.38μs -> 1.12μs (22.2% faster)


def test_cycle_graph():
    # Nodes in a cycle: 1->2->3->1; all have outgoing edges, so return None
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
    edges = [
        {"source": 1, "target": 2},
        {"source": 2, "target": 3},
        {"source": 3, "target": 1},
    ]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 2.17μs -> 1.21μs (79.4% faster)


def test_duplicate_node_ids():
    # Duplicate node IDs; function should return the first one with no outgoing edge
    nodes = [{"id": 1}, {"id": 1}, {"id": 2}]
    edges = [{"source": 1, "target": 2}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 2.04μs -> 1.21μs (69.0% faster)


def test_nodes_with_non_integer_ids():
    # Node IDs are strings
    nodes = [{"id": "A"}, {"id": "B"}]
    edges = [{"source": "A", "target": "B"}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.92μs -> 1.25μs (53.4% faster)


def test_edge_list_with_extra_keys():
    # Edges have extra keys; should ignore them
    nodes = [{"id": 1}, {"id": 2}]
    edges = [{"source": 1, "target": 2, "weight": 5}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.75μs -> 1.12μs (55.6% faster)


def test_node_dict_with_extra_keys():
    # Nodes have extra keys; should return the full node dict
    nodes = [{"id": 1, "label": "A", "foo": "bar"}, {"id": 2, "label": "B"}]
    edges = [{"source": 1, "target": 2}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.79μs -> 1.17μs (53.6% faster)


def test_all_nodes_have_outgoing_edges():
    # All nodes have outgoing edges; should return None
    nodes = [{"id": 1}, {"id": 2}]
    edges = [{"source": 1, "target": 2}, {"source": 2, "target": 1}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 2.00μs -> 1.25μs (60.0% faster)


def test_nodes_with_none_id():
    # Nodes with None as id
    nodes = [{"id": None}, {"id": 2}]
    edges = [{"source": 2, "target": None}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.62μs -> 1.12μs (44.4% faster)


# --------- Large Scale Test Cases ---------


def test_large_linear_chain():
    # 1000 nodes in a chain: 0 -> 1 -> 2 -> ... -> 999
    N = 1000
    nodes = [{"id": i} for i in range(N)]
    edges = [{"source": i, "target": i + 1} for i in range(N - 1)]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 18.6ms -> 56.1μs (33114% faster)


def test_large_star_graph():
    # Star graph: node 0 points to all others; last nodes are 1..999
    N = 1000
    nodes = [{"id": i} for i in range(N)]
    edges = [{"source": 0, "target": i} for i in range(1, N)]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 38.2μs -> 20.8μs (83.6% faster)


def test_large_disconnected_graph():
    # 1000 nodes, no edges; all are last nodes, should return the first node
    N = 1000
    nodes = [{"id": i} for i in range(N)]
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.38μs -> 1.08μs (27.0% faster)


def test_large_all_have_outgoing_edges():
    # 1000 nodes, each has an outgoing edge to next, last connects to first (cycle)
    N = 1000
    nodes = [{"id": i} for i in range(N)]
    edges = [{"source": i, "target": (i + 1) % N} for i in range(N)]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 18.6ms -> 56.1μs (33077% faster)


def test_large_with_some_isolated_nodes():
    # 990 nodes in a chain, 10 isolated nodes at the end
    N = 1000
    nodes = [{"id": i} for i in range(N)]
    edges = [{"source": i, "target": i + 1} for i in range(989)]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 18.3ms -> 55.7μs (32829% faster)


# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

To edit these changes git checkout codeflash/optimize-find_last_node-mjncnrj1 and push.