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

[robacourt]

Summary

This PR implements RFC "Arbitrary Boolean Expressions with Subqueries" which extends Electric's subquery support from single IN (SELECT ...) conditions to arbitrary boolean expressions with OR, NOT, and multiple subqueries.

What's Changed

• 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 Implementation Details

1. DNF Decomposer (lib/electric/replication/eval/decomposer.ex)

   • Converts WHERE clause AST to Disjunctive Normal Form
   • Applies De Morgan's laws for NOT handling
   • Assigns positions to atomic conditions for tracking

2. Active Conditions (lib/electric/shapes/where_clause.ex)

   • Computes active_conditions array indicating which atomic conditions are satisfied
   • Evaluates DNF against active conditions to determine inclusion
   • Used for both replication stream changes and initial snapshots

3. Move-in/Move-out Handling (lib/electric/shapes/consumer/move_handling.ex)

   • Position-based move-in/move-out with correct NOT inversion
   • Move-in to negated position triggers move-out
   • Move-out from negated position triggers move-in query
   • Deduplication logic prevents duplicate inserts for rows matching multiple disjuncts

4. Message Format (lib/electric/log_items.ex, lib/electric/shapes/querying.ex)

   • active_conditions array included in row message headers
   • SQL generation for initial snapshots includes active_conditions
   • Updated elixir-client to parse active_conditions field

Test plan

• All 1389 existing tests pass
• 21 new integration tests for OR/NOT with subqueries (test/electric/plug/subquery_router_test.exs)
• DNF decomposer tests (test/electric/replication/eval/decomposer_test.exs)
• WhereClause tests for active conditions (test/electric/shapes/where_clause_test.exs)
• Updated router tests no longer expect 409 for OR/NOT shapes

Test Coverage

• Initial snapshot with OR subqueries
• Move-in/move-out with OR subqueries
• NOT IN shapes with correct inversion
• Row matching both disjuncts (deduplication)
• Row removed when all disjuncts become false
• Complex De Morgan expressions

Related

• RFC: docs/rfcs/arbitrary-boolean-expressions-with-subqueries.md
• Implementation Plan: packages/sync-service/IMPLEMENTATION_PLAN.md

🤖 Generated with Claude Code

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

• 🔍 Trigger a full review

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[codecov]

❌ 7 Tests Failed:

Tests completed	Failed	Passed	Skipped
2519	7	2512	11

View the top 3 failed test(s) by shortest run time

Elixir.Electric.Plug.RouterTest::test /v1/shapes - subqueries subquery combined with OR should return a 409 on move-out

Stack Traces | 0.102s run time

44) test /v1/shapes - subqueries subquery combined with OR should return a 409 on move-out (Electric.Plug.RouterTest)
     .../electric/plug/router_test.exs:2495
     match (=) failed
     code:  assert {req, 200, response} = shape_req(orig_req, opts)
     left:  {req, 200, response}
     right: {%{offset: "-1", table: "child", where: "parent_id in (SELECT id FROM parent WHERE include_parent = true) OR include_child = true", live: false}, 400, %{"errors" => %{"where" => ["WHERE clauses with OR or NOT combined with subqueries require protocol version 2. Please upgrade your client or set the Electric-Protocol-Version: 2 header."]}, "message" => "Invalid request"}}
     stacktrace:
       .../electric/plug/router_test.exs:2505: (test)

Elixir.Electric.Plug.RouterTest::test /v1/shapes - subqueries subquery combined with OR should return a 409 on move-in

Stack Traces | 0.114s run time

16) test /v1/shapes - subqueries subquery combined with OR should return a 409 on move-in (Electric.Plug.RouterTest)
     .../electric/plug/router_test.exs:2527
     match (=) failed
     code:  assert {req, 200, response} = shape_req(orig_req, opts)
     left:  {req, 200, response}
     right: {%{offset: "-1", table: "child", where: "parent_id in (SELECT id FROM parent WHERE include_parent = true) OR include_child = true", live: false}, 400, %{"errors" => %{"where" => ["WHERE clauses with OR or NOT combined with subqueries require protocol version 2. Please upgrade your client or set the Electric-Protocol-Version: 2 header."]}, "message" => "Invalid request"}}
     stacktrace:
       .../electric/plug/router_test.exs:2537: (test)

Elixir.Electric.Plug.RouterTest::test /v1/shapes - subqueries NOT IN subquery should return 409 on move-in to subquery

Stack Traces | 0.121s run time

