fix: ensure primary field always exists on sync table (baserow#5089)

* Ensure primary field always exists on sync table

* Review changes

* Remove elidable from migration

Author: dimmur-brw

backend/src/baserow/contrib/database/data_sync/handler.py

@@ -622,8 +622,24 @@ def set_data_sync_synced_properties(
 
         handler = FieldHandler()
 
+        unique_primary_field = next(
+            (ep.field for ep in enabled_properties if ep.unique_primary), None
+        )
+
         for data_sync_property_instance in properties_to_be_removed:
             field = data_sync_property_instance.field
+            # If we're about to delete the primary field, first move primary to the
+            # unique_primary field so the table is never left without one.
+            if (
+                field.primary
+                and unique_primary_field
+                and unique_primary_field.id != field.id
+            ):
+                handler.change_primary_field(
+                    user=user,
+                    table=data_sync.table,
+                    new_primary_field=unique_primary_field,
+                )
             data_sync_property_instance.delete()
             handler.delete_field(
                 user=user,
@@ -655,7 +671,6 @@ def set_data_sync_synced_properties(
                 data_sync.table,
                 [field_kwargs.pop("name")],
             )
-            print(new_name)
             field = handler.create_field(
                 user=user,
                 table=data_sync.table,

backend/src/baserow/contrib/database/migrations/0207_fix_data_sync_missing_primary_field.py

@@ -0,0 +1,39 @@
+from django.db import migrations
+
+
+def fix_data_sync_missing_primary_field(apps, schema_editor):
+    """
+    Finds data sync tables that have no primary field and restores primary=True
+    on the field associated with their unique_primary synced property.
+
+    This fixes a bug where the primary field could be lost when a user changed
+    the primary to a non-unique_primary field, and that field was later removed
+    during a data sync.
+    """
+
+    DataSyncSyncedProperty = apps.get_model(
+        "database", "DataSyncSyncedProperty"
+    )
+    Field = apps.get_model("database", "Field")
+
+    tables_with_primary = Field.objects.filter(primary=True).values("table_id")
+    should_be_primary_field_ids = (
+        DataSyncSyncedProperty.objects.filter(unique_primary=True)
+        .exclude(data_sync__table_id__in=tables_with_primary)
+        .values("field_id")
+    )
+    Field.objects.filter(id__in=should_be_primary_field_ids).update(primary=True)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("database", "0207_add_view_default_values"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            fix_data_sync_missing_primary_field,
+            migrations.RunPython.noop,
+        ),
+    ]

changelog/entries/unreleased/bug/5088_ensure_primary_field_always_exists_on_sync_table.json

@@ -0,0 +1,9 @@
+{
+    "type": "bug",
+    "message": "Ensure primary field always exists on sync table",
+    "issue_origin": "github",
+    "issue_number": 5088,
+    "domain": "database",
+    "bullet_points": [],
+    "created_at": "2026-03-31"
+}

enterprise/backend/tests/baserow_enterprise_tests/data_sync/test_local_baserow_table_data_sync_type.py

@@ -1683,3 +1683,62 @@ def test_table_model_is_data_synced_table(enterprise_data_fixture):
 
     assert data_sync.table.is_data_synced_table
     assert not table_with_no_synced_table.is_data_synced_table
+
+
+@pytest.mark.django_db
+@pytest.mark.data_sync
+@override_settings(DEBUG=True)
+def test_sync_ensure_primary_field_always_exists(enterprise_data_fixture):
+    enterprise_data_fixture.enable_enterprise()
+    user = enterprise_data_fixture.create_user()
+
+    source_table = enterprise_data_fixture.create_database_table(
+        user=user, name="Source"
+    )
+    source_text_field = enterprise_data_fixture.create_text_field(
+        table=source_table, name="Name", primary=True
+    )
+    source_model = source_table.get_model()
+    source_model.objects.create(**{f"field_{source_text_field.id}": "Row 1"})
+
+    database = enterprise_data_fixture.create_database_application(user=user)
+    handler = DataSyncHandler()
+    data_sync = handler.create_data_sync_table(
+        user=user,
+        database=database,
+        table_name="Synced",
+        type_name="local_baserow_table",
+        synced_properties=["id", f"field_{source_text_field.id}"],
+        source_table_id=source_table.id,
+    )
+    handler.sync_data_sync_table(user=user, data_sync=data_sync)
+
+    synced_fields = specific_iterator(data_sync.table.field_set.all().order_by("id"))
+    row_id_field = synced_fields[0]
+    name_field = synced_fields[1]
+
+    assert row_id_field.primary is True
+    assert name_field.primary is False
+
+    FieldHandler().change_primary_field(
+        user=user, table=data_sync.table, new_primary_field=name_field
+    )
+    row_id_field.refresh_from_db()
+    name_field.refresh_from_db()
+    assert row_id_field.primary is False
+    assert name_field.primary is True
+
+    FieldHandler().update_field(
+        user=user,
+        field=source_text_field,
+        new_type_name="formula",
+        formula="'constant'",
+    )
+
+    handler.sync_data_sync_table(user=user, data_sync=data_sync)
+
+    assert not data_sync.table.field_set.filter(id=name_field.id).exists()
+
+    row_id_field.refresh_from_db()
+    assert row_id_field.primary is True
+    assert data_sync.table.field_set.filter(primary=True).count() == 1