fix: Correct dependency tracking for nested subqueries (#3757)

## Summary

- Fixes dependency tracking for nested subqueries when intermediate rows
change their linking column without changing the tracked column
- Previously, such updates were incorrectly filtered out, causing stale
tag tracking that led to incorrect row deletions

## Changes

- **shape.ex**: Keep updates with tag changes (`removed_move_tags !=
[]`) even if selected columns are unchanged
- **materializer.ex**: Skip value count updates when value is unchanged
to avoid spurious move_out/move_in events

## Test plan

- [x] Added unit tests in `shape_test.exs` for
`filtered_columns_changed?` behavior
- [x] Added unit tests in `materializer_test.exs` for tag-only update
handling
- [x] Integration tests in `subquery_dependency_update_test.exs` pass
- [x] Full test suite passes (1322 tests, 0 failures)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Bug Fixes**
* Fixed dependency tracking for nested subqueries so intermediate row
moves that don’t change tracked values no longer cause spurious
deletions.
* Preserved tag-only updates in change detection to prevent stale tag
tracking when rows move between parents.

* **Tests**
* Added extensive tests covering tag-only updates and nested subquery
dependency scenarios to validate correctness.

<sub>✏️ Tip: You can customize this high-level summary in your review
settings.</sub>
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: robacourt

.changeset/fix-nested-subquery-tag-tracking.md

@@ -0,0 +1,5 @@
+---
+"@core/sync-service": patch
+---
+
+Fix dependency tracking for nested subqueries when intermediate rows change their linking column without changing the tracked column. Previously, such updates were incorrectly filtered out, causing stale tag tracking that led to incorrect row deletions when the old parent lost its qualifying status.

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

@@ -327,20 +327,33 @@ defmodule Electric.Shapes.Consumer.Materializer do
           },
           {{index, tag_indices}, counts_and_events} ->
             # TODO: this is written as if it supports multiple selected columns, but it doesn't for now
-            if Enum.any?(state.columns, &is_map_key(record, &1)) do
-              {value, original_string} = cast!(record, state)
-              old_value = Map.fetch!(index, key)
-              index = Map.put(index, key, value)
+            columns_present = Enum.any?(state.columns, &is_map_key(record, &1))
+            has_tag_updates = removed_move_tags != []
 
+            if columns_present or has_tag_updates do
               tag_indices =
                 tag_indices
                 |> remove_row_from_tag_indices(key, removed_move_tags)
                 |> add_row_to_tag_indices(key, move_tags)
 
-              {{index, tag_indices},
-               counts_and_events
-               |> decrement_value(old_value, value_to_string(old_value, state))
-               |> increment_value(value, original_string)}
+              if columns_present do
+                {value, original_string} = cast!(record, state)
+                old_value = Map.fetch!(index, key)
+                index = Map.put(index, key, value)
+
+                # Skip decrement/increment dance if value hasn't changed to avoid
+                # spurious move_out/move_in events when only the tag changed
+                if old_value == value do
+                  {{index, tag_indices}, counts_and_events}
+                else
+                  {{index, tag_indices},
+                   counts_and_events
+                   |> decrement_value(old_value, value_to_string(old_value, state))
+                   |> increment_value(value, original_string)}
+                end
+              else
+                {{index, tag_indices}, counts_and_events}
+              end
             else
               # Nothing relevant to this materializer has been updated
               {{index, tag_indices}, counts_and_events}

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

@@ -613,7 +613,7 @@ defmodule Electric.Shapes.Shape do
     converted_changes
     |> Enum.map(&fill_move_tags(&1, shape, opts[:stack_id], opts[:shape_handle]))
     |> Enum.map(&filter_change_columns(&1, selected_columns))
-    |> Enum.filter(&filtered_columns_changed?/1)
+    |> Enum.filter(&should_keep_change?/1)
   end
 
   defp filter_change_columns(change, nil), do: change
@@ -684,10 +684,13 @@ defmodule Electric.Shapes.Shape do
     end)
   end
 
