feat(sync-service): remove subquery feature flags

.changeset/subqueries-generally-available.md

@@ -0,0 +1,7 @@
+---
+"@core/sync-service": patch
+"@electric-sql/client": patch
+"@electric-ax/agents-server": patch
+---
+
+Subqueries in shape WHERE clauses are now generally available and always enabled, including incremental move handling for compound `AND`/`OR`/`NOT` expressions. The `allow_subqueries` and `tagged_subqueries` feature flags have been removed — they no longer need to be set via `ELECTRIC_FEATURE_FLAGS`.

packages/agents-server/docker-compose.dev.yml

@@ -30,7 +30,6 @@ services:
     environment:
       DATABASE_URL: postgresql://electric_agents:electric_agents@postgres:5432/electric_agents
       ELECTRIC_INSECURE: 'true'
-      ELECTRIC_FEATURE_FLAGS: allow_subqueries
     depends_on:
       postgres:
         condition: service_healthy

packages/sync-service/lib/electric/config.ex

@@ -38,7 +38,7 @@ defmodule Electric.Config do
 
   @build_env Mix.env()
 
-  @known_feature_flags ~w[allow_subqueries tagged_subqueries]
+  @known_feature_flags ~w[]
   @default_storage_dir "./persistent"
 
   @defaults [
@@ -133,7 +133,7 @@ defmodule Electric.Config do
     consumer_gc_heap_threshold: nil,
     ## Misc
     process_registry_partitions: &Electric.Config.Defaults.process_registry_partitions/0,
-    feature_flags: if(Mix.env() == :test, do: @known_feature_flags, else: []),
+    feature_flags: [],
     publication_refresh_period: 60_000,
     schema_reconciler_period: 60_000,
     snapshot_timeout_to_first_data: :timer.seconds(30),

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

@@ -1026,10 +1026,6 @@ defmodule Electric.Shapes.Consumer do
     handle_txn_with_truncate(xid, state)
   end
 
-  defp handle_event_error(state, :unsupported_subquery) do
-    mark_for_removal(state)
-  end
-
   defp handle_event_error(state, :buffer_overflow) do
     Logger.warning("Subquery buffer overflow for #{state.shape_handle} - terminating shape")
 

packages/sync-service/lib/electric/shapes/consumer/event_handler/subqueries/steady.ex

@@ -36,15 +36,6 @@ defmodule Electric.Shapes.Consumer.EventHandler.Subqueries.Steady do
     {:ok, state, []}
   end
 
-  def handle_event(
-        %__MODULE__{
-          shape_info: %ShapeInfo{dependency_move_policy: :invalidate_on_dependency_move}
-        },
-        {:materializer_changes, _dep_handle, _payload}
-      ) do
-    {:error, :unsupported_subquery}
-  end
-
   def handle_event(%__MODULE__{} = state, {:materializer_changes, dep_handle, payload}) do
     subquery_ref = RefResolver.ref_from_dep_handle!(state.shape_info.ref_resolver, dep_handle)
     dep_index = subquery_ref |> List.last() |> String.to_integer()

packages/sync-service/lib/electric/shapes/consumer/event_handler_builder.ex

@@ -13,7 +13,6 @@ defmodule Electric.Shapes.Consumer.EventHandlerBuilder do
   def build(%State{shape: %Shape{shape_dependencies_handles: dep_handles}} = state, action)
       when dep_handles != [] do
     {:ok, dnf_plan} = DnfPlan.compile(state.shape)
-    dependency_move_policy = dependency_move_policy(state.stack_id, state.shape)
 
     {views, dep_handle_to_ref, dep_index_to_ref} =
       dep_handles
@@ -44,8 +43,7 @@ defmodule Electric.Shapes.Consumer.EventHandlerBuilder do
         dnf_plan: dnf_plan,
         ref_resolver:
           Electric.Shapes.Consumer.Subqueries.RefResolver.new(dep_handle_to_ref, dep_index_to_ref),
-        buffer_max_transactions: buffer_max_transactions,
-        dependency_move_policy: dependency_move_policy
+        buffer_max_transactions: buffer_max_transactions
       },
       views: views
     }
