⚡️ Speed up function find_last_node by 17,331%

[codeflash-ai]

📄 17,331% (173.31x) speedup for find_last_node in src/algorithms/graph.py

⏱️ Runtime : 64.2 milliseconds → 369 microseconds (best of 250 runs)

📝 Explanation and details

The optimized code achieves a 174x speedup (17,331% improvement) by eliminating redundant edge scanning through a fundamental algorithmic optimization.

Key Optimization: Set-Based Lookup

Original approach (O(|nodes| × |edges|)):

• For each node, scans all edges to check if n["id"] matches any e["source"]
• With 1000 nodes and 1000 edges, this performs ~1,000,000 comparisons
• Line profiler shows 444ms spent in the nested iteration

Optimized approach (O(|nodes| + |edges|)):

• Pre-computes a set of all source IDs: sources = {e["source"] for e in edges} (single pass)
• Uses set membership testing: n["id"] not in sources (O(1) lookup per node)
• Line profiler shows only 571μs total (296μs for set creation, 275μs for node iteration)

Why This Works in Python

Set membership testing (in operator) uses hash table lookups with O(1) average time complexity. The original code's nested all() generator with edge iteration forces repeated linear scans. By converting edges to a set once, we trade a small upfront cost (296μs) for massive savings on the node iteration phase.

Performance Impact by Test Case

The optimization excels when:

• Large edge counts: test_large_linear_chain_1000_nodes improves from 17.9ms → 46μs (389x faster)
• Dense graphs: test_large_graph_with_many_edges_single_sink improves from 9.24ms → 32.8μs (281x faster)
• Late sinks: test_large_linear_chain_find_middle_sink improves from 4.49ms → 24μs (187x faster)

Even small graphs benefit (40-90% faster) due to Python's efficient set implementation. The only minor regression is test_no_nodes_returns_none (24% slower) where the set creation overhead exceeds the benefit for empty graphs—an acceptable trade-off given the dramatic improvements in real-world scenarios.

Impact Summary

This optimization transforms the function from quadratic to linear complexity, making it viable for production workflows with large graph structures. The 174x average speedup means operations that took ~64ms now complete in ~369μs, enabling real-time graph analysis in data pipelines and workflow engines.

✅ Correctness verification report:

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

🌀 Click to see Generated Regression Tests

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


def test_single_edge_returns_sink_node():
    # One edge from node 'a' to 'b' -> 'b' has no outgoing edges, should be returned
    nodes = [{"id": "a"}, {"id": "b"}]  # nodes in iteration order
    edges = [{"source": "a", "target": "b"}]  # only 'a' is a source
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.58μs -> 1.04μs (51.9% faster)


