Reject shapes that fail DnfPlan on creation

Author: robacourt

packages/sync-service/lib/electric/shapes/api/params.ex

@@ -3,6 +3,7 @@ defmodule Electric.Shapes.Api.Params do
 
   alias Electric.Replication.LogOffset
   alias Electric.Shapes.Api
+  alias Electric.Shapes.DnfPlan
   alias Electric.Shapes.Shape
 
   import Ecto.Changeset
@@ -331,7 +332,13 @@ defmodule Electric.Shapes.Api.Params do
            log_mode: fetch_field!(changeset, :log)
          ) do
       {:ok, shape} ->
-        put_change(changeset, :shape_definition, shape)
+        case DnfPlan.compile(shape) do
+          {:error, reason} ->
+            add_error(changeset, :where, reason)
+
+          _ok ->
+            put_change(changeset, :shape_definition, shape)
+        end
 
       {:error, :connection_not_available} ->
         add_error(

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

@@ -537,16 +537,22 @@ defmodule Electric.Shapes.DnfPlan do
     with {:ok, decomposition} <- Decomposer.decompose(shape.where.eval) do
       positions = enrich_positions(decomposition.subexpressions, shape)
 
-      {:ok,
-       %__MODULE__{
-         disjuncts: decomposition.disjuncts,
-         disjuncts_positions: decomposition.disjuncts_positions,
-         position_count: decomposition.position_count,
-         positions: positions,
-         dependency_positions: build_dependency_positions(positions),
-         dependency_disjuncts: build_dependency_disjuncts(decomposition.disjuncts, positions),
-         dependency_polarities: build_dependency_polarities(positions)
-       }}
+      case build_dependency_polarities(positions) do
+        {:ok, dependency_polarities} ->
+          {:ok,
+           %__MODULE__{
+             disjuncts: decomposition.disjuncts,
+             disjuncts_positions: decomposition.disjuncts_positions,
+             position_count: decomposition.position_count,
+             positions: positions,
+             dependency_positions: build_dependency_positions(positions),
+             dependency_disjuncts: build_dependency_disjuncts(decomposition.disjuncts, positions),
+             dependency_polarities: dependency_polarities
+           }}
+
+        {:error, _} = error ->
+          error
+      end
     end
   end
 
@@ -621,17 +627,18 @@ defmodule Electric.Shapes.DnfPlan do
       fn {_pos, info} -> info.dependency_index end,
       fn {_pos, info} -> info.negated end
     )