@@ -63,14 +61,4 @@ defmodule Electric.Shapes.Consumer.EventHandlerBuilder do
 
     {:ok, handler, [%SetupEffects.SubscribeShape{action: action}]}
   end
-
-  defp dependency_move_policy(stack_id, _shape) do
-    feature_flags = Electric.StackConfig.lookup(stack_id, :feature_flags, [])
-
-    if "tagged_subqueries" not in feature_flags do
-      :invalidate_on_dependency_move
-    else
-      :stream_dependency_moves
-    end
-  end
 end

packages/sync-service/lib/electric/shapes/consumer/state.ex

@@ -119,7 +119,6 @@ defmodule Electric.Shapes.Consumer.State do
 
   @spec initialize_shape(uninitialized_t(), Shape.t(), map()) :: uninitialized_t()
   def initialize_shape(%__MODULE__{} = state, shape, opts) do
-    feature_flags = Map.get(opts, :feature_flags, [])
     is_subquery_shape? = Map.get(opts, :is_subquery_shape?, false)
 
     %{
@@ -128,8 +127,7 @@ defmodule Electric.Shapes.Consumer.State do
         # Enable direct fragment-to-storage streaming for shapes without subquery dependencies
         # and if the current shape itself isn't an inner shape of a shape with subqueries.
         write_unit:
-          if "allow_subqueries" in feature_flags or shape.shape_dependencies != [] or
-               is_subquery_shape? do
+          if shape.shape_dependencies != [] or is_subquery_shape? do
             @write_unit_txn
           else
             @write_unit_txn_fragment

packages/sync-service/lib/electric/shapes/consumer/subqueries/shape_info.ex

@@ -5,25 +5,21 @@ defmodule Electric.Shapes.Consumer.Subqueries.ShapeInfo do
   alias Electric.Shapes.DnfPlan
   alias Electric.Shapes.Shape
 
-  @type dependency_move_policy :: :stream_dependency_moves | :invalidate_on_dependency_move
-
   @enforce_keys [
     :shape,
     :stack_id,
     :shape_handle,
     :dnf_plan,
     :ref_resolver,
-    :buffer_max_transactions,
-    :dependency_move_policy
+    :buffer_max_transactions
   ]
   defstruct [
     :shape,
     :stack_id,
     :shape_handle,
     :dnf_plan,
     :ref_resolver,
-    :buffer_max_transactions,
-    :dependency_move_policy
+    :buffer_max_transactions
   ]
 
   @type t() :: %__MODULE__{
@@ -32,7 +28,6 @@ defmodule Electric.Shapes.Consumer.Subqueries.ShapeInfo do
           shape_handle: String.t(),
           dnf_plan: DnfPlan.t(),
           ref_resolver: RefResolver.t(),
-          buffer_max_transactions: pos_integer(),
-          dependency_move_policy: dependency_move_policy()
+          buffer_max_transactions: pos_integer()
         }
 end

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

@@ -295,7 +295,6 @@ defmodule Electric.Shapes.Shape do
   defp validate_where_clause(where, %{inspector: inspector} = opts, refs) do
     with {:ok, where} <- Parser.parse_query(where),
          {:ok, subqueries} <- Parser.extract_subqueries(where),
-         :ok <- check_feature_flag(subqueries, opts),
          {:ok, shape_dependencies, sublink_dependency_indexes} <-
            build_shape_dependencies(subqueries, opts),
          {:ok, dependency_refs} <- build_dependency_refs(shape_dependencies, inspector),
@@ -318,15 +317,6 @@ defmodule Electric.Shapes.Shape do
     end
   end
 
-  defp check_feature_flag(subqueries, opts) do
-    if subqueries != [] and
-         not Enum.member?(opts.feature_flags, "allow_subqueries") do
-      {:error, {:where, "Subqueries are not supported"}}
-    else
-      :ok
-    end
-  end
-
   defp make_opts_from_select(select, opts) do
     with {:ok, {columns, from, where}} <- Parser.extract_parts_from_select(select) do
       {:ok,

packages/sync-service/test/electric/plug/serve_shape_plug_test.exs

@@ -267,7 +267,6 @@ defmodule Electric.Plug.ServeShapePlugTest do
       ctx =
         ctx
         |> Map.put(:inspector, @subquery_inspector)
-        |> Map.put(:feature_flags, ["allow_subqueries"])
 
       Repatch.patch(Electric.Shapes, :fetch_handle_by_shape, fn _, _ ->
         flunk("should reject before checking whether the shape already exists")

packages/sync-service/test/electric/shapes/consumer/event_handler/subqueries_test.exs

@@ -42,38 +42,6 @@ defmodule Electric.Shapes.Consumer.EventHandler.SubqueriesTest do
              ] = plan
     end
 
-    test "still converts root transactions when dependency moves are configured to invalidate" do
-      handler =
-        new_handler(
-          subquery_view: MapSet.new([1]),
-          dependency_move_policy: :invalidate_on_dependency_move
-        )
-
-      assert {:ok, %Steady{}, plan} =
-               EventHandler.handle_event(
-                 handler,
-                 txn(50, [child_insert("1", "1"), child_insert("2", "2")])
-               )
-
-      assert [
-               %Effects.AppendChanges{
-                 changes: [%Changes.NewRecord{record: %{"id" => "1"}, last?: true}]
-               },
-               %Effects.NotifyFlushed{log_offset: _}
-             ] = plan
-    end
-
-    test "returns unsupported_subquery when dependency moves are configured to invalidate" do
-      handler = new_handler(dependency_move_policy: :invalidate_on_dependency_move)
-      dep_handle = dep_handle(handler)
-
-      assert {:error, :unsupported_subquery} =
-               EventHandler.handle_event(
-                 handler,
-                 {:materializer_changes, dep_handle, %{move_in: [{1, "1"}], move_out: []}}
-               )
-    end
-
     test "negated subquery turns dependency move-in into an outer move-out" do
       handler = new_handler(shape: negated_shape())
       dep_handle = dep_handle(handler)
@@ -876,9 +844,7 @@ defmodule Electric.Shapes.Consumer.EventHandler.SubqueriesTest do
         dnf_plan: dnf_plan,
         ref_resolver:
           RefResolver.new(%{dep_handle => {0, ["$sublink", "0"]}}, %{0 => ["$sublink", "0"]}),
-        buffer_max_transactions: Keyword.get(opts, :buffer_max_transactions, 1000),
-        dependency_move_policy:
-          Keyword.get(opts, :dependency_move_policy, :stream_dependency_moves)
+        buffer_max_transactions: Keyword.get(opts, :buffer_max_transactions, 1000)
       },
       views: %{["$sublink", "0"] => Keyword.get(opts, :subquery_view, MapSet.new())}
     }
