feat(sync-service): index for array contains (#2359)

This PR seeks to address trigger.dev's WAL lag issues by adding two
optimisations to where clause filtering:
- Allow multiple conditions in a where clause to be optimised (not just
one)
- Optimise where clauses that have a condition in the form 'array_field
@> array_const'

## Allow multiple conditions in a where clause to be optimised (not just
one)

This feature alone should halve the processing time for trigger.dev as
they have many where clauses in the form `WHERE runtimeEnvironmentId = ?
AND batchId = ?` where if a change matched the `runtimeEnvironmentId` it
would then have to iterate through the `batchId`s. Now it doesn't - the
`batchId` condition is indexed as well.

## Optimise where clauses that have a condition in the form 'array_field
@> array_const'

Trigger.dev's other problematic where clauses are in the form `WHERE
runtimeEnvironmentId = ? AND runTags @> ?` and this feature optimises
the `@>` operation. The algorithm for this optimisation can be seen in
the [InclusionIndex
module](https://github.com/electric-sql/electric/pull/2359/files#diff-44c9f3ec68658f9ea6690a7608b99ad0463d7dcc8f1c95e9a2617e1d3d30a3d2).

Comparing the performance of this index against the current method shows
quite a dramatic difference:
<img width="707" alt="Screenshot 2025-02-19 at 19 42 59"
src="https://github.com/user-attachments/assets/ea9de81c-50e7-45d1-a81f-d5a21c36daeb"
/>

Here you can see it performs well with 100k shapes (which for
trigger.dev would be 100k shapes with the same `runtimeEnvironmentId`,
in other words 100k shapes per user)
<img width="692" alt="Screenshot 2025-02-19 at 19 43 47"
src="https://github.com/user-attachments/assets/e2fde600-6c0f-4606-ac93-18daa8e50438"
/>

Both of the charts above are based on a shape array size of 3 and a
change array size of 10 which is typical for trigger.dev.

The algorithm seems to scale roughly linearly with change array size:
<img width="714" alt="Screenshot 2025-02-24 at 16 12 34"
src="https://github.com/user-attachments/assets/ce1703c1-9eea-4d19-bc44-8c92615f8da4"
/>

The algorithm seems to scale well with shape array size for a fixed
change array size (10):
<img width="714" alt="Screenshot 2025-02-24 at 16 14 35"
src="https://github.com/user-attachments/assets/fcd6f956-8c01-4580-8e55-4b0eeb17e86f"
/>

Author: robacourt

.changeset/mean-lies-kiss.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Allow multiple conditions in a where clause to be optimised (not just one)

.changeset/twelve-jeans-admire.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Optimise where clauses that have a condition in the form 'array_field @> array_const'

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

@@ -7,7 +7,7 @@ defmodule Electric.Shapes.Filter do
 
 
   The `Filter` module keeps track of what tables are referenced by the shapes and changes and delegates
-  the table specific logic to the `Filter.Table` module.
+  the table specific logic to the `Filter.WhereCondition` module.
   """
 
   alias Electric.Replication.Changes.DeletedRecord
@@ -17,7 +17,7 @@ defmodule Electric.Shapes.Filter do
   alias Electric.Replication.Changes.TruncatedRelation
   alias Electric.Replication.Changes.UpdatedRecord
   alias Electric.Shapes.Filter
-  alias Electric.Shapes.Filter.Table
+  alias Electric.Shapes.Filter.WhereCondition
   alias Electric.Shapes.Shape
 
   require Logger
@@ -45,9 +45,9 @@ defmodule Electric.Shapes.Filter do
         Map.update(
           tables,
           shape.root_table,
-          Table.add_shape(Table.new(shape.root_table), {shape_id, shape}),
-          fn table ->
-            Table.add_shape(table, {shape_id, shape})
+          WhereCondition.add_shape(WhereCondition.new(), {shape_id, shape}, shape.where),
+          fn condition ->
+            WhereCondition.add_shape(condition, {shape_id, shape}, shape.where)
           end
         )
     }
@@ -61,10 +61,10 @@ defmodule Electric.Shapes.Filter do
     %Filter{
       tables:
         tables
-        |> Enum.map(fn {table_name, table} ->
-          {table_name, Table.remove_shape(table, shape_id)}
+        |> Enum.map(fn {table_name, condition} ->
+          {table_name, WhereCondition.remove_shape(condition, shape_id)}
         end)
-        |> Enum.reject(fn {_table, table} -> Table.empty?(table) end)
+        |> Enum.reject(fn {_table, condition} -> WhereCondition.empty?(condition) end)
         |> Map.new()
     }
   end
@@ -134,14 +134,17 @@ defmodule Electric.Shapes.Filter do
 
   defp shapes_affected_by_record(filter, table_name, record) do
     case Map.get(filter.tables, table_name) do
-      nil -> MapSet.new()
-      table -> Table.affected_shapes(table, record)
+      nil ->
+        MapSet.new()
+
+      condition ->
+        WhereCondition.affected_shapes(condition, record)
     end
   end
 
   defp all_shapes(%Filter{} = filter) do
-    for {_table, table} <- filter.tables,
-        {shape_id, shape} <- Table.all_shapes(table),
+    for {_table, condition} <- filter.tables,
+        {shape_id, shape} <- WhereCondition.all_shapes(condition),
         into: %{} do
       {shape_id, shape}
     end
@@ -150,7 +153,7 @@ defmodule Electric.Shapes.Filter do
   defp all_shapes_for_table(%Filter{} = filter, table_name) do
     case Map.get(filter.tables, table_name) do
       nil -> %{}
-      table -> Table.all_shapes(table)
+      condition -> WhereCondition.all_shapes(condition)
     end
   end
 end

packages/sync-service/lib/electric/shapes/filter/index.ex

@@ -1,89 +1,26 @@
 defmodule Electric.Shapes.Filter.Index do
   @moduledoc """
-  Responsible for knowing which shapes are affected by a change to a specific field.
+  Efficiently finds shapes that are affected by a change, specifically for a particular operation in where clause.
 
-  The `%Table{}` struct contains `values` a map of values for a specific field to shapes that are affected by that field value.
-  This acts as an index for the shapes, providing a fast way to know which shapes have been affected without having to
-  iterate over all the shapes.
-
-  Currently only `=` operations are indexed.
+  Each type of operation that has been optimised such as `=` or `@>` will have it's own index module that implements the `Protocol` for this module.
   """
+  alias Electric.Shapes.Filter.Index.Protocol
+  alias Electric.Shapes.Filter.Indexes
 
-  alias Electric.Replication.Eval.Env
-  alias Electric.Shapes.Filter.Index
-  alias Electric.Shapes.WhereClause
-  alias Electric.Telemetry.OpenTelemetry
-  require Logger
-
-  defstruct [:type, :values]
-
-  def new(type), do: %Index{type: type, values: %{}}
-
-  def empty?(%Index{values: values}), do: values == %{}
-
-  def add_shape(%Index{} = index, value, {shape_id, shape}, and_where) do
-    shape_info = %{shape: shape, and_where: and_where}
-
-    %{
-      index
-      | values:
-          Map.update(
-            index.values,
-            value,
-            %{shape_id => shape_info},
-            &Map.put(&1, shape_id, shape_info)
-          )
-    }
-  end
+  def new("=", type), do: Indexes.EqualityIndex.new(type)
+  def new("@>", type), do: Indexes.InclusionIndex.new(type)
 
-  def remove_shape(%Index{} = index, shape_id) do
-    %{
-      index
-      | values:
-          index.values
-          |> Map.new(fn {value, shapes} -> {value, Map.delete(shapes, shape_id)} end)
-          |> Enum.reject(fn {_value, shapes} -> shapes == %{} end)
-          |> Map.new()
-    }
-  end
-
-  def affected_shapes(%Index{values: values, type: type}, field, record) do
-    case Map.get(values, value_from_record(record, field, type)) do
-      nil ->
-        MapSet.new()
-
-      shapes ->
-        OpenTelemetry.with_span(
-          "filter.index.filter_matched_shapes",
-          [field: field, matched_shapes_count: map_size(shapes)],
-          fn ->
-            for {shape_id, shape} <- shapes,
-                WhereClause.includes_record?(shape.and_where, record),
-                into: MapSet.new() do
-              shape_id
-            end
-          end
-        )
-    end
-  end
-
-  @env Env.new()
-  defp value_from_record(record, field, type) do
-    case Env.parse_const(@env, record[field], type) do
-      {:ok, value} ->
-        value
-
-      :error ->
-        raise RuntimeError,
-          message: "Could not parse value for field #{inspect(field)} of type #{inspect(type)}"
-    end
-  end
+  defdelegate empty?(index), to: Protocol
+  defdelegate add_shape(index, value, shape_instance, and_where), to: Protocol
+  defdelegate remove_shape(index, shape_id), to: Protocol
+  defdelegate affected_shapes(index, field, record), to: Protocol
+  defdelegate all_shapes(index), to: Protocol
+end
 
-  def all_shapes(%Index{values: values}) do
-    for {_value, shapes} <- values,
-        {shape_id, %{shape: shape}} <- shapes,
-        into: %{} do
-      {shape_id, shape}
-    end
-  end
+defprotocol Electric.Shapes.Filter.Index.Protocol do
+  def empty?(index)
+  def add_shape(index, value, shape_instance, and_where)
+  def remove_shape(index, shape_id)
+  def affected_shapes(index, field, record)
+  def all_shapes(index)
 end

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

@@ -0,0 +1,70 @@
+defmodule Electric.Shapes.Filter.Indexes.EqualityIndex do
+  @moduledoc """
+  Efficiently finds shapes that are affected by a change when the shape's where clause has `field = const` in it.
+
+  The index maps the values to the shapes that have that value as it's const in the `field = const` condition.
+
+  Rather than directly adding shapes, shapes are added to a `%WhereCondition{}` which represents can contain multiple
+  shapes and allows for further optimisations of other conditions in the shape's where clause.
+  """
+  alias Electric.Replication.Eval.Env
+  alias Electric.Shapes.Filter.Index
+  alias Electric.Shapes.Filter.Indexes.EqualityIndex
+  alias Electric.Shapes.Filter.WhereCondition
+  require Logger
+
+  defstruct [:type, :values]
+
+  def new(type), do: %EqualityIndex{type: type, values: %{}}
+
+  defimpl Index.Protocol, for: EqualityIndex do
+    def empty?(%EqualityIndex{values: values}), do: values == %{}
+
+    def add_shape(%EqualityIndex{} = index, value, {shape_id, shape}, and_where) do
+      index.values
+      |> Map.put_new(value, WhereCondition.new())
+      |> Map.update!(value, &WhereCondition.add_shape(&1, {shape_id, shape}, and_where))
+      |> then(&%{index | values: &1})
+    end
+
+    def remove_shape(%EqualityIndex{} = index, shape_id) do
+      index.values
+      |> Enum.map(fn {value, condition} ->
+        {value, WhereCondition.remove_shape(condition, shape_id)}
+      end)
+      |> Enum.reject(fn {_table, condition} -> WhereCondition.empty?(condition) end)
+      |> Map.new()
+      |> then(&%{index | values: &1})
+    end
+
+    def affected_shapes(%EqualityIndex{values: values, type: type}, field, record) do
+      case Map.get(values, value_from_record(record, field, type)) do
+        nil ->
+          MapSet.new()
+
+        condition ->
+          WhereCondition.affected_shapes(condition, record)
+      end
+    end
+
+    @env Env.new()
+    defp value_from_record(record, field, type) do
+      case Env.parse_const(@env, record[field], type) do
+        {:ok, value} ->
+          value
+
+        :error ->
+          raise RuntimeError,
+            message: "Could not parse value for field #{inspect(field)} of type #{inspect(type)}"
+      end
+    end
+
+    def all_shapes(%EqualityIndex{values: values}) do
+      for {_value, condition} <- values,
+          {shape_id, shape} <- WhereCondition.all_shapes(condition),
+          into: %{} do
+        {shape_id, shape}
+      end
+    end
+  end
+end