feat: Support arbitrary boolean expressions with subqueries (OR, NOT)

.changeset/arbitrary-boolean-expressions-with-subqueries.md

@@ -0,0 +1,19 @@
+---
+"@core/sync-service": minor
+"@core/elixir-client": minor
+---
+
+Add support for arbitrary boolean expressions with subqueries.
+
+Previously, WHERE clauses with OR or NOT combined with subqueries would cause shape invalidation. This update implements RFC "Arbitrary Boolean Expressions with Subqueries" which enables:
+
+- OR with multiple subqueries: `WHERE project_id IN (SELECT ...) OR assigned_to IN (SELECT ...)`
+- NOT with subqueries: `WHERE project_id NOT IN (SELECT ...)`
+- Complex expressions: `WHERE (a IN sq1 AND b='x') OR c NOT IN sq2`
+
+Key features:
+- DNF (Disjunctive Normal Form) decomposition for WHERE clause analysis
+- `active_conditions` array in row messages indicating which atomic conditions are satisfied
+- Position-based move-in/move-out handling that correctly inverts behavior for negated positions
+- Deduplication logic to prevent duplicate inserts when rows match multiple disjuncts
+- Updated elixir-client to parse `active_conditions` field in message headers

packages/elixir-client/lib/electric/client/message.ex

@@ -13,7 +13,8 @@ defmodule Electric.Client.Message do
       txids: [],
       op_position: 0,
       tags: [],
-      removed_tags: []
+      removed_tags: [],
+      active_conditions: []
     ]
 
     @type operation :: :insert | :update | :delete
@@ -29,7 +30,8 @@ defmodule Electric.Client.Message do
             txids: txids(),
             op_position: non_neg_integer(),
             tags: [tag()],
-            removed_tags: [tag()]
+            removed_tags: [tag()],
+            active_conditions: [boolean()]
           }
 
     @doc false
@@ -44,7 +46,8 @@ defmodule Electric.Client.Message do
         lsn: Map.get(msg, "lsn", nil),
         op_position: Map.get(msg, "op_position", 0),
         tags: Map.get(msg, "tags", []),
-        removed_tags: Map.get(msg, "removed_tags", [])
+        removed_tags: Map.get(msg, "removed_tags", []),
+        active_conditions: Map.get(msg, "active_conditions", [])
       }
     end
 
@@ -187,14 +190,15 @@ defmodule Electric.Client.Message do
 
     @enforce_keys [:shape_handle, :offset, :schema]
 
-    defstruct [:shape_handle, :offset, :schema, tag_to_keys: %{}, key_data: %{}]
+    defstruct [:shape_handle, :offset, :schema, tag_to_keys: %{}, key_data: %{}, disjunct_positions: nil]
 
     @type t :: %__MODULE__{
             shape_handle: Client.shape_handle(),
             offset: Offset.t(),
             schema: Client.schema(),
-            tag_to_keys: %{String.t() => MapSet.t(String.t())},
-            key_data: %{String.t() => %{tags: MapSet.t(String.t()), msg: ChangeMessage.t()}}
+            tag_to_keys: %{optional(term()) => MapSet.t(String.t())},
+            key_data: %{optional(String.t()) => map()},
+            disjunct_positions: [[non_neg_integer()]] | nil
           }
   end
 

packages/elixir-client/lib/electric/client/poll.ex

@@ -235,10 +235,11 @@ defmodule Electric.Client.Poll do
   end
 
   defp handle_message(%Message.ChangeMessage{} = msg, state) do
-    {tag_to_keys, key_data} =
-      TagTracker.update_tag_index(state.tag_to_keys, state.key_data, msg)
+    {tag_to_keys, key_data, disjunct_positions} =
+      TagTracker.update_tag_index(state.tag_to_keys, state.key_data, state.disjunct_positions, msg)
 
-    {:message, msg, %{state | tag_to_keys: tag_to_keys, key_data: key_data}}
+    {:message, msg,
+     %{state | tag_to_keys: tag_to_keys, key_data: key_data, disjunct_positions: disjunct_positions}}
   end
 
   defp handle_message(
@@ -249,6 +250,7 @@ defmodule Electric.Client.Poll do
       TagTracker.generate_synthetic_deletes(
         state.tag_to_keys,
         state.key_data,
+        state.disjunct_positions,
         patterns,
         request_timestamp
       )

packages/elixir-client/lib/electric/client/shape_state.ex

@@ -34,6 +34,7 @@ defmodule Electric.Client.ShapeState do
     up_to_date?: false,
     tag_to_keys: %{},
     key_data: %{},
+    disjunct_positions: nil,
     stale_cache_retry_count: 0
   ]
 
@@ -46,6 +47,7 @@ defmodule Electric.Client.ShapeState do
           up_to_date?: boolean(),
           tag_to_keys: %{optional(term()) => MapSet.t()},
           key_data: %{optional(term()) => %{tags: MapSet.t(), msg: term()}},
+          disjunct_positions: [[non_neg_integer()]] | nil,
           stale_cache_buster: String.t() | nil,
           stale_cache_retry_count: non_neg_integer()
         }
@@ -80,7 +82,8 @@ defmodule Electric.Client.ShapeState do
       schema: resume.schema,
       up_to_date?: true,
       tag_to_keys: Map.get(resume, :tag_to_keys, %{}),
-      key_data: Map.get(resume, :key_data, %{})
+      key_data: Map.get(resume, :key_data, %{}),
+      disjunct_positions: Map.get(resume, :disjunct_positions)
     }
   end
 
@@ -99,7 +102,8 @@ defmodule Electric.Client.ShapeState do
         up_to_date?: false,
         next_cursor: nil,
         tag_to_keys: %{},
-        key_data: %{}
+        key_data: %{},
+        disjunct_positions: nil
     }
   end
 
@@ -113,7 +117,8 @@ defmodule Electric.Client.ShapeState do
       offset: state.offset,
       schema: state.schema,
       tag_to_keys: state.tag_to_keys,
-      key_data: state.key_data
+      key_data: state.key_data,
+      disjunct_positions: state.disjunct_positions
     }
   end