Remove unique constraints (fixes #100 and #104) (#106)

* Add support for unique_constraint to introspection, and use it to determine if we should use DROP CONSTRAINT or DROP INDEX when altering or removing a unique (together) constraint. Fixes #100 and #104.

In the sys.indexes table, is_unique_constraint is true only when an actual constraint was created (using ALTER TABLE ... ADD CONSTRAINT ... UNIQUE).
Because this method is not suitable for nullable fields in practice (you cannot have more than one row with NULL), mssql-django always creates CREATE UNIQUE INDEX instead.
django-pyodbc-azure behaved differently and used a unique constraint whenever possible.

The problem that arises is that mssql-django assumes that an index is used to enforce all unique constraints, and always uses DROP INDEX to remove it.
When migrating a codebase from django-pyodbc-azure to mssql-django, this fails because the database contains actual unique constraints that need to be dropped using "ALTER TABLE ... DROP CONSTRAINT ...".

This commit adds support for is_unique_constraint to the introspection, so we can determine if the constraint is enforced by an actual SQL Server constraint or by a unique index.
Additionally, places that delete unique constraints have been refactored to use a common function that uses introspection to determine the proper method of deletion.

* Also treat primary keys as constraints instead of as indexes.

Co-authored-by: Ruben De Visscher <[email protected]>

Author: rubendv

mssql/introspection.py

@@ -253,9 +253,14 @@ def get_constraints(self, cursor, table_name):
                 constraints[constraint] = {
                     "columns": [],
                     "primary_key": kind.lower() == "primary key",
+                    # In the sys.indexes table, primary key indexes have is_unique_constraint as false,
+                    # but is_unique as true.
                     "unique": kind.lower() in ["primary key", "unique"],
+                    "unique_constraint": kind.lower() == "unique",
                     "foreign_key": (ref_table, ref_column) if kind.lower() == "foreign key" else None,
                     "check": False,
+                    # Potentially misleading: primary key and unique constraints still have indexes attached to them.
+                    # Should probably be updated with the additional info from the sys.indexes table we fetch later on.
                     "index": False,
                 }
             # Record the details
@@ -280,6 +285,7 @@ def get_constraints(self, cursor, table_name):
                     "columns": [],
                     "primary_key": False,
                     "unique": False,
+                    "unique_constraint": False,
                     "foreign_key": None,
                     "check": True,
                     "index": False,
@@ -291,6 +297,7 @@ def get_constraints(self, cursor, table_name):
             SELECT
                 i.name AS index_name,
                 i.is_unique,
+                i.is_unique_constraint,
                 i.is_primary_key,
                 i.type,
                 i.type_desc,
@@ -316,12 +323,13 @@ def get_constraints(self, cursor, table_name):
                 ic.index_column_id ASC
         """, [table_name])
         indexes = {}
-        for index, unique, primary, type_, desc, order, column in cursor.fetchall():
+        for index, unique, unique_constraint, primary, type_, desc, order, column in cursor.fetchall():
             if index not in indexes:
                 indexes[index] = {
                     "columns": [],
                     "primary_key": primary,
                     "unique": unique,
+                    "unique_constraint": unique_constraint,
                     "foreign_key": None,
                     "check": False,
                     "index": True,

mssql/schema.py

@@ -171,7 +171,12 @@ def alter_unique_together(self, model, old_unique_together, new_unique_together)
         news = {tuple(fields) for fields in new_unique_together}
         # Deleted uniques
         for fields in olds.difference(news):
-            self._delete_composed_index(model, fields, {'unique': True}, self.sql_delete_index)
+            meta_constraint_names = {constraint.name for constraint in model._meta.constraints}
+            meta_index_names = {constraint.name for constraint in model._meta.indexes}
+            columns = [model._meta.get_field(field).column for field in fields]
+            self._delete_unique_constraint_for_columns(
+                model, columns, exclude=meta_constraint_names | meta_index_names, strict=True)
+
         # Created uniques
         if django_version >= (4, 0):
             for field_names in news.difference(olds):
@@ -227,7 +232,7 @@ def _model_indexes_sql(self, model):
 
     def _db_table_constraint_names(self, db_table, column_names=None, column_match_any=False,
                                    unique=None, primary_key=None, index=None, foreign_key=None,
-                                   check=None, type_=None, exclude=None):
+                                   check=None, type_=None, exclude=None, unique_constraint=None):
         """
         Return all constraint names matching the columns and conditions. Modified from base `_constraint_names`
         `any_column_matches`=False: (default) only return constraints covering exactly `column_names`
@@ -247,6 +252,8 @@ def _db_table_constraint_names(self, db_table, column_names=None, column_match_a
             ):
                 if unique is not None and infodict['unique'] != unique:
                     continue
+                if unique_constraint is not None and infodict['unique_constraint'] != unique_constraint:
+                    continue
                 if primary_key is not None and infodict['primary_key'] != primary_key:
                     continue
                 if index is not None and infodict['index'] != index:
@@ -299,16 +306,7 @@ def _alter_field(self, model, old_field, new_field, old_type, new_type,
                 self.execute(self._delete_constraint_sql(self.sql_delete_fk, model, fk_name))
         # Has unique been removed?
         if old_field.unique and (not new_field.unique or self._field_became_primary_key(old_field, new_field)):
-            # Find the unique constraint for this field
-            constraint_names = self._constraint_names(model, [old_field.column], unique=True, primary_key=False)
-            if strict and len(constraint_names) != 1:
-                raise ValueError("Found wrong number (%s) of unique constraints for %s.%s" % (
-                    len(constraint_names),
-                    model._meta.db_table,
-                    old_field.column,
-                ))
-            for constraint_name in constraint_names:
-                self.execute(self._delete_constraint_sql(self.sql_delete_unique, model, constraint_name))
+            self._delete_unique_constraint_for_columns(model, [old_field.column], strict=strict)
         # Drop incoming FK constraints if the field is a primary key or unique,
         # which might be a to_field target, and things are going to change.
         drop_foreign_keys = (
@@ -694,21 +692,29 @@ def _delete_unique_constraints(self, model, old_field, new_field, strict=False):
             unique_columns.append([old_field.column])
         if unique_columns:
             for columns in unique_columns:
-                constraint_names_normal = self._constraint_names(model, columns, unique=True, index=False)
-                constraint_names_index = self._constraint_names(model, columns, unique=True, index=True)
-                constraint_names = constraint_names_normal + constraint_names_index
-                if strict and len(constraint_names) != 1:
-                    raise ValueError("Found wrong number (%s) of unique constraints for %s.%s" % (
-                        len(constraint_names),
-                        model._meta.db_table,
-                        old_field.column,
-                    ))
-                for constraint_name in constraint_names_normal:
-                    self.execute(self._delete_constraint_sql(self.sql_delete_unique, model, constraint_name))
-                # Unique indexes which are not table constraints must be deleted using the appropriate SQL.
-                # These may exist for example to enforce ANSI-compliant unique constraints on nullable columns.
-                for index_name in constraint_names_index:
-                    self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
+                self._delete_unique_constraint_for_columns(model, columns, strict=strict)
+
+    def _delete_unique_constraint_for_columns(self, model, columns, strict=False, **constraint_names_kwargs):
+        constraint_names_unique = self._db_table_constraint_names(
+            model._meta.db_table, columns, unique=True, unique_constraint=True, **constraint_names_kwargs)
+        constraint_names_primary = self._db_table_constraint_names(
+            model._meta.db_table, columns, unique=True, primary_key=True, **constraint_names_kwargs)
+        constraint_names_normal = constraint_names_unique + constraint_names_primary
+        constraint_names_index = self._db_table_constraint_names(
+            model._meta.db_table, columns, unique=True, unique_constraint=False, primary_key=False,
+            **constraint_names_kwargs)
+        constraint_names = constraint_names_normal + constraint_names_index
+        if strict and len(constraint_names) != 1:
+            raise ValueError("Found wrong number (%s) of unique constraints for columns %s" % (
+                len(constraint_names),
+                repr(columns),
+            ))
+        for constraint_name in constraint_names_normal:
+            self.execute(self._delete_constraint_sql(self.sql_delete_unique, model, constraint_name))
+        # Unique indexes which are not table constraints must be deleted using the appropriate SQL.
+        # These may exist for example to enforce ANSI-compliant unique constraints on nullable columns.
+        for index_name in constraint_names_index:
+            self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
 
     def _rename_field_sql(self, table, old_field, new_field, new_type):
         new_type = self._set_field_new_type_null_status(old_field, new_type)