def test_multiple_sinks_returns_first_in_nodes_order():
    # Two nodes with no outgoing edges; function should return the first one encountered
    nodes = [{"id": "first"}, {"id": "second"}, {"id": "third"}]
    # edges connect only 'third' to something else; 'first' and 'second' are sinks,
    # but 'first' is first in nodes order and must be returned.
    edges = [{"source": "third", "target": "x"}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.29μs -> 917ns (40.8% faster)


def test_edges_empty_all_nodes_are_sinks_first_returned():
    # With an empty edges list every node has zero outgoing edges.
    nodes = [{"id": 1}, {"id": 2}, {"id": 3}]
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.04μs -> 917ns (13.6% faster)


def test_no_nodes_returns_none():
    # If there are no nodes there can't be a sink; expect None
    nodes = []
    edges = [{"source": "anything"}]
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 667ns -> 875ns (23.8% slower)


def test_node_missing_id_raises_keyerror():
    # If a node dict lacks "id", the function will attempt to access it and raise KeyError.
    nodes = [{"name": "no-id"}, {"id": "ok"}]
    edges = [{"source": "ok"}]
    with pytest.raises(KeyError):
        # Attempting to evaluate n["id"] on the first node must raise
        find_last_node(nodes, edges)  # 1.62μs -> 1.21μs (34.4% faster)


def test_edge_missing_source_raises_keyerror():
    # If an edge dict lacks "source", accessing e["source"] should raise KeyError.
    nodes = [{"id": "a"}]
    edges = [{"src": "a"}]  # wrong key name
    with pytest.raises(KeyError):
        find_last_node(nodes, edges)  # 1.42μs -> 792ns (78.9% faster)


def test_type_sensitivity_between_string_and_int_ids():
    # Confirm that id types are compared strictly (no coercion).
    # node id is the string "1", edge source is the int 1 -> they are not equal,
    # so the node should be considered a sink.
    nodes = [{"id": "1"}, {"id": 2}]
    edges = [{"source": 1}]  # integer 1, not string "1"
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.38μs -> 1.00μs (37.5% faster)


def test_duplicate_node_ids_all_excluded_if_one_has_outgoing():
    # If multiple nodes share the same id and there is an edge with that id as source,
    # then all nodes with that id should be excluded (none considered sinks).
    nodes = [{"id": "dup"}, {"id": "dup"}, {"id": "other"}]
    edges = [{"source": "dup"}]  # any node with id "dup" has an outgoing edge
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.83μs -> 1.12μs (62.9% faster)


def test_none_inputs_raise_typeerror_for_nodes():
    # Passing None for nodes should raise a TypeError when the function tries to iterate
    edges = [{"source": "a"}]
    with pytest.raises(TypeError):
        find_last_node(None, edges)  # 833ns -> 1.04μs (20.1% slower)


def test_none_inputs_raise_typeerror_for_edges():
    # Passing None for edges should raise a TypeError when the function tries to iterate edges
    nodes = [{"id": "a"}]
    with pytest.raises(TypeError):
        find_last_node(nodes, None)  # 1.33μs -> 583ns (129% faster)


def test_large_chain_returns_last_node_as_sink():
    # Build a chain of 1..1000 where each node i points to i+1. The last node has no outgoing edges.
    n = 1000
    nodes = [{"id": f"node{i}"} for i in range(n)]
    edges = [{"source": f"node{i}", "target": f"node{i+1}"} for i in range(n - 1)]
    # The only sink is the final node, so it must be returned.
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 19.5ms -> 94.4μs (20546% faster)


def test_large_no_edges_returns_first_quickly():
    # With 1000 nodes and zero edges the first node must be returned quickly.
    n = 1000
    nodes = [{"id": i} for i in range(n)]
    edges = []
    codeflash_output = find_last_node(nodes, edges)
    result = codeflash_output  # 1.12μs -> 958ns (17.4% faster)


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

import pytest
from src.algorithms.graph import find_last_node


class TestFindLastNodeBasic:
    """Basic tests for fundamental functionality under normal conditions."""

    def test_single_node_no_edges(self):
        """A single node with no outgoing edges is the sink."""
        nodes = [{"id": "A"}]
        edges = []
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.04μs -> 916ns (13.8% faster)

    def test_two_nodes_linear_chain(self):
        """In a linear chain A→B, node B is the sink (no outgoing edge)."""
        nodes = [{"id": "A"}, {"id": "B"}]
        edges = [{"source": "A", "target": "B"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.62μs -> 1.04μs (56.0% faster)

    def test_three_nodes_linear_chain(self):
        """In A→B→C, node C is the sink (first node with no outgoing edges)."""
        nodes = [{"id": "A"}, {"id": "B"}, {"id": "C"}]
        edges = [{"source": "A", "target": "B"}, {"source": "B", "target": "C"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.88μs -> 1.17μs (60.7% faster)

    def test_multiple_sinks_returns_first(self):
        """When multiple sinks exist, returns the first in nodes order."""
        # Nodes: A→B, C (no outgoing), D (no outgoing)
        # B is first sink in iteration order
        nodes = [{"id": "A"}, {"id": "B"}, {"id": "C"}, {"id": "D"}]
        edges = [{"source": "A", "target": "B"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.54μs -> 1.00μs (54.1% faster)

    def test_node_with_extra_fields(self):
        """Nodes may contain additional fields beyond 'id'."""
        nodes = [{"id": "A", "label": "Start"}, {"id": "B", "label": "End", "data": 42}]
        edges = [{"source": "A", "target": "B"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.46μs -> 1.04μs (40.1% faster)

    def test_edge_with_extra_fields(self):
        """Edges may contain additional fields beyond 'source'."""
        nodes = [{"id": "A"}, {"id": "B"}]
        edges = [{"source": "A", "target": "B", "weight": 1.5, "label": "connects"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.54μs -> 1.04μs (48.0% faster)

    def test_diamond_graph_structure(self):
        """In a diamond graph A→B,C; B→D; C→D, node D is the sink."""
        nodes = [{"id": "A"}, {"id": "B"}, {"id": "C"}, {"id": "D"}]
        edges = [
            {"source": "A", "target": "B"},
            {"source": "A", "target": "C"},
            {"source": "B", "target": "D"},
            {"source": "C", "target": "D"},
        ]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 2.38μs -> 1.25μs (90.0% faster)

    def test_branching_graph_multiple_sinks(self):
        """In A→B; A→C (both B, C are sinks), returns first in order (B)."""
        nodes = [{"id": "A"}, {"id": "B"}, {"id": "C"}]
        edges = [{"source": "A", "target": "B"}, {"source": "A", "target": "C"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.62μs -> 1.08μs (50.0% faster)


class TestFindLastNodeEdgeCases:
    """Edge case tests — empty inputs, boundary values, special scenarios."""

    def test_empty_nodes_list(self):
        """Empty node list returns None."""
        nodes = []
        edges = []
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 708ns -> 792ns (10.6% slower)

    def test_empty_edges_all_nodes_are_sinks(self):
        """With no edges, first node is the sink."""
        nodes = [{"id": "X"}, {"id": "Y"}, {"id": "Z"}]
        edges = []
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.08μs -> 917ns (18.1% faster)

    def test_all_nodes_have_outgoing_edges(self):
        """If every node has an outgoing edge (cycle), returns None."""
        # A→B, B→C, C→A (cycle with no sink)
        nodes = [{"id": "A"}, {"id": "B"}, {"id": "C"}]
        edges = [
            {"source": "A", "target": "B"},
            {"source": "B", "target": "C"},
            {"source": "C", "target": "A"},
        ]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.92μs -> 1.17μs (64.4% faster)

    def test_self_loop_on_single_node(self):
        """A node with a self-loop has an outgoing edge to itself."""
        nodes = [{"id": "A"}]
        edges = [{"source": "A", "target": "A"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.21μs -> 958ns (26.1% faster)

    def test_node_id_is_zero(self):
        """Node IDs can be 0 (falsy value), must use equality not truthiness."""
        nodes = [{"id": 0}, {"id": 1}]
        edges = [{"source": 0, "target": 1}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.58μs -> 1.08μs (46.3% faster)

    def test_node_id_is_empty_string(self):
        """Node IDs can be empty strings."""
        nodes = [{"id": ""}, {"id": "B"}]
        edges = [{"source": "", "target": "B"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.54μs -> 1.04μs (48.0% faster)

    def test_node_id_is_none(self):
        """Node IDs can be None."""
        nodes = [{"id": None}, {"id": "B"}]
        edges = [{"source": None, "target": "B"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.67μs -> 1.08μs (53.8% faster)

    def test_node_id_is_numeric_string(self):
        """Node IDs as strings are distinct from integers."""
        nodes = [{"id": "0"}, {"id": 0}, {"id": "1"}]
        edges = [{"source": "0", "target": "1"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.62μs -> 1.04μs (56.0% faster)

    def test_duplicate_node_ids(self):
        """Duplicate node IDs in list; first occurrence with no outgoing edge is returned."""
        nodes = [{"id": "A"}, {"id": "A"}]
        edges = [{"source": "A", "target": "B"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.42μs -> 1.04μs (36.0% faster)

    def test_duplicate_edges(self):
        """Duplicate edges in list; each is checked independently."""
        nodes = [{"id": "A"}, {"id": "B"}]
        edges = [{"source": "A", "target": "B"}, {"source": "A", "target": "B"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.62μs -> 1.12μs (44.4% faster)

    def test_edge_to_nonexistent_node(self):
        """Edges can reference nodes not in the nodes list; only 'source' matters."""
        nodes = [{"id": "A"}, {"id": "B"}]
        edges = [{"source": "A", "target": "X"}]  # "X" not in nodes
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.50μs -> 1.00μs (50.0% faster)

    def test_single_edge_no_source_match(self):
        """Single edge with source not matching any node ID."""
        nodes = [{"id": "A"}, {"id": "B"}]
        edges = [{"source": "Z", "target": "A"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.21μs -> 958ns (26.2% faster)

    def test_nodes_with_only_id_key(self):
        """Minimal node dictionaries with only 'id' key."""
        nodes = [{"id": "sink"}]
        edges = []
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.04μs -> 916ns (13.8% faster)

    def test_large_id_values(self):
        """Node IDs can be large strings or numbers."""
        nodes = [{"id": "A" * 1000}, {"id": "B" * 1000}]
        edges = [{"source": "A" * 1000, "target": "B" * 1000}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.54μs -> 1.00μs (54.1% faster)

    def test_special_characters_in_node_id(self):
        """Node IDs with special characters."""
        nodes = [{"id": "node@1"}, {"id": "node#2"}, {"id": "node$end"}]
        edges = [{"source": "node@1", "target": "node#2"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.62μs -> 1.12μs (44.4% faster)

    def test_unicode_characters_in_node_id(self):
        """Node IDs with unicode characters."""
        nodes = [{"id": "α"}, {"id": "β"}, {"id": "ω"}]
        edges = [{"source": "α", "target": "β"}]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 1.58μs -> 1.00μs (58.3% faster)


class TestFindLastNodeLargeScale:
    """Large-scale tests — performance and scalability with substantial data."""

    def test_large_linear_chain_1000_nodes(self):
        """Linear chain of 1000 nodes; last node is the sink."""
        nodes = [{"id": i} for i in range(1000)]
        edges = [{"source": i, "target": i + 1} for i in range(999)]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 17.9ms -> 46.0μs (38949% faster)

    def test_large_linear_chain_find_middle_sink(self):
        """Linear chain where nodes after position 500 have no outgoing edges."""
        nodes = [{"id": i} for i in range(1000)]
        edges = [{"source": i, "target": i + 1} for i in range(500)]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 4.49ms -> 24.0μs (18652% faster)

    def test_large_branching_graph_first_node_sink(self):
        """Large graph where first node is sink (no outgoing edges)."""
        nodes = [{"id": 0}] + [{"id": i} for i in range(1, 1000)]
        edges = [{"source": i, "target": i + 1} for i in range(1, 999)]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 37.4μs -> 20.0μs (87.1% faster)

    def test_large_star_graph_center_has_outgoing(self):
        """Star graph: center node branches to 999 leaves."""
        # Center node (0) connects to all others; leaves have no outgoing edges
        nodes = [{"id": 0}] + [{"id": i} for i in range(1, 1000)]
        edges = [{"source": 0, "target": i} for i in range(1, 1000)]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 36.9μs -> 16.5μs (123% faster)

    def test_large_complete_outgoing_all_nodes_connected(self):
        """Every node has at least one outgoing edge (dense graph)."""
        nodes = [{"id": i} for i in range(100)]
        # Each node i connects to node (i+1) % 100, creating a cycle
        edges = [{"source": i, "target": (i + 1) % 100} for i in range(100)]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 199μs -> 6.46μs (2988% faster)

    def test_large_graph_with_many_edges_single_sink(self):
        """Graph with many edges but single clear sink node."""
        nodes = [{"id": i} for i in range(500)]
        # Create edges: each node 0-498 connects to multiple targets
        edges = []
        for i in range(499):
            for j in range(i + 1, min(i + 3, 500)):
                edges.append({"source": i, "target": j})
        # Node 499 has no outgoing edges
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 9.24ms -> 32.8μs (28053% faster)

    def test_large_graph_multiple_sinks_returns_earliest(self):
        """Large graph with multiple sinks; returns first in iteration order."""
        nodes = [{"id": i} for i in range(1000)]
        # Nodes 0-499 have outgoing edges, 500-999 are sinks
        edges = [{"source": i, "target": i + 500} for i in range(500)]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 4.49ms -> 24.0μs (18589% faster)

    def test_large_edges_sparse_nodes(self):
        """Few nodes (10) but many edges between them."""
        nodes = [{"id": i} for i in range(10)]
        edges = []
        for i in range(10):
            for j in range(10):
                if i != j:
                    edges.append({"source": i, "target": j})
        # Every node has outgoing edges
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 18.2μs -> 3.38μs (438% faster)

    def test_performance_1000_nodes_1000_edges(self):
        """Performance test: 1000 nodes, 1000 edges, should complete quickly."""
        nodes = [{"id": i, "label": f"node_{i}"} for i in range(1000)]
        edges = [{"source": i % 500, "target": (i // 2) % 1000} for i in range(1000)]
        # Nodes that appear as sources: 0-499
        # First node not a source in our iteration is 500 (if it's truly not a source)
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 4.48ms -> 33.9μs (13102% faster)

    def test_deeply_nested_graph_structure(self):
        """Tree-like structure with depth 20 and branching factor 5."""
        nodes = [{"id": i} for i in range(1000)]
        edges = []
        # Create a tree where node i connects to nodes 5*i+1 through 5*i+5 (if < 1000)
        for i in range(200):  # 200 internal nodes
            for j in range(1, 6):
                target = 5 * i + j
                if target < 1000:
                    edges.append({"source": i, "target": target})
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 3.56ms -> 23.0μs (15384% faster)

    def test_large_graph_with_isolated_nodes(self):
        """Large graph with isolated nodes (no incoming, no outgoing edges)."""
        nodes = [{"id": i} for i in range(1000)]
        # Only edges between first 100 nodes
        edges = [{"source": i, "target": i + 1} for i in range(99)]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 202μs -> 6.46μs (3030% faster)

    def test_alternating_source_pattern(self):
        """Edges with alternating source pattern."""
        nodes = [{"id": i} for i in range(100)]
        # Odd-indexed nodes connect to even-indexed
        edges = [{"source": i, "target": (i + 1) % 100} for i in range(0, 100, 2)]
        codeflash_output = find_last_node(nodes, edges)
        result = codeflash_output  # 3.54μs -> 2.42μs (46.5% 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-mlfs7qfs and push.