Multigraphs via an edge index and edge properties/metadata

.gitignore

@@ -22,3 +22,4 @@ erl_crash.dump
 # QuickCheck files
 /.eqc-info
 /*.eqc
+.elixir_ls/

README.md

@@ -111,6 +111,97 @@ keeps the initial graph; duplicate identifiers keep the first vertex and labels.
 `Graph.add_vertices/2` continues to treat every item as a vertex, so tuples and
 lists are never implicitly interpreted as labeling instructions.
 
+## Multigraphs
+
+Libgraph supports multigraphs — graphs where multiple edges with different labels can exist between
+the same pair of vertices. When `multigraph: true` is enabled, an edge adjacency index is maintained
+that lets partition queries start from matching endpoint pairs, avoiding scans of unrelated
+edges. Query cost depends on the selected pairs and their parallel labels. The index uses
+additional memory and adds maintenance work to edge mutations.
+
+### Creating a multigraph
+
+```elixir
+g =
+  Graph.new(multigraph: true)
+  |> Graph.add_edges([
+    {:a, :b, label: :uses},
+    {:a, :b, label: :contains},
+    {:b, :c, label: :uses},
+    {:b, :c, label: :owns, weight: 3}
+  ])
+```
+
+### Querying by partition
+
+By default, edges are partitioned by their label (via `Graph.Utils.by_edge_label/1`). You can
+query edges belonging to a specific partition:
+
+```elixir
+# Get only :uses edges
+Graph.edges(g, by: :uses)
+#=> [%Graph.Edge{v1: :a, v2: :b, label: :uses}, %Graph.Edge{v1: :b, v2: :c, label: :uses}]
+
+# Get out edges from :a with label :contains
+Graph.out_edges(g, :a, by: :contains)
+#=> [%Graph.Edge{v1: :a, v2: :b, label: :contains}]
+
+# Filter edges with a predicate
+Graph.edges(g, where: fn edge -> edge.weight > 2 end)
+#=> [%Graph.Edge{v1: :b, v2: :c, label: :owns, weight: 3}]
+```
+
+### Custom partition functions
+
+You can provide a custom `partition_by` function to control how edges are indexed:
+
+```elixir
+g = Graph.new(multigraph: true, partition_by: fn edge -> [edge.weight] end)
+|> Graph.add_edges([{:a, :b, weight: 1}, {:b, :c, weight: 2}])
+
+Graph.edges(g, by: 1)
+#=> [%Graph.Edge{v1: :a, v2: :b, weight: 1}]
+```
+
+### Partition-filtered traversals
+
+BFS, DFS, Dijkstra, A*, and Bellman-Ford all support a `by:` option to restrict traversal to
+edges in specific partitions:
+
+```elixir
+g =
+  Graph.new(multigraph: true)
+  |> Graph.add_edges([
+    {:a, :b, label: :fast, weight: 1},
+    {:a, :c, label: :slow, weight: 10},
+    {:b, :d, label: :fast, weight: 1},
+    {:c, :d, label: :slow, weight: 1}
+  ])
+
+# Shortest path using only :fast edges
+Graph.dijkstra(g, :a, :d, by: :fast)
+#=> [:a, :b, :d]
+
+# BFS following only :fast edges
+Graph.Reducers.Bfs.map(g, & &1, by: :fast)
+#=> [:a, :b, :d]
+```
+
+### Edge properties
+
+Edges support an arbitrary `properties` map for storing additional metadata. The existing
+`%{label => weight}` edge records are retained; properties are stored separately in
+`edge_properties` only when populated:
+
+```elixir
+g = Graph.new()
+|> Graph.add_edge(:a, :b, label: :link, properties: %{color: "red", style: :dashed})
+
+[edge] = Graph.edges(g)
+edge.properties
+#=> %{color: "red", style: :dashed}
+```
+
 ## Rationale
 
 The original motivation for me to start working on this library is the fact that `:digraph` requires a

bench/multigraph.exs

@@ -0,0 +1,53 @@
+defmodule MultigraphBench.Helpers do
+  def build_graphs(num_vertices, num_edges, num_labels) do
+    labels = Enum.map(1..num_labels, fn i -> :"label_#{i}" end)
+
+    edges =
+      for _ <- 1..num_edges do
+        v1 = :rand.uniform(num_vertices)
+        v2 = :rand.uniform(num_vertices)
+        label = Enum.random(labels)
+        {v1, v2, label: label, weight: :rand.uniform(100)}
+      end
+
+    plain =
+      Enum.reduce(edges, Graph.new(), fn {v1, v2, opts}, g ->
+        Graph.add_edge(g, v1, v2, opts)
+      end)
+
+    multi =
+      Enum.reduce(edges, Graph.new(multigraph: true), fn {v1, v2, opts}, g ->
+        Graph.add_edge(g, v1, v2, opts)
+      end)
+
+    target_label = Enum.random(labels)
+    some_vertex = :rand.uniform(num_vertices)
+
+    {plain, multi, target_label, some_vertex}
+  end
+end
+
+alias MultigraphBench.Helpers
+
+Benchee.run(
+  %{
+    "scan all edges + filter (no multigraph)" => fn {g_plain, _g_multi, target_label, _v} ->
+      g_plain |> Graph.edges() |> Enum.filter(fn e -> e.label == target_label end)
+    end,
+    "indexed lookup (multigraph by:)" => fn {_g_plain, g_multi, target_label, _v} ->
+      Graph.edges(g_multi, by: target_label)
+    end,
+    "scan out_edges + filter (no multigraph)" => fn {g_plain, _g_multi, target_label, v} ->
+      g_plain |> Graph.out_edges(v) |> Enum.filter(fn e -> e.label == target_label end)
+    end,
+    "indexed out_edges (multigraph by:)" => fn {_g_plain, g_multi, target_label, v} ->
+      Graph.out_edges(g_multi, v, by: target_label)
+    end
+  },
+  inputs: %{
+    "1k vertices, 5k edges, 10 labels" => Helpers.build_graphs(1_000, 5_000, 10),
+    "10k vertices, 50k edges, 50 labels" => Helpers.build_graphs(10_000, 50_000, 50)
+  },
+  time: 10,
+  memory_time: 5
+)

bench/multigraph_creation.exs

@@ -0,0 +1,35 @@
+defmodule MultigraphCreationBench.Helpers do
+  def build_edges(size) do
+    labels = Enum.map(1..10, fn i -> :"label_#{i}" end)
+
+    for i <- 1..size do
+      v1 = :rand.uniform(div(size, 5))
+      v2 = :rand.uniform(div(size, 5))
+      label = Enum.random(labels)
+      {v1, v2, label: label, weight: :rand.uniform(100)}
+    end
+  end
+end
+
+alias MultigraphCreationBench.Helpers
+
+Benchee.run(
+  %{
+    "plain graph (no index)" => fn edges ->
+      Enum.reduce(edges, Graph.new(), fn {v1, v2, opts}, g ->
+        Graph.add_edge(g, v1, v2, opts)
+      end)
+    end,
+    "multigraph (indexed)" => fn edges ->
+      Enum.reduce(edges, Graph.new(multigraph: true), fn {v1, v2, opts}, g ->
+        Graph.add_edge(g, v1, v2, opts)
+      end)
+    end
+  },
+  inputs: %{
+    "10k edges" => Helpers.build_edges(10_000),
+    "100k edges" => Helpers.build_edges(100_000)
+  },
+  time: 10,
+  memory_time: 5
+)

bench/multigraph_memory.exs

@@ -0,0 +1,49 @@
+defmodule MultigraphMemoryBench.Helpers do
+  def build_graphs(num_edges, num_labels) do
+    num_vertices = div(num_edges, 5)
+    labels = Enum.map(1..num_labels, fn i -> :"label_#{i}" end)
+
+    edges =
+      for _ <- 1..num_edges do
+        v1 = :rand.uniform(num_vertices)
+        v2 = :rand.uniform(num_vertices)
+        label = Enum.random(labels)
+        {v1, v2, label: label, weight: :rand.uniform(100)}
+      end
+
+    plain =
+      Enum.reduce(edges, Graph.new(), fn {v1, v2, opts}, g ->
+        Graph.add_edge(g, v1, v2, opts)
+      end)
+
+    multi =
+      Enum.reduce(edges, Graph.new(multigraph: true), fn {v1, v2, opts}, g ->
+        Graph.add_edge(g, v1, v2, opts)
+      end)
+
+    {plain, multi}
+  end
+end
+
+alias MultigraphMemoryBench.Helpers
+
+IO.puts("Multigraph Memory Overhead Report")
+IO.puts("==================================\n")
+
+for {name, {size, labels}} <- [
+      {"1k edges / 5 labels", {1_000, 5}},
+      {"10k edges / 10 labels", {10_000, 10}},
+      {"10k edges / 100 labels", {10_000, 100}},
+      {"100k edges / 50 labels", {100_000, 50}}
+    ] do
+  {plain, multi} = Helpers.build_graphs(size, labels)
+
+  plain_info = Graph.info(plain)
+  multi_info = Graph.info(multi)
+
+  ratio = multi_info.size_in_bytes / plain_info.size_in_bytes
+
+  IO.puts(
+    "#{name}: plain=#{plain_info.size_in_bytes}B, multi=#{multi_info.size_in_bytes}B, ratio=#{Float.round(ratio, 2)}x"
+  )
+end

bench/partition_indexes.exs

@@ -0,0 +1,101 @@
+# Run the identical script against baseline and optimized dependencies.
+# Compilation/setup and explicit GC are outside timing. No concurrent runs.
+# Example: ERL_FLAGS='+S 4:4' mix run bench/partition_indexes.exs 128 512 2048 8192
+# Query loops hold selected degree at one while unrelated history grows.
+# A single relabel starts from the same immutable graph on every repetition.
+# consume_all changes n distinct edges and retains the resulting history.
+defmodule Graph.Bench.PartitionIndexes do
+
+  @queries 500
+
+  def run(sizes) do
+    IO.puts(
+      "operation,n,trial,iterations,microseconds,reductions,post_gc_bytes,result_flat_words"
+    )
+
+    for n <- sizes do
+      graph = build(n)
+
+      jobs = [
+        {"incoming_partition", @queries,
+         fn -> repeat(@queries, fn -> Graph.in_edges(graph, :hub, by: :fan_in) end) end},
+        {"outgoing_partition", @queries,
+         fn -> repeat(@queries, fn -> Graph.out_edges(graph, :hub, by: :flow) end) end},
+        {"incident_partition", @queries,
+         fn -> repeat(@queries, fn -> Graph.edges(graph, :hub, by: :flow) end) end},
+        {"single_relabel", @queries,
+         fn ->
+           repeat(@queries, fn ->
+             Graph.update_labelled_edge(graph, {:pending, 1}, :hub, :runnable, label: :ran)
+           end)
+         end},
+        {"consume_all", n,
+         fn ->
+           Enum.reduce(1..n, graph, fn i, acc ->
+             Graph.update_labelled_edge(acc, {:pending, i}, :hub, :runnable, label: :ran)
+           end)
+         end}
+      ]
+
+      for {name, iterations, work} <- jobs do
+        isolated(work)
+
+        for trial <- 1..3 do
+          {time, reductions, memory, words} = isolated(work)
+          IO.puts(Enum.join([name, n, trial, iterations, time, reductions, memory, words], ","))
+        end
+      end
+    end
+  end
+
+  defp build(n) do
+    Enum.reduce(1..n, Graph.new(multigraph: true), fn i, g ->
+      g
+      |> Graph.add_edge({:history, i}, :hub, label: :ran)
+      |> Graph.add_edge(:hub, {:output, i}, label: :produced)
+      |> Graph.add_edge({:pending, i}, :hub, label: :runnable)
+    end)
+    |> Graph.add_edge(:source, :hub, label: :fan_in)
+    |> Graph.add_edge(:hub, :sink, label: :flow)
+  end
+
+  defp repeat(n, work), do: Enum.reduce(1..n, nil, fn _, _ -> work.() end)
+
+  defp isolated(work) do
+    caller = self()
+    ref = make_ref()
+
+    {pid, monitor} =
+      spawn_monitor(fn ->
+        :erlang.garbage_collect()
+        {:reductions, before} = Process.info(self(), :reductions)
+        {time, result} = :timer.tc(work)
+        {:reductions, after_work} = Process.info(self(), :reductions)
+        :erlang.garbage_collect()
+        {:memory, memory} = Process.info(self(), :memory)
+        words = :erts_debug.flat_size(result)
+        send(caller, {ref, time, after_work - before, memory, words})
+      end)
+
+    receive do
+      {^ref, time, reductions, memory, words} ->
+        Process.demonitor(monitor, [:flush])
+        {time, reductions, memory, words}
+
+      {:DOWN, ^monitor, :process, ^pid, reason} ->
+        raise "probe failed: #{inspect(reason)}"
+    after
+      120_000 ->
+        Process.exit(pid, :kill)
+        raise "probe exceeded 120 seconds"
+    end
+  end
+end
+
+sizes =
+  case System.argv() do
+    [] -> [128, 512, 2048, 8192]
+    args -> Enum.map(args, &String.to_integer/1)
+  end
+
+Graph.Bench.PartitionIndexes.run(sizes)