Fix O(n) shape removal from the subquery index (#4594)

## Summary

Removing a shape from the subquery index in the where-clause filter had
become **O(n)** in the number of shapes — and **O(shapes × values)**
once subqueries were seeded. Shape removal runs on the replication path
and blocks it, so slow removal caused WAL lag in production (the
"Materializer shape invalidated" path). This restores **O(1)** removal
(as in v1.5).

Fixes #4279

## Problem

The subquery index is a single ETS `:bag` table. Several deletions used
`:ets.match_delete/2` with a **wildcard in the key position** (e.g.
`{:membership, handle, :_, :_}`, `{:node_positive_member, node, :_}`). A
`:bag` (hash) table can only use its key index when the key is **fully
bound**, so a partially-bound key forces a **full-table scan** that
grows with the number of shapes. The worst offender was
`delete_node_members`, which scanned every shape's per-`(node, value)`
routing row on each removal — O(shapes × values).

## Solution

Switch the table from `:bag` to **`:ordered_set`** and lift the
discriminating fields (`value`, `shape`, `branch`, `next_cond`) out of
the value tuples into the **keys**. In an `:ordered_set`, a delete whose
key has a bound **prefix** is range-limited (not a full scan), and a
point delete by a fully-bound key is O(log n). Concretely:

- `unregister_shape/2` replaces its full-table `match_delete` scans with
prefix-bounded `select_delete`s.
- `remove_shape/5` point-deletes the shape's node rows, and **derives**
its `node_*_member` deletions from the removed shape's own membership
rows (scoped by `subquery_ref`), applying exact point-deletes. This
touches only the removed shape's rows — never the other shapes sharing a
value.

Removal is now **O(V_shape · log n)** — proportional to the removed
shape's own view size, independent of the total number of shapes, the
number of shapes on a node, and the number of shapes sharing a value.

This is a pure internal storage change: every public function and all
externally-observable behavior is preserved. The per-change routing hot
path (`affected_shapes`) and the exact-evaluation `member?` path are
equal-or-faster (the `:ordered_set` prefix `select` projects in C);
table memory is roughly flat (measured slightly lower in a
microbenchmark).

> Note: a dedicated **memory** reduction (interning / row-shrinking) is
intentionally **out of scope** here — this PR is the removal-complexity
fix only, kept minimal and behavior-preserving. Memory can be tackled
separately.

## Test plan

- [x] `:performance` **flatness** tests prove removal is independent of
each dimension — total shapes, shapes-on-node, and shapes-per-value — by
measuring removal at two well-separated sizes and asserting the delta
stays within noise (observed delta 0 at 1k vs 20k).
- [x] **Positive control**: removal cost *does* scale with the removed
shape's own view size (pins the complexity class as O(V_shape), so a
regression can't hide under a flat budget).
- [x] **No-orphan guard**: a full `Filter.remove_shape` of a seeded
shape, and draining a shared node one-by-one, leaves the subquery index
byte-identical to its pre-add state.
- [x] Existing `subquery_index` / `filter` / `consumer` /
`shape_log_collector` suites green; full `mix test` green.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: robacourt

.changeset/subquery-index-o1-removal.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Restore O(1) removal of subquery shapes from the where-clause filter. The v1.6 subquery index removed shapes with full-table ETS scans, so removing a shape cost O(total shapes × values) and blocked replication processing — causing WAL lag when many seeded subquery shapes were present. The index now uses an `:ordered_set` with prefix-bounded deletes, making shape removal independent of the total number of shapes, the number of shapes on a node, and the number of shapes sharing a value.

packages/sync-service/lib/electric/shapes/filter/indexes/subquery_index.ex

@@ -47,10 +47,10 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
   def new(opts \\ []) do
     case Keyword.get(opts, :stack_id) do
       nil ->
-        :ets.new(:subquery_index, [:bag, :public])
+        :ets.new(:subquery_index, [:ordered_set, :public])
 
       stack_id ->
-        :ets.new(table_name(stack_id), [:bag, :public, :named_table])
+        :ets.new(table_name(stack_id), [:ordered_set, :public, :named_table])
     end
   end
 
@@ -94,10 +94,10 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
   """
   @spec unregister_shape(t(), term()) :: :ok
   def unregister_shape(table, shape_handle) do
-    :ets.match_delete(table, {{:membership, shape_handle, :_, :_}, true})
-    :ets.match_delete(table, {{:polarity, shape_handle, :_}, :_})
-    :ets.match_delete(table, {{:shape_node, shape_handle}, :_})
-    :ets.match_delete(table, {{:shape_dep_node, shape_handle, :_}, :_})
+    delete_by_key_prefix(table, {:membership, shape_handle, :_, :_})
+    delete_by_key_prefix(table, {:polarity, shape_handle, :_})
+    delete_by_key_prefix(table, {:shape_node, shape_handle, :_, :_})
+    delete_by_key_prefix(table, {:shape_dep_node, shape_handle, :_, :_, :_})
     :ets.delete(table, {:fallback, shape_handle})
     :ok
   end
@@ -130,27 +130,27 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
 
     :ets.insert(
       table,
-      {{:node_shape, node_id},
-       {shape_id, optimisation.dep_index, optimisation.polarity, next_condition_id, branch_key}}
+      {{:node_shape, node_id, shape_id, branch_key},
+       {optimisation.dep_index, optimisation.polarity, next_condition_id}}
     )
 
     if optimisation.polarity == :negated do
-      :ets.insert(table, {{:node_negated_shape, node_id}, {shape_id, next_condition_id}})
+      :ets.insert(table, {{:node_negated_shape, node_id, shape_id, next_condition_id}, true})
     end
 
     :ets.insert(
       table,
-      {{:shape_node, shape_id},
-       {node_id, optimisation.dep_index, optimisation.polarity, next_condition_id, branch_key}}
+      {{:shape_node, shape_id, node_id, branch_key},
+       {optimisation.dep_index, optimisation.polarity, next_condition_id}}
     )
 
     :ets.insert(
       table,
-      {{:shape_dep_node, shape_id, optimisation.dep_index},
-       {node_id, optimisation.polarity, next_condition_id, branch_key}}
+      {{:shape_dep_node, shape_id, optimisation.dep_index, node_id, branch_key},
+       {optimisation.polarity, next_condition_id}}
     )
 
-    :ets.insert(table, {{:node_fallback, node_id}, {shape_id, next_condition_id}})
+    :ets.insert(table, {{:node_fallback, node_id, shape_id, next_condition_id}, true})
     :ok
   end
 
@@ -181,31 +181,24 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
             branch_key
           )
 
-        :ets.match_delete(
+        delete_node_members(
           table,
-          {{:node_shape, node_id}, {shape_id, dep_index, polarity, next_condition_id, branch_key}}
+          node_id,
+          shape_id,
+          polarity,
+          next_condition_id,
+          optimisation.subquery_ref
         )
 
+        :ets.delete(table, {:node_shape, node_id, shape_id, branch_key})
+
         if polarity == :negated do
-          :ets.match_delete(
-            table,
-            {{:node_negated_shape, node_id}, {shape_id, next_condition_id}}
-          )
+          :ets.delete(table, {:node_negated_shape, node_id, shape_id, next_condition_id})
         end
 
-        :ets.match_delete(
-          table,
-          {{:shape_node, shape_id}, {node_id, dep_index, polarity, next_condition_id, branch_key}}
-        )
-
-        :ets.match_delete(
-          table,
-          {{:shape_dep_node, shape_id, dep_index},
-           {node_id, polarity, next_condition_id, branch_key}}
-        )
-
-        :ets.match_delete(table, {{:node_fallback, node_id}, {shape_id, next_condition_id}})
-        delete_node_members(table, node_id, shape_id, polarity, next_condition_id)
+        :ets.delete(table, {:node_fallback, node_id, shape_id, next_condition_id})
+        :ets.delete(table, {:shape_node, shape_id, node_id, branch_key})
+        :ets.delete(table, {:shape_dep_node, shape_id, dep_index, node_id, branch_key})
 
         if node_empty?(table, node_id) do
           :ets.delete(table, {:node_meta, node_id})
@@ -237,7 +230,7 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
 
     for {node_id, _dep_index, _polarity, _next_condition_id, _branch_key} <-
           nodes_for_shape(table, shape_handle) do
-      :ets.match_delete(table, {{:node_fallback, node_id}, {shape_handle, :_}})
+      delete_by_key_prefix(table, {:node_fallback, node_id, shape_handle, :_})
     end
 
     :ok
@@ -250,19 +243,8 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
   def add_value(table, shape_handle, subquery_ref, dep_index, value) do
     for {node_id, polarity, next_condition_id, _branch_key} <-
           nodes_for_shape_dependency(table, shape_handle, dep_index) do
-      case polarity do
-        :positive ->
-          :ets.insert(
-            table,
-            {{:node_positive_member, node_id, value}, {shape_handle, next_condition_id}}
-          )
-
-        :negated ->
-          :ets.insert(
-            table,
-            {{:node_negated_member, node_id, value}, {shape_handle, next_condition_id}}
-          )
-      end
+      tag = if polarity == :positive, do: :node_positive_member, else: :node_negated_member
+      :ets.insert(table, {{tag, node_id, value, shape_handle, next_condition_id}, true})
     end
 
     :ets.insert(table, {{:membership, shape_handle, subquery_ref, value}, true})
@@ -276,19 +258,8 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
   def remove_value(table, shape_handle, subquery_ref, dep_index, value) do
     for {node_id, polarity, next_condition_id, _branch_key} <-
           nodes_for_shape_dependency(table, shape_handle, dep_index) do
-      case polarity do
-        :positive ->
-          :ets.match_delete(
-            table,
-            {{:node_positive_member, node_id, value}, {shape_handle, next_condition_id}}
-          )
-
-        :negated ->
-          :ets.match_delete(
-            table,
-            {{:node_negated_member, node_id, value}, {shape_handle, next_condition_id}}
-          )
-      end
+      tag = if polarity == :positive, do: :node_positive_member, else: :node_negated_member
+      :ets.delete(table, {tag, node_id, value, shape_handle, next_condition_id})
     end
 
     :ets.delete(table, {:membership, shape_handle, subquery_ref, value})
@@ -305,16 +276,15 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
     candidates =
       case evaluate_node_lhs(table, node_id, record) do
         {:ok, typed_value} ->
-          positive =
-            values_for_key(table, {:node_positive_member, node_id, typed_value}) |> MapSet.new()
+          positive = members_for(table, :node_positive_member, node_id, typed_value)
 
           negated =
             MapSet.difference(
-              values_for_key(table, {:node_negated_shape, node_id}) |> MapSet.new(),
-              values_for_key(table, {:node_negated_member, node_id, typed_value}) |> MapSet.new()
+              negated_shapes_for(table, node_id),
+              members_for(table, :node_negated_member, node_id, typed_value)
             )
 
-          fallback = values_for_key(table, {:node_fallback, node_id}) |> MapSet.new()
+          fallback = fallback_for(table, node_id)
 
           positive
           |> MapSet.union(negated)
@@ -407,61 +377,104 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
     end
   end
 
-  defp delete_node_members(table, node_id, shape_id, polarity, next_condition_id) do
-    case polarity do
-      :positive ->
-        :ets.match_delete(
-          table,
-          {{:node_positive_member, node_id, :_}, {shape_id, next_condition_id}}
-        )
+  # Range-bounded delete of every row whose key matches `key_pattern` (a tuple with a
+  # bound prefix and `:_` wildcards in trailing positions). On an :ordered_set this is
+  # O(log n + matched), never a full scan.
+  defp delete_by_key_prefix(table, key_pattern) do
+    :ets.select_delete(table, [{{key_pattern, :_}, [], [true]}])
+  end
 
-      :negated ->
-        :ets.match_delete(
-          table,
-          {{:node_negated_member, node_id, :_}, {shape_id, next_condition_id}}
-        )
+  # Delete this shape's node-local member rows for this node by enumerating the shape's
+  # own values (scoped to the node's subquery_ref) from its membership rows and
+  # point-deleting each. O(V_node · log n); touches only this shape's rows.
+  #
+  # INVARIANT (the safety of this approach rests on it): a node-member row is always a
+  # subset of the shape's membership values — `add_value/5` and `remove_value/5` write
+  # and remove the membership row alongside the node-member rows — so enumerating
+  # membership finds every node-member row to delete. This requires membership rows to
+  # outlive node removal, and to not be mutated concurrently:
+  #   * Ordering: `Filter.remove_shape/2` runs this (via `remove_shape/5`) BEFORE
+  #     `unregister_shape/2` deletes the `:membership` rows, so they are still present.
+  #   * No concurrent writer: the shape's consumer is the only process that writes
+  #     membership/node-member rows, and it is stopped synchronously before
+  #     `Filter.remove_shape/2` runs (`ShapeCleaner.remove_shape_immediate/3`), so no
+  #     `add_value`/`remove_value` can race this.
+  # The "no orphan rows" test guards exactly this path. Reordering cleanup to remove
+  # from the filter before stopping the consumer would reintroduce orphaned node-member
+  # rows.
+  defp delete_node_members(table, node_id, shape_id, polarity, next_condition_id, subquery_ref) do
+    tag =
+      case polarity do
+        :positive -> :node_positive_member
+        :negated -> :node_negated_member
+      end
+
+    values =
+      :ets.select(table, [
+        {{{:membership, shape_id, subquery_ref, :"$1"}, :_}, [], [:"$1"]}
+      ])
+
+    for value <- values do
+      :ets.delete(table, {tag, node_id, value, shape_id, next_condition_id})
     end
+
+    :ok
+  end
+
+  defp members_for(table, tag, node_id, value) do
+    :ets.select(table, [
+      {{{tag, node_id, value, :"$1", :"$2"}, :_}, [], [{{:"$1", :"$2"}}]}
+    ])
+    |> MapSet.new()
+  end
+
+  defp negated_shapes_for(table, node_id) do
+    :ets.select(table, [
+      {{{:node_negated_shape, node_id, :"$1", :"$2"}, :_}, [], [{{:"$1", :"$2"}}]}
+    ])
+    |> MapSet.new()
+  end
+
+  defp fallback_for(table, node_id) do
+    :ets.select(table, [
+      {{{:node_fallback, node_id, :"$1", :"$2"}, :_}, [], [{{:"$1", :"$2"}}]}
+    ])
+    |> MapSet.new()
+  end
+
+  defp all_node_shapes(table, node_id) do
+    :ets.select(table, [
+      {{{:node_shape, node_id, :"$1", :_}, {:_, :_, :"$2"}}, [], [{{:"$1", :"$2"}}]}
+    ])
+    |> MapSet.new()
   end
 
   defp nodes_for_shape(table, shape_handle) do
-    table
-    |> :ets.lookup({:shape_node, shape_handle})
-    |> Enum.map(&elem(&1, 1))
+    :ets.select(table, [
+      {{{:shape_node, shape_handle, :"$1", :"$2"}, {:"$3", :"$4", :"$5"}}, [],
+       [{{:"$1", :"$3", :"$4", :"$5", :"$2"}}]}
+    ])
   end
 
   defp nodes_for_shape_dependency(table, shape_handle, dep_index) do
-    table
-    |> :ets.lookup({:shape_dep_node, shape_handle, dep_index})
-    |> Enum.map(&elem(&1, 1))
+    :ets.select(table, [
+      {{{:shape_dep_node, shape_handle, dep_index, :"$1", :"$2"}, {:"$3", :"$4"}}, [],
+       [{{:"$1", :"$3", :"$4", :"$2"}}]}
+    ])
   end
 
   defp node_shape_entry_for_shape(table, shape_id, node_id, branch_key) do
-    table
-    |> nodes_for_shape(shape_id)
-    |> Enum.find_value(fn
-      {^node_id, dep_index, polarity, next_condition_id, ^branch_key} ->
-        {dep_index, polarity, next_condition_id}
-
-      _ ->
-        nil
-    end)
+    case :ets.lookup(table, {:shape_node, shape_id, node_id, branch_key}) do
+      [{_, {dep_index, polarity, next_condition_id}}] -> {dep_index, polarity, next_condition_id}
+      [] -> nil
+    end
   end
 
   defp node_empty?(table, node_id) do
-    :ets.lookup(table, {:node_shape, node_id}) == []
-  end
-
-  defp all_node_shapes(table, node_id) do
-    table
-    |> :ets.lookup({:node_shape, node_id})
-    |> Enum.reduce(MapSet.new(), fn
-      {{:node_shape, ^node_id}, {shape_id, _dep_index, _polarity, next_condition_id, _branch_key}},
-      acc ->
-        MapSet.put(acc, {shape_id, next_condition_id})
-
-      _, acc ->
-        acc
-    end)
+    case :ets.select(table, [{{{:node_shape, node_id, :_, :_}, :_}, [], [true]}], 1) do
+      :"$end_of_table" -> true
+      _ -> false
+    end
   end
 
   defp evaluate_node_lhs(table, node_id, record) do
@@ -485,12 +498,6 @@ defmodule Electric.Shapes.Filter.Indexes.SubqueryIndex do
     end
   end
 
-  defp values_for_key(table, key) do
-    table
-    |> :ets.lookup(key)
-    |> Enum.map(&elem(&1, 1))
-  end
-
   defp shape_ready?(table, shape_handle) do
     not fallback?(table, shape_handle)
   end