-  defp filtered_columns_changed?(%Changes.UpdatedRecord{old_record: record, record: record}),
+  defp should_keep_change?(%Changes.UpdatedRecord{removed_move_tags: removed_move_tags})
+       when removed_move_tags != [], do: true
+
+  defp should_keep_change?(%Changes.UpdatedRecord{old_record: record, record: record}),
     do: false
 
-  defp filtered_columns_changed?(_), do: true
+  defp should_keep_change?(_), do: true
 
   # If neither oid nor schema/table name matches, then shape is not affected
   def is_affected_by_relation_change?(

packages/sync-service/test/electric/shapes/consumer/materializer_test.exs

@@ -370,6 +370,150 @@ defmodule Electric.Shapes.Consumer.MaterializerTest do
     end
   end
 
+  describe "tag-only updates (value unchanged)" do
+    @tag snapshot_data: [%Changes.NewRecord{record: %{"id" => "1", "value" => "10"}}]
+    test "update with tag change but unchanged value updates tags without events", ctx do
+      ctx = with_materializer(ctx)
+
+      assert Materializer.get_link_values(ctx) == MapSet.new([10])
+
+      # Update where tags change but the tracked value stays the same
+      Materializer.new_changes(
+        ctx,
+        [
+          %Changes.UpdatedRecord{
+            key: ~s("public"."test_table"/"1"),
+            record: %{"id" => "1", "value" => "10"},
+            old_record: %{"id" => "1", "value" => "10"},
+            move_tags: ["new_tag"],
+            removed_move_tags: ["old_tag"]
+          }
+        ]
+      )
+
+      # Value should still be present
+      assert Materializer.get_link_values(ctx) == MapSet.new([10])
+
+      # No move events should be emitted since the value didn't change
+      refute_received {:materializer_changes, _, _}
+    end
+
+    @tag snapshot_data: {
+           [%Changes.NewRecord{record: %{"id" => "1", "value" => "10"}, move_tags: ["old_tag"]}],
+           []
+         }
+    test "tag is updated so subsequent move_out for old tag finds nothing", ctx do
+      ctx = with_materializer(ctx)
+
+      assert Materializer.get_link_values(ctx) == MapSet.new([10])
+
+      # Update that changes the tag from old_tag to new_tag but keeps value the same
+      Materializer.new_changes(
+        ctx,
+        [
+          %Changes.UpdatedRecord{
+            key: ~s("public"."test_table"/"1"),
+            record: %{"id" => "1", "value" => "10"},
+            old_record: %{"id" => "1", "value" => "10"},
+            move_tags: ["new_tag"],
+            removed_move_tags: ["old_tag"]
+          }
+        ]
+      )
+
+      # No events from the tag-only update
+      refute_received {:materializer_changes, _, _}
+
+      # Now send a move_out for the OLD tag - should find nothing since the row moved to new_tag
+      Materializer.new_changes(ctx, [
+        %{headers: %{event: "move-out", patterns: [%{pos: 0, value: "old_tag"}]}}
+      ])
+
+      # Value should still be present (row wasn't removed)
+      assert Materializer.get_link_values(ctx) == MapSet.new([10])
+
+      # No move events since the row was already moved to new_tag
+      refute_received {:materializer_changes, _, _}
+    end
+
+    @tag snapshot_data: {
+           [%Changes.NewRecord{record: %{"id" => "1", "value" => "10"}, move_tags: ["old_tag"]}],
+           []
+         }
+    test "move_out for new tag after tag update removes the row", ctx do
+      ctx = with_materializer(ctx)
+
+      assert Materializer.get_link_values(ctx) == MapSet.new([10])
+
+      # Update that changes the tag from old_tag to new_tag
+      Materializer.new_changes(
+        ctx,
+        [
+          %Changes.UpdatedRecord{
+            key: ~s("public"."test_table"/"1"),
+            record: %{"id" => "1", "value" => "10"},
+            old_record: %{"id" => "1", "value" => "10"},
+            move_tags: ["new_tag"],
+            removed_move_tags: ["old_tag"]
+          }
+        ]
+      )
+
+      refute_received {:materializer_changes, _, _}
+
+      # Now send a move_out for the NEW tag - should find and remove the row
+      Materializer.new_changes(ctx, [
+        %{headers: %{event: "move-out", patterns: [%{pos: 0, value: "new_tag"}]}}
+      ])
+
+      # Value should be gone
+      assert Materializer.get_link_values(ctx) == MapSet.new([])
+
+      # Should emit move_out event
+      assert_receive {:materializer_changes, _, %{move_out: [{10, "10"}]}}
+    end
+
+    @tag snapshot_data: {
+           [
+             %Changes.NewRecord{record: %{"id" => "1", "value" => "10"}, move_tags: ["tag_a"]},
+             %Changes.NewRecord{record: %{"id" => "2", "value" => "20"}, move_tags: ["tag_a"]}
+           ],
+           []
+         }
+    test "multiple rows with same tag, one updates tag, move_out only affects remaining", ctx do
+      ctx = with_materializer(ctx)
+
+      assert Materializer.get_link_values(ctx) == MapSet.new([10, 20])
+
+      # Row 1 moves from tag_a to tag_b, row 2 stays in tag_a
+      Materializer.new_changes(
+        ctx,
+        [
+          %Changes.UpdatedRecord{
+            key: ~s("public"."test_table"/"1"),
+            record: %{"id" => "1", "value" => "10"},
+            old_record: %{"id" => "1", "value" => "10"},
+            move_tags: ["tag_b"],
+            removed_move_tags: ["tag_a"]
+          }
+        ]
+      )
+
+      refute_received {:materializer_changes, _, _}
+
+      # move_out for tag_a should only affect row 2 (row 1 moved to tag_b)
+      Materializer.new_changes(ctx, [
+        %{headers: %{event: "move-out", patterns: [%{pos: 0, value: "tag_a"}]}}
+      ])
+
+      # Only value 10 should remain (row 1 is now under tag_b)
+      assert Materializer.get_link_values(ctx) == MapSet.new([10])
+
+      # Should emit move_out only for row 2's value
+      assert_receive {:materializer_changes, _, %{move_out: [{20, "20"}]}}
+    end
+  end
+
   describe "move_out events" do
     test "runtime move_out event removes rows matching the pattern", ctx do
       ctx = with_materializer(ctx)

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

@@ -210,6 +210,40 @@ defmodule Electric.Shapes.ShapeTest do
       assert Shape.convert_change(shape, non_matching_update) == []
     end
 
+    test "keeps update with tag changes even if filtered columns have not changed" do
+      shape = %Shape{
+        root_table: {"public", "table"},
+        root_table_id: @relation_id,
+        selected_columns: ["id"],
+        # tag_structure means this shape tracks tags based on the "parent_id" column
+        tag_structure: [["parent_id"]]
+      }
+
+      # Update where only parent_id changed (tag change), but id (selected column) didn't
+      update_with_tag_change = %UpdatedRecord{
+        relation: {"public", "table"},
+        old_record: %{"id" => 1, "parent_id" => "old_parent"},
+        record: %{"id" => 1, "parent_id" => "new_parent"}
+      }
+
+      result =
+        Shape.convert_change(shape, update_with_tag_change,
+          stack_id: "test_stack",
+          shape_handle: "test_handle"
+        )
+
+      # The change should be kept (not filtered out) because it has tag changes
+      assert length(result) == 1
+      [converted] = result
+
+      # The converted change should have removed_move_tags set (indicating the old tag)
+      assert converted.removed_move_tags != []
+      # And move_tags for the new tag
+      assert converted.move_tags != []
+      # Tags should be different since parent_id changed
+      assert converted.move_tags != converted.removed_move_tags
+    end
+
     test "correctly keeps updates with subqueries if the referenced set has not changed" do
       shape = %Shape{
         root_table: {"public", "table"},