@@ -895,17 +861,15 @@ defmodule Electric.Shapes.Consumer.EventHandler.SubqueriesTest do
   defp shape do
     Shape.new!("child",
       where: "parent_id IN (SELECT id FROM public.parent WHERE value = 'keep')",
-      inspector: @inspector,
-      feature_flags: ["allow_subqueries"]
+      inspector: @inspector
     )
     |> fill_handles()
   end
 
   defp negated_shape do
     Shape.new!("child",
       where: "parent_id NOT IN (SELECT id FROM public.parent WHERE value = 'keep')",
-      inspector: @inspector,
-      feature_flags: ["allow_subqueries"]
+      inspector: @inspector
     )
     |> fill_handles()
   end

packages/sync-service/test/electric/shapes/consumer_test.exs

@@ -647,10 +647,6 @@ defmodule Electric.Shapes.ConsumerTest do
         Map.get(ctx, :shape_suspend_after, 60_000)
       )
 
-      if not Map.get(ctx, :allow_subqueries, true) do
-        Electric.StackConfig.put(ctx.stack_id, :feature_flags, [])
-      end
-
       :ok
     end
 
@@ -781,7 +777,6 @@ defmodule Electric.Shapes.ConsumerTest do
                get_log_items_from_storage(LogOffset.last_before_real_offsets(), shape_storage)
     end
 