12) test /v1/shapes - subqueries NOT IN subquery should return 409 on move-in to subquery (Electric.Plug.RouterTest)
     .../electric/plug/router_test.exs:2334
     match (=) failed
     code:  assert {req, 200, [data, snapshot_end]} = shape_req(req, opts)
     left:  {req, 200, [data, snapshot_end]}
     right: {%{offset: "-1", table: "child", where: "parent_id NOT IN (SELECT id FROM parent WHERE excluded = true)", live: false}, 400, %{"errors" => %{"where" => ["WHERE clauses with OR or NOT combined with subqueries require protocol version 2. Please upgrade your client or set the Electric-Protocol-Version: 2 header."]}, "message" => "Invalid request"}}
     stacktrace:
       .../electric/plug/router_test.exs:2346: (test)

Elixir.Electric.Plug.RouterTest::test /v1/shapes - subqueries subqueries work with quoted column names and are tagged correctly

Stack Traces | 0.151s run time

76) test /v1/shapes - subqueries subqueries work with quoted column names and are tagged correctly (Electric.Plug.RouterTest)
     .../electric/plug/router_test.exs:2793
     match (=) failed
     The following variables were pinned:
       tag = "edef6e0e7454bf0ad3adc4bf774e1fd0"
     code:  assert %{"value" => %{"id" => "1", "parentId" => "1", "Value" => "10"}, "headers" => %{"tags" => [^tag]}} =
              Enum.find(response, &Map.has_key?(&1, "key"))
     left:  %{"value" => %{"id" => "1", "parentId" => "1", "Value" => "10"}, "headers" => %{"tags" => [^tag]}}
     right: %{"headers" => %{"active_conditions" => [true, true], "operation" => "insert", "relation" => ["public", "Child"], "tags" => ["edef6e0e7454bf0ad3adc4bf774e1fd0/6de8df718bc42a1f77e75ddfed67be68"]}, "key" => "\"public\".\"Child\"/\"1\"", "value" => %{"Value" => "10", "id" => "1", "parentId" => "1"}}
     stacktrace:
       .../electric/plug/router_test.exs:2807: (test)

Elixir.Electric.Plug.RouterTest::test /v1/shapes - subqueries subqueries work with params

Stack Traces | 0.153s run time

33) test /v1/shapes - subqueries subqueries work with params (Electric.Plug.RouterTest)
     .../electric/plug/router_test.exs:2745
     match (=) failed
     The following variables were pinned:
       tag = "fdbe0699a0e90077a1104f673dc5fff7"
     code:  assert {_, 200,
             [
               %{"headers" => %{"tags" => [^tag]}, "value" => %{"id" => "3"}},
               %{"headers" => %{"control" => "snapshot-end"}},
               up_to_date_ctl()
             ]} = Task.await(task)
     left:  {_, 200,
             [
               %{"headers" => %{"tags" => [^tag]}, "value" => %{"id" => "3"}},
               %{"headers" => %{"control" => "snapshot-end"}},
               up_to_date_ctl()
             ]}
     right: {%{handle: "129707345-1771410743115937", offset: "814342736_2", table: "child", where: "value in (SELECT value FROM parent WHERE other_value >= $2) AND other_value >= $1", params: %{"1" => "10", "2" => "6"}, live: true}, 200, [%{"headers" => %{"active_conditions" => [true, true], "is_move_in" => true, "operation" => "insert", "relation" => ["public", "child"], "tags" => ["fdbe0699a0e90077a1104f673dc5fff7/fdbe0699a0e90077a1104f673dc5fff7"]}, "key" => "\"public\".\"child\"/\"3\"", "value" => %{"id" => "3", "other_value" => "20", "value" => "20"}}, %{"headers" => %{"control" => "snapshot-end", "xip_list" => [], "xmax" => "2948", "xmin" => "2948"}}, %{"headers" => %{"control" => "up-to-date", "global_last_seen_lsn" => "814342736"}}]}
     stacktrace:
       .../electric/plug/router_test.exs:2778: (test)

Elixir.Electric.Plug.RouterTest::test /v1/shapes - subqueries supports two subqueries at the same level but returns 409 on move-in

Stack Traces | 0.198s run time

