feat(electric): Proposed speedup of Sent Rows Graph popping method

components/electric/lib/electric/replication/shapes/sent_rows_graph.ex

@@ -96,9 +96,9 @@ defmodule Electric.Replication.Shapes.SentRowsGraph do
             end)
             |> Enum.split_with(&predicate.(elem(&1, 2)))
             |> case do
-              {_all_edges, []} ->
+              {new_edges, []} ->
                 # If all incoming edges match the request ID, we'll pop the vertex
-                {:next, {edges, [v | vertices]}}
+                {:next, {new_edges ++ edges, [v | vertices]}}
 
               {new_edges, _rest} ->
                 # If some incoming edges are unaffected, we'll pop the edges explicitly
@@ -108,10 +108,18 @@ defmodule Electric.Replication.Shapes.SentRowsGraph do
         fn meta -> any_key_matches_predicate?(meta, predicate) end
       )
 
+    # Remove all edges relating to the request IDs from the graph
     graph =
       edges
       |> Enum.reduce(graph, fn {v1, v2, label}, acc -> Graph.delete_edge(acc, v1, v2, label) end)
-      |> Graph.delete_vertices(vertices)
+
+    # Retain the maximally connected subgraph that does not contain the
+    # vertices that have been popped
+    vertices_to_keep =
+      MapSet.difference(MapSet.new(Graph.vertices(graph)), MapSet.new(vertices))
+      |> MapSet.to_list()
+
+    graph = Graph.subgraph(graph, vertices_to_keep)
 
     {vertices, graph}
   end

components/electric/test/electric/replication/shapes/sent_rows_graph_test.exs

@@ -1,5 +1,47 @@
 defmodule Electric.Replication.Shapes.SentRowsGraphTest do
   use ExUnit.Case, async: true
+  alias Electric.Replication.Shapes.SentRowsGraph
 
   doctest Electric.Replication.Shapes.SentRowsGraph, import: true
+
+  describe "pop_by_request_ids/3" do
+    test "should return the popped vertices and the new graph" do
+      graph_init =
+        Graph.new()
+        |> Graph.add_edge(:root, :v1, label: {"r1", "l1"})
+        |> Graph.add_edge(:root, :v1, label: {"r2", "l1"})
+        |> Graph.add_edge(:v1, :v2, label: {"r1", "l2"})
+
+      assert {[:v2], %Graph{} = graph_new} =
+               SentRowsGraph.pop_by_request_ids(graph_init, "r1", root_vertex: :root)
+
+      assert [:v1, :root] = Graph.vertices(graph_new)
+      assert 1 = Graph.num_edges(graph_new)
+      assert Graph.edge(graph_new, :root, :v1, {"r2", "l1"})
+    end
+
+    test "should be able to pop list of requests" do
+      graph_init =
+        Graph.new()
+        |> Graph.add_edge(:root, :v1, label: {"r1", "l1"})
+        |> Graph.add_edge(:root, :v1, label: {"r2", "l1"})
+        |> Graph.add_edge(:v1, :v2, label: {"r1", "l2"})
+
+      assert {[:v2, :v1], %Graph{} = graph_new} =
+               SentRowsGraph.pop_by_request_ids(graph_init, ["r1", "r2"], root_vertex: :root)
+
+      assert [:root] = Graph.vertices(graph_new)
+      assert 0 = Graph.num_edges(graph_new)
+    end
+
+    test "should be able to pop empty graph" do
+      graph_init = Graph.new()
+
+      assert {[], %Graph{} = graph_new} =
+               SentRowsGraph.pop_by_request_ids(graph_init, ["r1", "r2"])
+
+      assert [] = Graph.vertices(graph_new)
+      assert 0 = Graph.num_edges(graph_new)
+    end
+  end
 end