-    @tag allow_subqueries: false
     test "duplicate txn fragment handling is idempotent", ctx do
       {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape1, ctx.stack_id)
       :started = ShapeCache.await_snapshot_start(shape_handle, ctx.stack_id)
@@ -890,7 +885,6 @@ defmodule Electric.Shapes.ConsumerTest do
       refute_receive {^ref, :new_changes, _}
     end
 
-    @tag allow_subqueries: false
     test "skips an already-applied multi-fragment transaction replayed past a fresh log collector",
          ctx do
       # Multi-fragment variant of "skips an already-applied transaction replayed
@@ -1106,8 +1100,7 @@ defmodule Electric.Shapes.ConsumerTest do
       assert_receive {:flush_boundary_updated, ^tx_offset}
     end
 
-    @tag allow_subqueries: false,
-         delay_snapshot_creation?: true,
+    @tag delay_snapshot_creation?: true,
          with_pure_file_storage_opts: [flush_period: 1]
     test "transaction fragments are buffered until snapshot xmin is known", ctx do
       register_as_replication_client(ctx.stack_id)
@@ -1319,8 +1312,7 @@ defmodule Electric.Shapes.ConsumerTest do
       assert {:ok, last_log_offset} == Storage.fetch_latest_offset(shape_storage)
     end
 
-    @tag allow_subqueries: false,
-         pg_snapshot: {10, 13, [10]},
+    @tag pg_snapshot: {10, 13, [10]},
          with_pure_file_storage_opts: [flush_period: 1]
     test "fragments that belong to transactions already included in the snapshot are skipped",
          ctx do
@@ -1966,7 +1958,7 @@ defmodule Electric.Shapes.ConsumerTest do
       assert [] == :ets.tab2list(table)
     end
 
-    @tag allow_subqueries: false, with_pure_file_storage_opts: [flush_period: 1]
+    @tag with_pure_file_storage_opts: [flush_period: 1]
     test "writes txn fragments to storage immediately but keeps txn boundaries when flushing",
          ctx do
       {shape_handle, _} = ShapeCache.get_or_create_shape_handle(@shape1, ctx.stack_id)
@@ -2095,7 +2087,7 @@ defmodule Electric.Shapes.ConsumerTest do
       assert_receive {:flush_boundary_updated, ^offset}
     end
 
-    @tag allow_subqueries: false, with_pure_file_storage_opts: [flush_period: 1]
+    @tag with_pure_file_storage_opts: [flush_period: 1]
     test "flush notification for multi-fragment txn is not lost when storage flushes before commit fragment",
          %{stack_id: stack_id} = ctx do
       # Regression test for https://github.com/electric-sql/electric/issues/3985
@@ -2195,7 +2187,7 @@ defmodule Electric.Shapes.ConsumerTest do
       assert_receive {:flush_boundary_updated, ^tx_offset}, @receive_timeout
     end
 
-    @tag allow_subqueries: false, with_pure_file_storage_opts: [flush_period: 10_000]
+    @tag with_pure_file_storage_opts: [flush_period: 10_000]
     test "flush notification offset is aligned when storage flushes before commit arrives at consumer",
          %{stack_id: stack_id} do
       # Regression test for https://github.com/electric-sql/electric/issues/4063
@@ -2299,7 +2291,7 @@ defmodule Electric.Shapes.ConsumerTest do
       assert_receive {:flush_boundary_updated, ^tx_offset}, @receive_timeout
     end
 
-    @tag allow_subqueries: false, with_pure_file_storage_opts: [flush_period: 1]
+    @tag with_pure_file_storage_opts: [flush_period: 1]
     test "dead consumer doesn't block flush notifications from advancing as live consumers flush to storage",
          ctx do
       {shape_handle1, _} = ShapeCache.get_or_create_shape_handle(@shape1, ctx.stack_id)