15) test /v1/shapes - subqueries supports two subqueries at the same level but returns 409 on move-in (Electric.Plug.RouterTest)
     .../electric/plug/router_test.exs:2985
     match (=) failed
     code:  assert %{status: 409} = Task.await(task)
     left:  %{status: 409}
     right: %Plug.Conn{adapter: {Plug.Adapters.Test.Conn, :...}, assigns: %{request: %Electric.Shapes.Api.Request{chunk_end_offset: LogOffset.new(890861112, 0), handle: "16467413-1771410753254019", last_offset: LogOffset.new(890861112, 0), global_last_seen_lsn: 890861288, new_changes_ref: #Reference<0.504478028.72351746.167460>, new_changes_pid: #PID<0.32246.0>, api: %Electric.Shapes.Api{inspector: {Electric.Postgres.Inspector.EtsInspector, [stack_id: "Electric.Plug.RouterTest test /v1/shapes - subqueries supports two subqueries at the same level but returns 409 on move-in", server: {:via, Registry, {:"Electric.ProcessRegistry:Electric.Plug.RouterTest test /v1/shapes - subqueries supports two subqueries at the same level but returns 409 on move-in", {Electric.Postgres.Inspector.EtsInspector, nil}}}]}, shape: nil, stack_id: "Electric.Plug.RouterTest test /v1/shapes - subqueries supports two subqueries at the same level but returns 409 on move-in", feature_flags: ["allow_subqueries", "tagged_subqueries"], max_concurrent_requests: %{existing: 1000, initial: 300}, allow_shape_deletion: true, keepalive_interval: 21000, long_poll_timeout: 4000, sse_timeout: 60000, max_age: 60, stack_ready_timeout: 5000, stale_age: 300, send_cache_headers?: true, encoder: Electric.Shapes.Api.Encoder.JSON, sse_encoder: Electric.Shapes.Api.Encoder.SSE, configured: true}, params: %Electric.Shapes.Api.Params{table: "projects", offset: LogOffset.new(890861112, 0), handle: "16467413-1771410753254019", live: true, where: "workspace_id IN (SELECT workspace_id FROM workspace_members WHERE user_id = 100) AND id IN (SELECT project_id FROM project_members WHERE user_id = 100)", columns: nil, shape_definition: Shape.new!({22921, "public.projects"}, where: "workspace_id IN (SELECT workspace_id FROM public.workspace_members WHERE user_id = 100) AND id IN (SELECT project_id FROM public.project_members WHERE user_id = 100)", deps: [Shape.new!({22928, "public.workspace_members"}, where: "user_id = 100", columns: ["workspace_id"]), Shape.new!({22933, "public.project_members"}, where: "user_id = 100", columns: ["project_id"])]), replica: :default, params: %{}, experimental_compaction: false, live_sse: false, log: :full, electric_protocol_version: 1, subset: nil}, response: %Electric.Shapes.Api.Response{handle: "16467413-1771410753254019", offset: LogOffset.new(890861112, 0), shape_definition: Shape.new!({22921, "public.projects"}, where: "workspace_id IN (SELECT workspace_id FROM public.workspace_members WHERE user_id = 100) AND id IN (SELECT project_id FROM public.project_members WHERE user_id = 100)", deps: [Shape.new!({22928, "public.workspace_members"}, where: "user_id = 100", columns: ["workspace_id"]), Shape.new!({22933, "public.project_members"}, where: "user_id = 100", columns: ["project_id"])]), known_error: nil, retry_after: nil, api: %Electric.Shapes.Api{inspector: {Electric.Postgres.Inspector.EtsInspector, [stack_id: "Electric.Plug.RouterTest test /v1/shapes - subqueries supports two subqueries at the same level but returns 409 on move-in", server: {:via, Registry, {:"Electric.ProcessRegistry:Electric.Plug.RouterTest test /v1/shapes - subqueries supports two subqueries at the same level but returns 409 on move-in", {Electric.Postgres.Inspector.EtsInspector, nil}}}]}, shape: nil, stack_id: "Electric.Plug.RouterTest test /v1/shapes - subqueries supports two subqueries at the same level but returns 409 on move-in", feature_flags: ["allow_subqueries", "tagged_subqueries"], max_concurrent_requests: %{existing: 1000, initial: 300}, allow_shape_deletion: true, keepalive_interval: 21000, long_poll_timeout: 4000, sse_timeout: 60000, max_age: 60, stack_ready_timeout: 5000, stale_age: 300, send_cache_headers?: true, encoder: Electric.Shapes.Api.Encoder.JSON, sse_encoder: Electric.Shapes.Api.Encoder.SSE, configured: true}, chunked: false, up_to_date: true, no_changes: false, response_type: :normal_log, ...}}, ...}, ...}
     stacktrace:
       .../electric/plug/router_test.exs:3054: (test)

Elixir.Electric.Replication.PublicationManagerTest::test component restarts handles relation tracker restart

Stack Traces | 0.221s run time

1) test component restarts handles relation tracker restart (Electric.Replication.PublicationManagerTest)
     .../electric/replication/publication_manager_test.exs:503
     ** (exit) exited in: GenServer.call({:via, Registry, {:"Electric.ProcessRegistry:Electric.Replication.PublicationManagerTest test component restarts handles relation tracker restart", {Electric.Replication.PublicationManager.RelationTracker, nil}}}, {:remove_shape, "44497180-1771410662370176"}, 5000)
         ** (EXIT) no process: the process is not alive or there's no process currently associated with the given name, possibly because its application isn't started
     code: PublicationManager.remove_shape(ctx.stack_id, shape_handle)
     stacktrace:
       (elixir 1.19.1) lib/gen_server.ex:1135: GenServer.call/3
       (electric 1.4.4) .../replication/publication_manager/relation_tracker.ex:73: Electric.Replication.PublicationManager.RelationTracker.remove_shape/2
       .../electric/replication/publication_manager_test.exs:522: (test)

Elixir.Electric.ClientTest::test move-out handling multiple patterns matching same row generates single delete

Stack Traces | 0.528s run time

25) test move-out handling multiple patterns matching same row generates single delete (Electric.ClientTest)
     test/electric/client_test.exs:2009
     ** (Electric.Client.Error) 
     code: msgs = stream(ctx, 4)
     stacktrace:
       (electric_client 0.9.0) .../electric/client/stream.ex:218: Electric.Client.Stream.handle_error/2
       (electric_client 0.9.0) .../electric/client/stream.ex:265: Enumerable.Electric.Client.Stream.reduce/3
       (elixir 1.19.1) lib/enum.ex:3603: Enum.take/2
       test/electric/client_test.exs:2064: (test)

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[netlify]

✅ Deploy Preview for electric-next ready!

Name	Link
🔨 Latest commit	76452d9
🔍 Latest deploy log	https://app.netlify.com/projects/electric-next/deploys/69a58579b70f100008934552
😎 Deploy Preview	https://deploy-preview-3791--electric-next.netlify.app
📱 Preview on mobile	Toggle QR Code... Use your smartphone camera to open QR code link.

---

To edit notification comments on pull requests, go to your Netlify project configuration.

[robacourt]

@coderabbitai review

[blacksmith-sh]

Found 26 test failures on Blacksmith runners:

Failures

Test	View Logs
Elixir.Electric.ClientTest/ test move-out handling multiple patterns matching same row generates single delete	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries NOT IN subquery should return 409 on move-in to subquery	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries NOT IN subquery should return 409 on move-in to subquery	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries NOT IN subquery should return 409 on move-in to subquery	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries NOT IN subquery should return 409 on move-in to subquery	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subqueries work with params	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subqueries work with params	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subqueries work with params	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subqueries work with params	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subqueries work with quoted column names and are tagged correctly	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subqueries work with quoted column names and are tagged correctly	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subqueries work with quoted column names and are tagged correctly	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subqueries work with quoted column names and are tagged correctly	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subquery combined with OR should return a 409 on move-in	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subquery combined with OR should return a 409 on move-in	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subquery combined with OR should return a 409 on move-in	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subquery combined with OR should return a 409 on move-in	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subquery combined with OR should return a 409 on move-out	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subquery combined with OR should return a 409 on move-out	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subquery combined with OR should return a 409 on move-out	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries subquery combined with OR should return a 409 on move-out	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries supports two subqueries at the same level but returns 409 on move-i n	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries supports two subqueries at the same level but returns 409 on move-i n	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries supports two subqueries at the same level but returns 409 on move-i n	View Logs
Elixir.Electric.Plug.RouterTest/test /v1/ shapes - subqueries supports two subqueries at the same level but returns 409 on move-i n	View Logs
Elixir.Electric.Replication.PublicationManagerTest/ test component restarts handles relation tracker restart	View Logs

[kevin-dp]

Hi @robacourt, while implementing the DNF/active_conditions support in the ts/db client, I noticed that tag_tracker.ex stores disjunct_positions per-key in key_data and re-derives it every time a message with new tags arrives:

disjunct_positions =
  case derive_disjunct_positions(normalized_new) do
    [] -> current_data && current_data.disjunct_positions
    positions -> positions
  end

This is redundant work — disjunct_positions is determined solely by the shape's WHERE clause (the DNF decomposition structure), which is fixed for the lifetime of a shape. Every row has the same tag structure; the hash values differ per row but which positions are nil vs participating is always identical. So there's no need to store it per-key or recompute it on each tagged message.

In the TS client we derive it once from the first tagged message and store it as a single variable shared across all rows. Might be worth doing the same in the Elixir client — just a single disjunct_positions field alongside tag_to_keys/key_data instead of inside each key's data, derived once on the first insert.

[kevin-dp]

Also, i think it would be good to add the unit tests we have in ts/db for this feature (cf. https://github.com/TanStack/db/pull/1270/changes#diff-286083ece2106d963fcda491df8579332e28dd8219a45a51c967afff26d70747) to the Elixir client too.