-    |> Map.new(fn {dep_index, negated_flags} ->
+    |> Enum.reduce_while({:ok, %{}}, fn {dep_index, negated_flags}, {:ok, acc} ->
       case Enum.uniq(negated_flags) do
         [false] ->
-          {dep_index, :positive}
+          {:cont, {:ok, Map.put(acc, dep_index, :positive)}}
 
         [true] ->
-          {dep_index, :negated}
+          {:cont, {:ok, Map.put(acc, dep_index, :negated)}}
 
-        mixed ->
-          raise ArgumentError,
-                "dependency #{dep_index} has inconsistent polarity across positions: #{inspect(mixed)}"
+        _mixed ->
+          {:halt,
+           {:error,
+            "a subquery dependency cannot be used with both positive and negative polarity in the same filter"}}
       end
     end)
   end

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

@@ -3,7 +3,6 @@ defmodule Electric.Shapes.Shape do
   Struct describing the requested shape
   """
   alias Electric.Shapes.Consumer.Subqueries
-  alias Electric.Replication.Eval.Decomposer
   alias Electric.Replication.Eval.Expr
   alias Electric.Postgres.Inspector
   alias Electric.Replication.Eval.Parser
@@ -234,8 +233,7 @@ defmodule Electric.Shapes.Shape do
            validate_selected_columns(column_info, pk_cols, supported_features, opts),
          refs = Inspector.columns_to_expr(column_info),
          {:ok, where, shape_dependencies} <-
-           validate_where_clause(Map.get(opts, :where), opts, refs),
-         :ok <- validate_dnf_decomposition(where, shape_dependencies) do
+           validate_where_clause(Map.get(opts, :where), opts, refs) do
       flags =
         [
           if(is_nil(Map.get(opts, :columns)), do: :selects_all_columns),
@@ -306,15 +304,6 @@ defmodule Electric.Shapes.Shape do
     end
   end
 
-  defp validate_dnf_decomposition(_where, []), do: :ok
-  defp validate_dnf_decomposition(nil, _deps), do: :ok
-
-  defp validate_dnf_decomposition(where, [_ | _]) do
-    case Decomposer.decompose(where.eval) do
-      {:ok, _} -> :ok
-      {:error, reason} -> {:error, {:where, reason}}
-    end
-  end
 
   defp check_feature_flag(subqueries, opts) do
     if subqueries != [] and

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

@@ -2249,6 +2249,23 @@ defmodule Electric.Plug.RouterTest do
                |> Router.call(opts)
     end
 
+    @tag with_sql: [
+           "CREATE TABLE parent (id INT PRIMARY KEY, value INT NOT NULL)",
+           "CREATE TABLE child (id INT PRIMARY KEY, parent_id INT NOT NULL REFERENCES parent(id), value INT NOT NULL)"
+         ]
+    test "return 400 if same subquery is used with both positive and negative polarity", %{
+      opts: opts
+    } do
+      assert %{status: 400} =
+               conn("GET", "/v1/shape", %{
+                 table: "child",
+                 offset: "-1",
+                 where:
+                   "parent_id IN (SELECT id FROM parent) OR NOT parent_id IN (SELECT id FROM parent)"
+               })
+               |> Router.call(opts)
+    end
+
     @tag with_sql: [
            "CREATE TABLE parent (id INT PRIMARY KEY, value INT NOT NULL)",
            "CREATE TABLE child (id INT PRIMARY KEY, parent_id INT NOT NULL REFERENCES parent(id), value INT NOT NULL)",

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

@@ -206,6 +206,33 @@ defmodule Electric.Shapes.DnfPlanTest do
     end
   end
 
+  describe "compile/1 - mixed polarity" do
+    test "returns error when same subquery is used with both positive and negative polarity" do
+      # Parse with 2 separate sublinks, then remap $sublink/1 -> $sublink/0
+      # to simulate what Shape.new's canonicalize_where_sublink_refs does
+      # when the same subquery appears with opposite polarity.
+      {where, deps} =
+        parse_where_with_sublinks(
+          ~S"x IN (SELECT id FROM dep1) OR NOT x IN (SELECT id FROM dep2)",
+          2
+        )
+
+      # Remap $sublink/1 refs to $sublink/0 in the AST, simulating deduplication
+      remapped_eval = remap_sublink_ref(where.eval, "1", "0")
+
+      remapped_used_refs =
+        where.used_refs
+        |> Map.delete(["$sublink", "1"])
+
+      where = %{where | eval: remapped_eval, used_refs: remapped_used_refs}
+
+      # Only 1 dependency since both refs now point to the same sublink
+      shape = make_shape(where, [hd(deps)])
+      assert {:error, reason} = DnfPlan.compile(shape)
+      assert reason =~ "positive and negative polarity"
+    end
+  end
+
   describe "compile/1 - nested subqueries compile per level" do
     test "outer and inner shapes compile independently" do
       # Outer shape: x IN sq1 (where sq1 itself has subqueries)
@@ -823,4 +850,23 @@ defmodule Electric.Shapes.DnfPlanTest do
       shape_dependencies_handles: Enum.with_index(deps, fn _, i -> "dep_handle_#{i}" end)
     }
   end
+
+  # Recursively remap $sublink refs in an eval AST
+  defp remap_sublink_ref(%Parser.Ref{path: ["$sublink", from]} = ref, from, to) do
+    %{ref | path: ["$sublink", to]}
+  end
+
+  defp remap_sublink_ref(%Parser.Func{args: args} = func, from, to) do
+    %{func | args: Enum.map(args, &remap_sublink_ref(&1, from, to))}
+  end
+
+  defp remap_sublink_ref(%Parser.Array{elements: elements} = arr, from, to) do
+    %{arr | elements: Enum.map(elements, &remap_sublink_ref(&1, from, to))}
+  end
+
+  defp remap_sublink_ref(%Parser.RowExpr{elements: elements} = row, from, to) do
+    %{row | elements: Enum.map(elements, &remap_sublink_ref(&1, from, to))}
+  end
+
+  defp remap_sublink_ref(other, _from, _to), do: other
 end

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

@@ -882,31 +882,6 @@ defmodule Electric.Shapes.ShapeTest do
                  params: params
                )
     end
-
-    @tag with_sql: [
-           "CREATE TABLE IF NOT EXISTS project (id INT PRIMARY KEY, value INT NOT NULL)",
-           "CREATE TABLE IF NOT EXISTS item (id INT PRIMARY KEY, value INT NOT NULL)"
-         ]
-    test "rejects subquery shapes whose WHERE clause is too complex for DNF decomposition", %{
-      inspector: inspector
-    } do
-      # Build a WHERE with >100 disjuncts by cross-producting ORs with ANDs,
-      # plus a subquery to trigger the DNF validation path.
-      # (a1 OR a2 OR ... OR a11) AND (b1 OR b2 OR ... OR b11) = 121 disjuncts
-      group_a = Enum.map_join(1..11, " OR ", fn i -> "value = #{i}" end)
-      group_b = Enum.map_join(12..22, " OR ", fn i -> "value = #{i}" end)
-
-      where =
-        "(#{group_a}) AND (#{group_b}) AND value IN (SELECT value FROM project WHERE value > 5)"
-
-      assert {:error,
-              {:where,
-               "WHERE clause too complex for DNF decomposition" <> _rest}} =
-               Shape.new("item",
-                 inspector: inspector,
-                 where: where
-               )
-    end
   end
 
   describe "new!/2" do