# Copyright (c) Microsoft Corporation.
# Licensed under the BSD license.

import binascii
import datetime

from collections import defaultdict

from django.db.backends.base.schema import (
    BaseDatabaseSchemaEditor,
    _is_relevant_relation,
    _related_non_m2m_objects,
    logger,
)
from django.db.backends.ddl_references import (
    Columns,
    IndexName,
    Statement as DjStatement,
    Table,
)
from django import VERSION as django_version
from django.db.models import NOT_PROVIDED, Index, UniqueConstraint
from django.db.models.fields import AutoField, BigAutoField
from django.db.models.fields.related import ForeignKey
from django.db.models.sql.where import AND
from django.db.transaction import TransactionManagementError
from django.utils.encoding import force_str

if django_version >= (4, 0):
    from django.db.models.sql import Query
    from django.db.backends.ddl_references import Expressions


class Statement(DjStatement):
    def __hash__(self):
        return hash((self.template, str(self.parts['name'])))

    def __eq__(self, other):
        return self.template == other.template and str(self.parts['name']) == str(other.parts['name'])

    def rename_column_references(self, table, old_column, new_column):
        for part in self.parts.values():
            if hasattr(part, 'rename_column_references'):
                part.rename_column_references(table, old_column, new_column)
            condition = self.parts['condition']
            if condition:
                self.parts['condition'] = condition.replace(f'[{old_column}]', f'[{new_column}]')


class DatabaseSchemaEditor(BaseDatabaseSchemaEditor):

    _sql_check_constraint = " CONSTRAINT %(name)s CHECK (%(check)s)"
    _sql_select_default_constraint_name = "SELECT" \
                                          " d.name " \
                                          "FROM sys.default_constraints d " \
                                          "INNER JOIN sys.tables t ON" \
                                          " d.parent_object_id = t.object_id " \
                                          "INNER JOIN sys.columns c ON" \
                                          " d.parent_object_id = c.object_id AND" \
                                          " d.parent_column_id = c.column_id " \
                                          "INNER JOIN sys.schemas s ON" \
                                          " t.schema_id = s.schema_id " \
                                          "WHERE" \
                                          " t.name = %(table)s AND" \
                                          " c.name = %(column)s"
    sql_alter_column_default = "ADD DEFAULT %(default)s FOR %(column)s"
    sql_alter_column_no_default = "DROP CONSTRAINT %(column)s"
    sql_alter_column_not_null = "ALTER COLUMN %(column)s %(type)s NOT NULL"
    sql_alter_column_null = "ALTER COLUMN %(column)s %(type)s NULL"
    sql_alter_column_type = "ALTER COLUMN %(column)s %(type)s"
    sql_create_column = "ALTER TABLE %(table)s ADD %(column)s %(definition)s"
    sql_delete_column = "ALTER TABLE %(table)s DROP COLUMN %(column)s"
    sql_delete_default = "ALTER TABLE %(table)s DROP CONSTRAINT %(name)s"
    sql_delete_index = "DROP INDEX %(name)s ON %(table)s"
    sql_delete_table = """
        DECLARE @sql_foreign_constraint_name nvarchar(128)
        DECLARE @sql_drop_constraint nvarchar(300)
        WHILE EXISTS(SELECT 1
            FROM sys.foreign_keys
            WHERE referenced_object_id = object_id('%(table)s'))
        BEGIN
            SELECT TOP 1 @sql_foreign_constraint_name = name
            FROM sys.foreign_keys
            WHERE referenced_object_id = object_id('%(table)s')
            SELECT
            @sql_drop_constraint = 'ALTER TABLE [' + OBJECT_NAME(parent_object_id) + '] ' +
            'DROP CONSTRAINT [' + @sql_foreign_constraint_name + '] '
            FROM sys.foreign_keys
            WHERE referenced_object_id = object_id('%(table)s') and name = @sql_foreign_constraint_name
            exec sp_executesql @sql_drop_constraint
        END
        DROP TABLE %(table)s
"""
    sql_rename_column = "EXEC sp_rename '%(table)s.%(old_column)s', %(new_column)s, 'COLUMN'"
    sql_rename_table = "EXEC sp_rename %(old_table)s, %(new_table)s"
    sql_create_unique_null = "CREATE UNIQUE INDEX %(name)s ON %(table)s(%(columns)s) " \
                             "WHERE %(columns)s IS NOT NULL"
    sql_alter_table_comment= """
        IF NOT EXISTS (SELECT NULL FROM sys.extended_properties ep
            WHERE ep.major_id = OBJECT_ID('%(table)s')
            AND ep.name = 'MS_Description'
            AND ep.minor_id = 0)
                        EXECUTE sp_addextendedproperty 
                        @name = 'MS_Description', @value = %(comment)s, 
                        @level0type = 'SCHEMA', @level0name = 'dbo',
                        @level1type = 'TABLE', @level1name = %(table)s
            ELSE
                        EXECUTE sp_updateextendedproperty 
                        @name = 'MS_Description', @value = %(comment)s,
                        @level0type = 'SCHEMA', @level0name = 'dbo',
                        @level1type = 'TABLE', @level1name = %(table)s
    """
    sql_alter_column_comment= """
        IF NOT EXISTS (SELECT NULL FROM sys.extended_properties ep
            WHERE ep.major_id = OBJECT_ID('%(table)s')
            AND ep.name = 'MS_Description'
            AND ep.minor_id = (SELECT column_id FROM sys.columns 
                            WHERE name = '%(column)s'
                            AND object_id = OBJECT_ID('%(table)s')))
                EXECUTE sp_addextendedproperty 
                @name = 'MS_Description', @value = %(comment)s, 
                @level0type = 'SCHEMA', @level0name = 'dbo',
                @level1type = 'TABLE', @level1name = %(table)s,
                @level2type = 'COLUMN', @level2name = %(column)s
            ELSE
                EXECUTE sp_updateextendedproperty 
                @name = 'MS_Description', @value = %(comment)s,
                @level0type = 'SCHEMA', @level0name = 'dbo',
                @level1type = 'TABLE', @level1name = %(table)s,
                @level2type = 'COLUMN', @level2name = %(column)s
    """
    _deferred_unique_indexes = defaultdict(list)

    def _alter_column_default_sql(self, model, old_field, new_field, drop=False):
        """
        Hook to specialize column default alteration.

        Return a (sql, params) fragment to add or drop (depending on the drop
        argument) a default to new_field's column.
        """
        new_default = self.effective_default(new_field)
        default = '%s'
        params = [new_default]
        column = self.quote_name(new_field.column)

        if drop:
            params = []
            # SQL Server requires the name of the default constraint
            result = self.execute(
                self._sql_select_default_constraint_name % {
                    "table": self.quote_value(model._meta.db_table),
                    "column": self.quote_value(new_field.column),
                },
                has_result=True
            )
            if result:
                for row in result:
                    column = self.quote_name(next(iter(row)))
        elif self.connection.features.requires_literal_defaults:
            # Some databases (Oracle) can't take defaults as a parameter
            # If this is the case, the SchemaEditor for that database should
            # implement prepare_default().
            default = self.prepare_default(new_default)
            params = []

        new_db_params = new_field.db_parameters(connection=self.connection)
        sql = self.sql_alter_column_no_default if drop else self.sql_alter_column_default
        return (
            sql % {
                'column': column,
                'type': new_db_params['type'],
                'default': default,
            },
            params,
        )    

    def _alter_column_database_default_sql(
        self, model, old_field, new_field, drop=False
    ):
        """
        Hook to specialize column database default alteration.

        Return a (sql, params) fragment to add or drop (depending on the drop
        argument) a default to new_field's column.
        """
        column = self.quote_name(new_field.column)

        if drop:
            # SQL Server requires the name of the default constraint
            result = self.execute(
                self._sql_select_default_constraint_name % {
                    "table": self.quote_value(model._meta.db_table),
                    "column": self.quote_value(new_field.column),
                },
                has_result=True
            )
            if result:
                for row in result:
                    column = self.quote_name(next(iter(row)))

            sql = self.sql_alter_column_no_default
            default_sql = ""
            params = []
        else:
            sql = self.sql_alter_column_default
            default_sql, params = self.db_default_sql(new_field)

        new_db_params = new_field.db_parameters(connection=self.connection)
        return (
            sql
            % {
                "column": column,
                "type": new_db_params["type"],
                "default": default_sql,
            },
            params,
        )

    def _alter_column_comment_sql(self, model, new_field, new_type, new_db_comment):
        return (
            self.sql_alter_column_comment
            % {
                "table": self.quote_name(model._meta.db_table),
                "column": new_field.column,
                "comment": self._comment_sql(new_db_comment),
            },
            [],
        )

    def _alter_column_null_sql(self, model, old_field, new_field):
        """
        Hook to specialize column null alteration.

        Return a (sql, params) fragment to set a column to null or non-null
        as required by new_field, or None if no changes are required.
        """
        if (self.connection.features.interprets_empty_strings_as_nulls and
                new_field.get_internal_type() in ("CharField", "TextField")):
            # The field is nullable in the database anyway, leave it alone.
            return
        else:
            new_db_params = new_field.db_parameters(connection=self.connection)
            sql = self.sql_alter_column_null if new_field.null else self.sql_alter_column_not_null
            return (
                sql % {
                    'column': self.quote_name(new_field.column),
                    'type': new_db_params['type'],
                },
                [],
            )

    if django_version >= (4, 2):
        def _alter_column_type_sql(self, model, old_field, new_field, new_type, old_collation, new_collation):
            new_type = self._set_field_new_type_null_status(old_field, new_type)
            return super()._alter_column_type_sql(model, old_field, new_field, new_type, old_collation, new_collation)
    else:
        def _alter_column_type_sql(self, model, old_field, new_field, new_type):
            new_type = self._set_field_new_type_null_status(old_field, new_type)
            return super()._alter_column_type_sql(model, old_field, new_field, new_type)

    def alter_unique_together(self, model, old_unique_together, new_unique_together):
        """
        Deal with a model changing its unique_together. The input
        unique_togethers must be doubly-nested, not the single-nested
        ["foo", "bar"] format.
        """
        olds = {tuple(fields) for fields in old_unique_together}
        news = {tuple(fields) for fields in new_unique_together}
        # Deleted uniques
        for fields in olds.difference(news):
            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):
                fields = [model._meta.get_field(field) for field in field_names]
                columns = [model._meta.get_field(field).column for field in field_names]
                condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
                sql = self._create_unique_sql(model, fields, condition=condition)
                self.execute(sql)
        else:
            for fields in news.difference(olds):
                columns = [model._meta.get_field(field).column for field in fields]
                condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
                sql = self._create_unique_sql(model, columns, condition=condition)
                self.execute(sql)

    def _model_indexes_sql(self, model):
        """
        Return a list of all index SQL statements (field indexes,
        Meta.indexes) for the specified model.
        """
        if not model._meta.managed or model._meta.proxy or model._meta.swapped:
            return []
        output = []
        for field in model._meta.local_fields:
            output.extend(self._field_indexes_sql(model, field))
         # meta.index_together is removed in Django 5.1, so add a version check to handle compatibility
        if django_version < (5, 1):
            # Iterate over each set of field names defined in index_together
            for field_names in model._meta.index_together:
               # Get the actual field objects for each field name
               fields = [model._meta.get_field(field) for field in field_names]
               # Generate the SQL statement to create the index for these fields
               sql=self._create_index_sql(model, fields, suffix="_idx")
               # If SQL was generated (not None), add it to the output list
               if sql:
                 output.append(sql)

        if django_version >= (4, 0):
            for field_names in model._meta.unique_together:
                fields = [model._meta.get_field(field) for field in field_names]
                columns = [model._meta.get_field(field).column for field in field_names]
                condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
                sql = self._create_unique_sql(model, fields, condition=condition)
                output.append(sql)
        else:
            for field_names in model._meta.unique_together:
                columns = [model._meta.get_field(field).column for field in field_names]
                condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
                sql = self._create_unique_sql(model, columns, condition=condition)
                output.append(sql)
        for index in model._meta.indexes:
            if django_version >= (3, 2) and (
                not index.contains_expressions or
                self.connection.features.supports_expression_indexes
            ):
                output.append(index.create_sql(model, self))
            else:
                output.append(index.create_sql(model, self))
        return output

    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, unique_constraint=None):
        """
        Return all constraint names matching the columns and conditions.
        Modified from base `_constraint_names` but with the following new arguments:
         - `unique_constraint` which explicitly finds unique implemented by CONSTRAINT not by an INDEX
         - `column_match_any`:
                False: (default) only return constraints covering exactly `column_names`
                True : return any constraints which include at least 1 of `column_names`
        """
        if column_names is not None:
            column_names = [
                self.connection.introspection.identifier_converter(name)
                for name in column_names
            ]
        with self.connection.cursor() as cursor:
            constraints = self.connection.introspection.get_constraints(cursor, db_table)
        result = []
        for name, infodict in constraints.items():
            if column_names is None or column_names == infodict['columns'] or (
                column_match_any and any(col in infodict['columns'] for col in column_names)
            ):
                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:
                    continue
                if check is not None and infodict['check'] != check:
                    continue
                if foreign_key is not None and not infodict['foreign_key']:
                    continue
                if type_ is not None and infodict['type'] != type_:
                    continue
                if not exclude or name not in exclude:
                    result.append(name)
        return result

    def _db_table_delete_constraint_sql(self, template, db_table, name):
        return Statement(
            template,
            table=Table(db_table, self.quote_name),
            name=self.quote_name(name),
            include=''
        )

    def _delete_deferred_unique_indexes_for_field(self, field):
        deferred_statements = self._deferred_unique_indexes.get(str(field), [])
        for stmt in deferred_statements:
            if stmt in self.deferred_sql:
                self.deferred_sql.remove(stmt)

    def _add_deferred_unique_index_for_field(self, field, statement):
        self._deferred_unique_indexes[str(field)].append(statement)

    def _column_generated_sql(self, field):
        """Return the SQL to use in a GENERATED ALWAYS clause."""
        expression_sql, params = field.generated_sql(self.connection)
        persistency_sql = "PERSISTED" if field.db_persist else ""
        if self.connection.features.requires_literal_defaults:
            expression_sql = expression_sql % tuple(self.quote_value(p) for p in params)
            params = ()
        return f"GENERATED ALWAYS AS ({expression_sql}) {persistency_sql}", params

    def _alter_field(self, model, old_field, new_field, old_type, new_type,
                     old_db_params, new_db_params, strict=False):
        """Actually perform a "physical" (non-ManyToMany) field update."""

        # the backend doesn't support altering a column to/from AutoField as
        # SQL Server cannot alter columns to add and remove IDENTITY properties
        old_is_auto = False
        new_is_auto = False
        for t in (AutoField, BigAutoField):
            if isinstance(old_field, t):
                old_is_auto = True
            if isinstance(new_field, t):
                new_is_auto = True
        if (old_is_auto and not new_is_auto) or (not old_is_auto and new_is_auto):
            raise NotImplementedError("the backend doesn't support altering from %s to %s." %
                (old_field.get_internal_type(), new_field.get_internal_type()))

        # Drop any FK constraints, we'll remake them later
        fks_dropped = set()
        if (
            old_field.remote_field 
            and old_field.db_constraint 
            and (django_version < (4,2) 
                or 
                (django_version >= (4, 2) 
                and self._field_should_be_altered(
                    old_field,
                    new_field,
                    ignore={"db_comment"})
                )
            )
        ):
            # Drop index, SQL Server requires explicit deletion
            if (
                not hasattr(new_field, "db_constraint")
                or not new_field.db_constraint
            ):
                if(django_version < (4, 2) 
                   or (
                       not isinstance(new_field, ForeignKey)
                       or type(new_field.db_comment) == type(None)
                       or "fk_on_delete_keep_index" not in new_field.db_comment
                       )
                   ):
                    index_names = self._constraint_names(model, [old_field.column], index=True)
                    for index_name in index_names:
                        self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))

            fk_names = self._constraint_names(model, [old_field.column], foreign_key=True)
            if strict and len(fk_names) != 1:
                raise ValueError("Found wrong number (%s) of foreign key constraints for %s.%s" % (
                    len(fk_names),
                    model._meta.db_table,
                    old_field.column,
                ))
            for fk_name in fk_names:
                fks_dropped.add((old_field.column,))
                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)):
            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 = (
            (
                (old_field.primary_key and new_field.primary_key) or
                (old_field.unique and new_field.unique)
            ) and old_type != new_type
        )
        if drop_foreign_keys:
            # '_meta.related_field' also contains M2M reverse fields, these
            # will be filtered out
            for _old_rel, new_rel in _related_non_m2m_objects(old_field, new_field):
                rel_fk_names = self._constraint_names(
                    new_rel.related_model, [new_rel.field.column], foreign_key=True
                )
                for fk_name in rel_fk_names:
                    self.execute(self._delete_constraint_sql(self.sql_delete_fk, new_rel.related_model, fk_name))
        # If working with an AutoField or BigAutoField drop all indexes on the related table
        # This is needed when doing ALTER column statements on IDENTITY fields
        # https://stackoverflow.com/questions/33429775/sql-server-alter-table-alter-column-giving-set-option-error
        for t in (AutoField, BigAutoField):
            if isinstance(old_field, t) or isinstance(new_field, t):
                index_names = self._constraint_names(model, index=True)
                for index_name in index_names:
                    self.execute(
                        self._delete_constraint_sql(self.sql_delete_index, model, index_name)
                    )
                break
        # Removed an index? (no strict check, as multiple indexes are possible)
        # Remove indexes if db_index switched to False or a unique constraint
        # will now be used in lieu of an index. The following lines from the
        # truth table show all True cases; the rest are False:
        #
        # old_field.db_index | old_field.unique | new_field.db_index | new_field.unique
        # ------------------------------------------------------------------------------
        # True               | False            | False              | False
        # True               | False            | False              | True
        # True               | False            | True               | True
        if (old_field.db_index and not old_field.unique and (not new_field.db_index or new_field.unique)) or (
                # Drop indexes on nvarchar columns that are changing to a different type
                # SQL Server requires explicit deletion
                (old_field.db_index or old_field.unique) and (
                    (old_type.startswith('nvarchar') and not new_type.startswith('nvarchar'))
                )):
            # Find the index for this field
            meta_index_names = {index.name for index in model._meta.indexes}
            # Retrieve only BTREE indexes since this is what's created with
            # db_index=True.
            index_names = self._constraint_names(model, [old_field.column], index=True, type_=Index.suffix)
            for index_name in index_names:
                if index_name not in meta_index_names:
                    # The only way to check if an index was created with
                    # db_index=True or with Index(['field'], name='foo')
                    # is to look at its name (refs #28053).
                    self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
        # Change check constraints?
        if (old_db_params['check'] != new_db_params['check'] and old_db_params['check']) or (
            # SQL Server requires explicit deletion befor altering column type with the same constraint
            old_db_params['check'] == new_db_params['check'] and old_db_params['check'] and
            old_db_params['type'] != new_db_params['type']
        ):
            constraint_names = self._constraint_names(model, [old_field.column], check=True)
            if strict and len(constraint_names) != 1:
                raise ValueError("Found wrong number (%s) of check 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_check, model, constraint_name))
        # Have they renamed the column?
        if old_field.column != new_field.column:
            sql_restore_index = ''
            # Drop any unique indexes which include the column to be renamed
            index_names = self._db_table_constraint_names(
                db_table=model._meta.db_table, column_names=[old_field.column], column_match_any=True,
                index=True, unique=True,
            )
            for index_name in index_names:
                # Before dropping figure out how to recreate it afterwards
                with self.connection.cursor() as cursor:
                    cursor.execute(f"""
                    SELECT COL_NAME(ic.object_id,ic.column_id) AS column_name,
                           filter_definition
                    FROM sys.indexes AS i
                    INNER JOIN sys.index_columns AS ic
                        ON i.object_id = ic.object_id AND i.index_id = ic.index_id
                    WHERE i.object_id = OBJECT_ID('{model._meta.db_table}')
                    and i.name = '{index_name}'
                    """)
                    result = cursor.fetchall()
                    columns_to_recreate_index = ', '.join(['%s' % self.quote_name(column[0]) for column in result])
                    filter_definition = result[0][1]
                sql_restore_index += 'CREATE UNIQUE INDEX %s ON %s (%s) WHERE %s;' % (
                    index_name, model._meta.db_table, columns_to_recreate_index, filter_definition)
                self.execute(self._db_table_delete_constraint_sql(
                    self.sql_delete_index, model._meta.db_table, index_name))
            self.execute(self._rename_field_sql(model._meta.db_table, old_field, new_field, new_type))
            # Restore index(es) now the column has been renamed
            if sql_restore_index:
                self.execute(sql_restore_index.replace(f'[{old_field.column}]', f'[{new_field.column}]'))
            # Rename all references to the renamed column.
            for sql in self.deferred_sql:
                if isinstance(sql, DjStatement):
                    sql.rename_column_references(model._meta.db_table, old_field.column, new_field.column)

        # Next, start accumulating actions to do
        actions = []
        null_actions = []
        post_actions = []
        # Type or comment change?
        if old_type != new_type or (django_version >= (4, 2) and
                self.connection.features.supports_comments
                and old_field.db_comment != new_field.db_comment
            ):
            if django_version >= (4, 2):
                fragment, other_actions = self._alter_column_type_sql(
                    model, old_field, new_field, new_type, old_collation=None, new_collation=None
                )
            else:
                fragment, other_actions =  self._alter_column_type_sql(model, old_field, new_field, new_type)
            actions.append(fragment)
            post_actions.extend(other_actions)
            # Drop unique constraint, SQL Server requires explicit deletion
            self._delete_unique_constraints(model, old_field, new_field, strict)
            # Drop indexes, SQL Server requires explicit deletion
            self._delete_indexes(model, old_field, new_field)
        # db_default change?
        if django_version >= (5,0):
            if new_field.db_default is not NOT_PROVIDED:
                if (
                    old_field.db_default is NOT_PROVIDED
                    or new_field.db_default != old_field.db_default
                ):
                    actions.append(
                        self._alter_column_database_default_sql(model, old_field, new_field)
                    )
            elif old_field.db_default is not NOT_PROVIDED:
                actions.append(
                    self._alter_column_database_default_sql(
                        model, old_field, new_field, drop=True
                    )
                )
        # When changing a column NULL constraint to NOT NULL with a given
        # default value, we need to perform 4 steps:
        #  1. Add a default for new incoming writes
        #  2. Update existing NULL rows with new default
        #  3. Replace NULL constraint with NOT NULL
        #  4. Drop the default again.
        # Default change?
        old_default = self.effective_default(old_field)
        new_default = self.effective_default(new_field)
        needs_database_default = (
            old_field.null and
            not new_field.null and
            old_default != new_default and
            new_default is not None and
            not self.skip_default(new_field)
        )
        if django_version >= (5,0):
            needs_database_default = needs_database_default and new_field.db_default is NOT_PROVIDED
        if needs_database_default:
            actions.append(self._alter_column_default_sql(model, old_field, new_field))
        # Nullability change?
        if old_field.null != new_field.null:
            fragment = self._alter_column_null_sql(model, old_field, new_field)
            if fragment:
                null_actions.append(fragment)
                # Drop unique constraint, SQL Server requires explicit deletion
                self._delete_unique_constraints(model, old_field, new_field, strict)
                # Drop indexes, SQL Server requires explicit deletion
                indexes_dropped = self._delete_indexes(model, old_field, new_field)
                auto_index_names = []
                for index_from_meta in model._meta.indexes:
                    auto_index_names.append(self._create_index_name(model._meta.db_table, index_from_meta.fields))

                if (
                    new_field.get_internal_type() not in ("JSONField", "TextField") and
                    (old_field.db_index or not new_field.db_index) and
                    new_field.db_index or
                    ((indexes_dropped and sorted(indexes_dropped) == sorted([index.name for index in model._meta.indexes])) or
                     (indexes_dropped and sorted(indexes_dropped) == sorted(auto_index_names)))
                ):
                    create_index_sql_statement = self._create_index_sql(model, [new_field])
                    if create_index_sql_statement.__str__() not in [sql.__str__() for sql in self.deferred_sql]:
                        post_actions.append((create_index_sql_statement, ()))
        # Only if we have a default and there is a change from NULL to NOT NULL
        four_way_default_alteration = (
            (new_field.has_default() or (django_version >= (5,0) and new_field.db_default is not NOT_PROVIDED)) and
            (old_field.null and not new_field.null)
        )
        if actions or null_actions:
            if not four_way_default_alteration:
                # If we don't have to do a 4-way default alteration we can
                # directly run a (NOT) NULL alteration
                actions = actions + null_actions
            # Combine actions together if we can (e.g. postgres)
            if self.connection.features.supports_combined_alters and actions:
                sql, params = tuple(zip(*actions))
                actions = [(", ".join(sql), sum(params, []))]
            # Apply those actions
            for sql, params in actions:
                self.execute(
                    self.sql_alter_column % {
                        "table": self.quote_name(model._meta.db_table),
                        "changes": sql,
                    },
                    params,
                )
            if four_way_default_alteration:
                if django_version >= (5,0) and new_field.db_default is not NOT_PROVIDED:
                    default_sql, params = self.db_default_sql(new_field)
                else:
                    default_sql = "%s"
                    params = [new_default]
                # Update existing rows with default value
                self.execute(
                    self.sql_update_with_default % {
                        "table": self.quote_name(model._meta.db_table),
                        "column": self.quote_name(new_field.column),
                        "default": default_sql,
                    },
                    params,
                )
                # Since we didn't run a NOT NULL change before we need to do it
                # now
                for sql, params in null_actions:
                    self.execute(
                        self.sql_alter_column % {
                            "table": self.quote_name(model._meta.db_table),
                            "changes": sql,
                        },
                        params,
                    )
        if post_actions:
            for sql, params in post_actions:
                self.execute(sql, params)
        # If primary_key changed to False, delete the primary key constraint.
        if old_field.primary_key and not new_field.primary_key:
            self._delete_primary_key(model, strict)
        # Added a unique?
        if self._unique_should_be_added(old_field, new_field):
            if (self.connection.features.supports_nullable_unique_constraints and
                    not new_field.many_to_many and new_field.null):

                self.execute(
                    self._create_index_sql(
                        model, [new_field], sql=self.sql_create_unique_null, suffix="_uniq"
                    )
                )
            else:
                if django_version >= (4, 0):
                    self.execute(self._create_unique_sql(model, [new_field]))
                else:
                    self.execute(self._create_unique_sql(model, [new_field.column]))
            self._delete_deferred_unique_indexes_for_field(new_field)
        # Added an index?
        # constraint will no longer be used in lieu of an index. The following
        # lines from the truth table show all True cases; the rest are False:
        #
        # old_field.db_index | old_field.unique | new_field.db_index | new_field.unique
        # ------------------------------------------------------------------------------
        # False              | False            | True               | False
        # False              | True             | True               | False
        # True               | True             | True               | False
        if (not old_field.db_index or old_field.unique) and new_field.db_index and not new_field.unique:
            self.execute(self._create_index_sql(model, [new_field]))

        # Restore indexes & unique constraints deleted above, SQL Server requires explicit restoration
        if (old_type != new_type or (old_field.null != new_field.null)) and (
            old_field.column == new_field.column  # column rename is handled separately above
        ):
            # Restore unique constraints
            # Note: if nullable they are implemented via an explicit filtered UNIQUE INDEX (not CONSTRAINT)
            # in order to get ANSI-compliant NULL behaviour (i.e. NULL != NULL, multiple are allowed)
            # Note: Don't restore primary keys, we need to re-create those seperately
            if old_field.unique and new_field.unique and not new_field.primary_key:
                if new_field.null:
                    self.execute(
                        self._create_index_sql(
                            model, [old_field], sql=self.sql_create_unique_null, suffix="_uniq"
                        )
                    )
                else:
                    if django_version >= (4, 0):
                        self.execute(self._create_unique_sql(model, [old_field]))
                    else:
                        self.execute(self._create_unique_sql(model, columns=[old_field.column]))
                self._delete_deferred_unique_indexes_for_field(old_field)
            else:
                if django_version >= (4, 0):
                    for field_names in model._meta.unique_together:
                        columns = [model._meta.get_field(field).column for field in field_names]
                        fields = [model._meta.get_field(field) for field in field_names]
                        if old_field.column in columns:
                            condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
                            self.execute(self._create_unique_sql(model, fields, condition=condition))
                else:
                    for fields in model._meta.unique_together:
                        columns = [model._meta.get_field(field).column for field in fields]
                        if old_field.column in columns:
                            condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
                            self.execute(self._create_unique_sql(model, columns, condition=condition))
            # Restore primary keys
            if old_field.primary_key and new_field.primary_key:
                self.execute(
                    self.sql_create_pk % {
                        "table": self.quote_name(model._meta.db_table),
                        "name": self.quote_name(
                            self._create_index_name(model._meta.db_table, [new_field.column], suffix="_pk")
                        ),
                        "columns": self.quote_name(new_field.column),
                    }
                )
            # Restore unqiue_together
            # If we have ALTERed an AutoField or BigAutoField we need to recreate all unique_together clauses
            for t in (AutoField, BigAutoField):
                if isinstance(old_field, t) or isinstance(new_field, t):
                    for field_names in model._meta.unique_together:
                        columns = [model._meta.get_field(field).column for field in field_names]
                        fields = [model._meta.get_field(field) for field in field_names]
                        condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
                        # We need to pass fields instead of columns when using >= Django 4.0 because
                        # of a backwards incompatible change to _create_unique_sql
                        if django_version >= (4, 0):
                            self.execute(
                                self._create_unique_sql(model, fields, condition=condition)
                            )
                        else:
                            self.execute(
                                self._create_unique_sql(model, columns, condition=condition)
                            )
                    break

            # Restore indexes
            # If we have ALTERed an AutoField or BigAutoField we need to recreate all indexes
            for t in (AutoField, BigAutoField):
                if isinstance(old_field, t) or isinstance(new_field, t):
                    for field in model._meta.fields:
                        if field.db_index:
                            self.execute(
                                self._create_index_sql(model, [field])
                            )
                    break
            index_columns = []
            if old_field.db_index and new_field.db_index:
                index_columns.append([old_field])
            else:
                # Handle index_together for only django version < 5.1
                if django_version < (5, 1):
                   # Get the field objects for each field name in the index_together. 
                   for fields in model._meta.index_together:
                      # If the old field's column is among the columns for this index,
                      # add this set of columns to index_columns for later index recreation.
                      columns = [model._meta.get_field(field) for field in fields]
                      if old_field.column in [c.column for c in columns]:
                         index_columns.append(columns)
            if index_columns:
                for columns in index_columns:
                    create_index_sql_statement = self._create_index_sql(model, columns)
                    if (create_index_sql_statement.__str__()
                            not in [sql.__str__() for sql in self.deferred_sql] + [statement[0].__str__() for statement in post_actions]
                            ):
                        self.execute(create_index_sql_statement)

        # Type alteration on primary key? Then we need to alter the column
        # referring to us.
        rels_to_update = []
        if old_field.primary_key and new_field.primary_key and old_type != new_type:
            rels_to_update.extend(_related_non_m2m_objects(old_field, new_field))
        # Changed to become primary key?
        if self._field_became_primary_key(old_field, new_field):
            # Make the new one
            self.execute(
                self.sql_create_pk % {
                    "table": self.quote_name(model._meta.db_table),
                    "name": self.quote_name(
                        self._create_index_name(model._meta.db_table, [new_field.column], suffix="_pk")
                    ),
                    "columns": self.quote_name(new_field.column),
                }
            )
            # Update all referencing columns
            rels_to_update.extend(_related_non_m2m_objects(old_field, new_field))
        # Handle our type alters on the other end of rels from the PK stuff above
        for old_rel, new_rel in rels_to_update:
            rel_db_params = new_rel.field.db_parameters(connection=self.connection)
            rel_type = rel_db_params['type']
            if django_version >= (4, 2):
                fragment, other_actions = self._alter_column_type_sql(
                    new_rel.related_model, old_rel.field, new_rel.field, rel_type, old_collation=None, new_collation=None
                )
            else:
                fragment, other_actions = self._alter_column_type_sql(
                    new_rel.related_model, old_rel.field, new_rel.field, rel_type
                )
            # Drop related_model indexes, so it can be altered
            index_names = self._db_table_constraint_names(old_rel.related_model._meta.db_table, index=True)
            for index_name in index_names:
                self.execute(self._db_table_delete_constraint_sql(
                    self.sql_delete_index, old_rel.related_model._meta.db_table, index_name))
            self.execute(
                self.sql_alter_column % {
                    "table": self.quote_name(new_rel.related_model._meta.db_table),
                    "changes": fragment[0],
                },
                fragment[1],
            )
            for sql, params in other_actions:
                self.execute(sql, params)
            # Restore related_model indexes
            for field in new_rel.related_model._meta.fields:
                if field.db_index:
                    self.execute(
                        self._create_index_sql(new_rel.related_model, [field])
                    )
            # Restore unique_together clauses
            for field_names in new_rel.related_model._meta.unique_together:
                columns = [new_rel.related_model._meta.get_field(field).column for field in field_names]
                fields = [new_rel.related_model._meta.get_field(field) for field in field_names]
                condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
                # We need to pass fields instead of columns when using >= Django 4.0 because
                # of a backwards incompatible change to _create_unique_sql
                if django_version >= (4, 0):
                    self.execute(
                        self._create_unique_sql(new_rel.related_model, fields, condition=condition)
                    )
                else:
                    self.execute(
                        self._create_unique_sql(new_rel.related_model, columns, condition=condition)
                    )
        # Does it have a foreign key?
        if (new_field.remote_field and
                (fks_dropped or not old_field.remote_field or not old_field.db_constraint) and
                new_field.db_constraint):
            self.execute(self._create_fk_sql(model, new_field, "_fk_%(to_table)s_%(to_column)s"))
        # Rebuild FKs that pointed to us if we previously had to drop them
        if drop_foreign_keys:
            for rel in new_field.model._meta.related_objects:
                if _is_relevant_relation(rel, new_field) and rel.field.db_constraint:
                    self.execute(self._create_fk_sql(rel.related_model, rel.field, "_fk"))
        # Does it have check constraints we need to add?
        if (old_db_params['check'] != new_db_params['check'] and new_db_params['check']) or (
            # SQL Server requires explicit creation after altering column type with the same constraint
            old_db_params['check'] == new_db_params['check'] and new_db_params['check'] and
            old_db_params['type'] != new_db_params['type']
        ):
            self.execute(
                self.sql_create_check % {
                    "table": self.quote_name(model._meta.db_table),
                    "name": self.quote_name(
                        self._create_index_name(model._meta.db_table, [new_field.column], suffix="_check")
                    ),
                    "column": self.quote_name(new_field.column),
                    "check": new_db_params['check'],
                }
            )
        # Drop the default if we need to
        # (Django usually does not use in-database defaults)
        if needs_database_default:
            changes_sql, params = self._alter_column_default_sql(model, old_field, new_field, drop=True)
            sql = self.sql_alter_column % {
                "table": self.quote_name(model._meta.db_table),
                "changes": changes_sql,
            }
            self.execute(sql, params)

        # Reset connection if required
        if self.connection.features.connection_persists_old_columns:
            self.connection.close()

    def _delete_indexes(self, model, old_field, new_field):
        if (
            django_version >= (4, 2)
            and isinstance(new_field, ForeignKey)
            and type(new_field.db_comment) != type(None)
            and "fk_on_delete_keep_index" in new_field.db_comment
        ):
            return
        index_columns = []
        index_names = []
        if old_field.db_index and new_field.db_index:
            index_columns.append([old_field.column])
        elif old_field.null != new_field.null:
            index_columns.append([old_field.column])
        # Handle index_together for only django version < 5.1    
        if django_version < (5, 1):  
           # Iterate over each set of field names defined in index_together  
           for fields in model._meta.index_together:
              # Get the actual column names for each field in the set
              columns = [model._meta.get_field(field).column for field in fields]
              # If the old field's column is among these columns, add to index_columns for later index deletion
              if old_field.column in columns:
                 index_columns.append(columns)

        for index in model._meta.indexes:
            columns = [model._meta.get_field(field).column for field in index.fields]
            if old_field.column in columns:
                index_columns.append(columns)

        for fields in model._meta.unique_together:
            columns = [model._meta.get_field(field).column for field in fields]
            if old_field.column in columns:
                index_columns.append(columns)
        if index_columns:
            # remove duplicates first
            temp = []
            for columns in index_columns:
                if columns not in temp:
                    temp.append(columns)
            index_columns = temp

            for columns in index_columns:
                index_names = self._constraint_names(model, columns, index=True)
                for index_name in index_names:
                    self.execute(self._delete_constraint_sql(self.sql_delete_index, model, index_name))
        return index_names

    def _delete_unique_constraints(self, model, old_field, new_field, strict=False):
        unique_columns = []
        # Considering just this column, we only need to drop unique constraints in advance of altering the field
        # *if* it remains unique - if it wasn't unique before there's nothing to drop; if it won't remain unique
        # afterwards then that is handled separately in _alter_field
        if old_field.unique and new_field.unique:
            unique_columns.append([old_field.column])

        # Also consider unique_together because, although this is implemented with a filtered unique INDEX now, we
        # need to handle the possibility that we're acting on a database previously created by an older version of
        # this backend, where unique_together used to be implemented with a CONSTRAINT
        for fields in model._meta.unique_together:
            columns = [model._meta.get_field(field).column for field in fields]
            if old_field.column in columns:
                unique_columns.append(columns)

        if unique_columns:
            for columns in unique_columns:
                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 django_version >= (4, 1):
            if constraint_names and self.connection.features.allows_multiple_constraints_on_same_fields:
                # Constraint matching the unique_together name.
                default_name = str(
                    self._unique_constraint_name(model._meta.db_table, columns, quote=False)
                )
                if default_name in constraint_names:
                    constraint_names = [default_name]
        if strict and len(constraint_names) != 1:
            raise ValueError("Found wrong number (%s) of unique constraints for columns %s" % (
                len(constraint_names),
                repr(columns),
            ))
        # Delete constraints which are implemented as a table CONSTRAINT (this may include some created by an
        # older version of this backend, even if the current version would implement it with an INDEX instead)
        for constraint_name in constraint_names_normal:
            self.execute(self._delete_constraint_sql(self.sql_delete_unique, model, constraint_name))
        # Delete constraints which are implemented with an explicit index instead (not a table CONSTRAINT)
        # These are used 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)
        return super()._rename_field_sql(table, old_field, new_field, new_type)

    def _set_field_new_type_null_status(self, field, new_type):
        """
        Keep the null property of the old field. If it has changed, it will be
        handled separately.
        """
        if field.null:
            new_type += " NULL"
        else:
            new_type += " NOT NULL"
        return new_type

    def add_field(self, model, field):
        """
        Create a field on a model. Usually involves adding a column, but may
        involve adding a table instead (for M2M fields).
        """
        # Special-case implicit M2M tables
        if field.many_to_many and field.remote_field.through._meta.auto_created:
            return self.create_model(field.remote_field.through)
        # Get the column's definition
        definition, params = self.column_sql(model, field, include_default=True)
        # It might not actually have a column behind it
        if definition is None:
            return
        if col_type_suffix := field.db_type_suffix(connection=self.connection):
            definition += f" {col_type_suffix}"
        # Remove column type from definition if field is generated
        if (django_version >= (5,0) and field.generated):
            definition = definition[definition.find('AS'):]
        # Nullable columns with default values require 'WITH VALUES' to set existing rows
        if 'DEFAULT' in definition and field.null:
            definition = definition.replace('NULL', 'WITH VALUES')

        if (self.connection.features.supports_nullable_unique_constraints and
                not field.many_to_many and field.null and field.unique):

            definition = definition.replace(' UNIQUE', '')
            statement = self._create_index_sql(
                model, [field], sql=self.sql_create_unique_null, suffix="_uniq"
            )
            self.deferred_sql.append(statement)
            self._add_deferred_unique_index_for_field(field, statement)

        # Check constraints can go on the column SQL here
        db_params = field.db_parameters(connection=self.connection)
        if db_params['check']:
            definition += " CHECK (%s)" % db_params['check']
        # Build the SQL and run it
        sql = self.sql_create_column % {
            "table": self.quote_name(model._meta.db_table),
            "column": self.quote_name(field.column),
            "definition": definition,
        }
        self.execute(sql, params)
        # Drop the default if we need to
        # (Django usually does not use in-database defaults)
        if (
            ((django_version >= (5,0) and field.db_default is NOT_PROVIDED) or django_version < (5,0))
            and not self.skip_default(field)
            and self.effective_default(field) is not None
        ):
            changes_sql, params = self._alter_column_default_sql(model, None, field, drop=True)
            sql = self.sql_alter_column % {
                "table": self.quote_name(model._meta.db_table),
                "changes": changes_sql,
            }
            self.execute(sql, params)
        # Add field comment, if required.
        if django_version >= (4, 2):
            if (
                field.db_comment
                and self.connection.features.supports_comments
                and not self.connection.features.supports_comments_inline
            ):
                field_type = db_params["type"]
                self.execute(
                    *self._alter_column_comment_sql(
                        model, field, field_type, field.db_comment
                    )
                )
        # Add an index, if required
        self.deferred_sql.extend(self._field_indexes_sql(model, field))
        # Add any FK constraints later
        if field.remote_field and self.connection.features.supports_foreign_keys and field.db_constraint:
            self.deferred_sql.append(self._create_fk_sql(model, field, "_fk_%(to_table)s_%(to_column)s"))
        # Reset connection if required
        if self.connection.features.connection_persists_old_columns:
            self.connection.close()

    if django_version >= (4, 0):
        def _create_unique_sql(
                self,
                model,
                fields,
                name=None,
                condition=None,
                deferrable=None,
                include=None,
                opclasses=None,
                expressions=None,
                nulls_distinct=None
            ):
            if not self._unique_supported(
                condition=condition,
                deferrable=deferrable,
                include=include,
                expressions=expressions,
                nulls_distinct=nulls_distinct,
            ):
                return None

            def create_unique_name(*args, **kwargs):
                return self.quote_name(self._create_index_name(*args, **kwargs))

            compiler = Query(model, alias_cols=False).get_compiler(connection=self.connection)
            columns = [field.column for field in fields]
            table = model._meta.db_table

            if name is None:
                name = IndexName(table, columns, '_uniq', create_unique_name)
            else:
                name = self.quote_name(name)

            if columns:
                columns = self._index_columns(table, columns, col_suffixes=(), opclasses=opclasses)
            else:
                columns = Expressions(table, expressions, compiler, self.quote_value)
            statement_args = {
                "deferrable": self._deferrable_constraint_sql(deferrable)
            }
            include = self._index_include_sql(model, include)

            if condition:
                return Statement(
                    self.sql_create_unique_index,
                    table=self.quote_name(table),
                    name=name,
                    columns=columns,
                    condition=' WHERE ' + condition,
                    **statement_args,
                    include=include,
                    nulls_distinct=''
                ) if self.connection.features.supports_partial_indexes else None
            else:
                return Statement(
                    self.sql_create_unique,
                    table=self.quote_name(table),
                    name=name,
                    columns=columns,
                    **statement_args,
                    include=include,
                    nulls_distinct=''
                )
    else:
        def _create_unique_sql(self, model, columns,
                               name=None, condition=None, deferrable=None,
                               include=None, opclasses=None, expressions=None):
            if (deferrable and not getattr(self.connection.features, 'supports_deferrable_unique_constraints', False) or
                (condition and not self.connection.features.supports_partial_indexes) or
                (include and not self.connection.features.supports_covering_indexes) or
                    (expressions and not self.connection.features.supports_expression_indexes)):
                return None

            def create_unique_name(*args, **kwargs):
                return self.quote_name(self._create_index_name(*args, **kwargs))

            table = Table(model._meta.db_table, self.quote_name)
            if name is None:
                name = IndexName(model._meta.db_table, columns, '_uniq', create_unique_name)
            else:
                name = self.quote_name(name)
            columns = Columns(table, columns, self.quote_name)
            statement_args = {
                "deferrable": self._deferrable_constraint_sql(deferrable)
            } if django_version >= (3, 1) else {}
            include = self._index_include_sql(model, include) if django_version >= (3, 2) else ''

            if condition:
                return Statement(
                    self.sql_create_unique_index,
                    table=self.quote_name(table) if isinstance(table, str) else table,
                    name=name,
                    columns=columns,
                    condition=' WHERE ' + condition,
                    **statement_args,
                    include=include,
                ) if self.connection.features.supports_partial_indexes else None
            else:
                return Statement(
                    self.sql_create_unique,
                    table=self.quote_name(table) if isinstance(table, str) else table,
                    name=name,
                    columns=columns,
                    **statement_args,
                    include=include,
                )

    def _create_index_sql(self, model, fields, *, name=None, suffix='', using='',
                          db_tablespace=None, col_suffixes=(), sql=None, opclasses=(),
                          condition=None, include=None, expressions=None):
        """
        Return the SQL statement to create the index for one or several fields.
        `sql` can be specified if the syntax differs from the standard (GIS
        indexes, ...).
        """
        if django_version >= (3, 2):
            return super()._create_index_sql(
                model, fields=fields, name=name, suffix=suffix, using=using,
                db_tablespace=db_tablespace, col_suffixes=col_suffixes, sql=sql,
                opclasses=opclasses, condition=condition, include=include,
                expressions=expressions,
            )
        return super()._create_index_sql(
            model, fields=fields, name=name, suffix=suffix, using=using,
            db_tablespace=db_tablespace, col_suffixes=col_suffixes, sql=sql,
            opclasses=opclasses, condition=condition,
        )

    def create_model(self, model):
        """
        Takes a model and creates a table for it in the database.
        Will also create any accompanying indexes or unique constraints.
        """
        # Create column SQL, add FK deferreds if needed
        column_sqls = []
        params = []
        for field in model._meta.local_fields:
            # SQL
            definition, extra_params = self.column_sql(model, field)
            if definition is None:
                continue
            # Remove column type from definition if field is generated
            if (django_version >= (5,0) and field.generated):
                definition = definition[definition.find('AS'):]

            if (self.connection.features.supports_nullable_unique_constraints and
                    not field.many_to_many and field.null and field.unique):

                definition = definition.replace(' UNIQUE', '')
                statement = self._create_index_sql(
                    model, [field], sql=self.sql_create_unique_null, suffix="_uniq"
                )
                self.deferred_sql.append(statement)
                self._add_deferred_unique_index_for_field(field, statement)

            # Check constraints can go on the column SQL here
            db_params = field.db_parameters(connection=self.connection)
            if db_params['check']:
                # SQL Server requires a name for the check constraint
                definition += self._sql_check_constraint % {
                    "name": self._create_index_name(model._meta.db_table, [field.column], suffix="_check"),
                    "check": db_params['check']
                }
            # Autoincrement SQL (for backends with inline variant)
            col_type_suffix = field.db_type_suffix(connection=self.connection)
            if col_type_suffix:
                definition += " %s" % col_type_suffix
            params.extend(extra_params)
            # FK
            if field.remote_field and field.db_constraint:
                to_table = field.remote_field.model._meta.db_table
                to_column = field.remote_field.model._meta.get_field(field.remote_field.field_name).column
                if self.sql_create_inline_fk:
                    definition += " " + self.sql_create_inline_fk % {
                        "to_table": self.quote_name(to_table),
                        "to_column": self.quote_name(to_column),
                    }
                elif self.connection.features.supports_foreign_keys:
                    self.deferred_sql.append(self._create_fk_sql(model, field, "_fk_%(to_table)s_%(to_column)s"))
            # Add the SQL to our big list
            column_sqls.append("%s %s" % (
                self.quote_name(field.column),
                definition,
            ))
            # Autoincrement SQL (for backends with post table definition variant)
            if field.get_internal_type() in ("AutoField", "BigAutoField", "SmallAutoField"):
                autoinc_sql = self.connection.ops.autoinc_sql(model._meta.db_table, field.column)
                if autoinc_sql:
                    self.deferred_sql.extend(autoinc_sql)

        # Add any unique_togethers (always deferred, as some fields might be
        # created afterwards, like geometry fields with some backends)
        for field_names in model._meta.unique_together:
            fields = [model._meta.get_field(field) for field in field_names]
            columns = [model._meta.get_field(field).column for field in field_names]
            condition = ' AND '.join(["[%s] IS NOT NULL" % col for col in columns])
            if django_version >= (4, 0):
                self.deferred_sql.append(self._create_unique_sql(model, fields, condition=condition))
            else:
                self.deferred_sql.append(self._create_unique_sql(model, columns, condition=condition))

        constraints = [constraint.constraint_sql(model, self) for constraint in model._meta.constraints]
        # Make the table
        sql = self.sql_create_table % {
            "table": self.quote_name(model._meta.db_table),
            'definition': ', '.join(constraint for constraint in (*column_sqls, *constraints) if constraint),
        }
        if model._meta.db_tablespace:
            tablespace_sql = self.connection.ops.tablespace_sql(model._meta.db_tablespace)
            if tablespace_sql:
                sql += ' ' + tablespace_sql
        # Prevent using [] as params, in the case a literal '%' is used in the definition
        self.execute(sql, params or None)

        if django_version >= (4, 2) and self.connection.features.supports_comments:
            # Add table comment.
            if model._meta.db_table_comment:
                self.alter_db_table_comment(model, None, model._meta.db_table_comment)
            # Add column comments.
            if not self.connection.features.supports_comments_inline:
                for field in model._meta.local_fields:
                    if field.db_comment:
                        field_db_params = field.db_parameters(
                            connection=self.connection
                        )
                        field_type = field_db_params["type"]
                        self.execute(
                            *self._alter_column_comment_sql(
                                model, field, field_type, field.db_comment
                            )
                        )
        # Add any field index and index_together's (deferred as SQLite3 _remake_table needs it)
        self.deferred_sql.extend(self._model_indexes_sql(model))
        self.deferred_sql = list(set(self.deferred_sql))

        # Make M2M tables
        for field in model._meta.local_many_to_many:
            if field.remote_field.through._meta.auto_created:
                self.create_model(field.remote_field.through)

    def _delete_unique_sql(
        self,
        model,
        name,
        condition=None,
        deferrable=None,
        include=None,
        opclasses=None,
        expressions=None,
        nulls_distinct=None,
    ):
        if not self._unique_supported(
            condition=condition,
            deferrable=deferrable,
            include=include,
            expressions=expressions,
            nulls_distinct=nulls_distinct,
        ):
            return None
        if condition or include or opclasses:
            sql = self.sql_delete_index
            with self.connection.cursor() as cursor:
                cursor.execute(
                    "SELECT 1 FROM INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE WHERE CONSTRAINT_NAME = '%s'" % name)
                row = cursor.fetchone()
                if row:
                    sql = self.sql_delete_unique
        else:
            sql = self.sql_delete_unique
        return self._delete_constraint_sql(sql, model, name)

    def delete_model(self, model):
        super().delete_model(model)

    def execute(self, sql, params=(), has_result=False):
        """
        Executes the given SQL statement, with optional parameters.
        """
        result = None
        # Don't perform the transactional DDL check if SQL is being collected
        # as it's not going to be executed anyway.
        if not self.collect_sql and self.connection.in_atomic_block and not self.connection.features.can_rollback_ddl:
            raise TransactionManagementError(
                "Executing DDL statements while in a transaction on databases "
                "that can't perform a rollback is prohibited."
            )
        # Account for non-string statement objects.
        sql = str(sql)
        # Log the command we're running, then run it
        logger.debug("%s; (params %r)", sql, params, extra={'params': params, 'sql': sql})
        if self.collect_sql:
            ending = "" if sql.endswith(";") else ";"
            if params is not None:
                self.collected_sql.append((sql % tuple(map(self.quote_value, params))) + ending)
            else:
                self.collected_sql.append(sql + ending)
        else:
            cursor = self.connection.cursor()
            cursor.execute(sql, params)
            if has_result:
                result = cursor.fetchall()
            # the cursor can be closed only when the driver supports opening
            # multiple cursors on a connection because the migration command
            # has already opened a cursor outside this method
            if self.connection.supports_mars:
                cursor.close()
        return result

    def prepare_default(self, value):
        return self.quote_value(value)

    def quote_value(self, value):
        """
        Returns a quoted version of the value so it's safe to use in an SQL
        string. This is not safe against injection from user code; it is
        intended only for use in making SQL scripts or preparing default values
        for particularly tricky backends (defaults are not user-defined, though,
        so this is safe).
        """
        if isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
            return "'%s'" % value
        elif isinstance(value, str):
            return "'%s'" % value.replace("'", "''")
        elif isinstance(value, (bytes, bytearray, memoryview)):
            return "0x%s" % force_str(binascii.hexlify(value))
        elif isinstance(value, bool):
            return "1" if value else "0"
        else:
            return str(value)

    def remove_field(self, model, field):
        """
        Removes a field from a model. Usually involves deleting a column,
        but for M2Ms may involve deleting a table.
        """
        # Special-case implicit M2M tables
        if field.many_to_many and field.remote_field.through._meta.auto_created:
            return self.delete_model(field.remote_field.through)
        # It might not actually have a column behind it
        if field.db_parameters(connection=self.connection)['type'] is None:
            return
        # Drop any FK constraints, SQL Server requires explicit deletion
        with self.connection.cursor() as cursor:
            constraints = self.connection.introspection.get_constraints(cursor, model._meta.db_table)
        for name, infodict in constraints.items():
            if field.column in infodict['columns'] and infodict['foreign_key']:
                self.execute(self._delete_constraint_sql(self.sql_delete_fk, model, name))
        # Drop any indexes, SQL Server requires explicit deletion
        for name, infodict in constraints.items():
            if field.column in infodict['columns'] and infodict['index']:
                self.execute(self.sql_delete_index % {
                    "table": self.quote_name(model._meta.db_table),
                    "name": self.quote_name(name),
                })
        # Drop primary key constraint, SQL Server requires explicit deletion
        for name, infodict in constraints.items():
            if field.column in infodict['columns'] and infodict['primary_key']:
                self.execute(self.sql_delete_pk % {
                    "table": self.quote_name(model._meta.db_table),
                    "name": self.quote_name(name),
                })
        # Drop check constraints, SQL Server requires explicit deletion
        for name, infodict in constraints.items():
            if field.column in infodict['columns'] and infodict['check']:
                self.execute(self.sql_delete_check % {
                    "table": self.quote_name(model._meta.db_table),
                    "name": self.quote_name(name),
                })
        # Drop unique constraints, SQL Server requires explicit deletion
        for name, infodict in constraints.items():
            if (field.column in infodict['columns'] and infodict['unique'] and
                    not infodict['primary_key'] and not infodict['index']):
                self.execute(self.sql_delete_unique % {
                    "table": self.quote_name(model._meta.db_table),
                    "name": self.quote_name(name),
                })
        # Drop default constraint, SQL Server requires explicit deletion
        for name, infodict in constraints.items():
            if field.column in infodict['columns'] and infodict['default']:
                self.execute(self.sql_delete_default % {
                    "table": self.quote_name(model._meta.db_table),
                    "name": self.quote_name(name),
                })
        # Delete the column
        sql = self.sql_delete_column % {
            "table": self.quote_name(model._meta.db_table),
            "column": self.quote_name(field.column),
        }
        self.execute(sql)
        # Reset connection if required
        if self.connection.features.connection_persists_old_columns:
            self.connection.close()
        # Remove all deferred statements referencing the deleted column.
        for sql in list(self.deferred_sql):
            if isinstance(sql, Statement) and sql.references_column(model._meta.db_table, field.column):
                self.deferred_sql.remove(sql)

    def add_constraint(self, model, constraint):
        if isinstance(constraint, UniqueConstraint) and constraint.condition and constraint.condition.connector != AND:
            raise NotImplementedError("The backend does not support %s conditions on unique constraint %s." %
                                      (constraint.condition.connector, constraint.name))
        super().add_constraint(model, constraint)

    if django_version >= (4, 2):
        def _collate_sql(self, collation, old_collation=None, table_name=None):
            return ' COLLATE ' + collation if collation else ""
    else:
        def _collate_sql(self, collation):
            return ' COLLATE ' + collation

    def _create_index_name(self, table_name, column_names, suffix=""):
        index_name = super()._create_index_name(table_name, column_names, suffix)
        # Check if the db_table specified a user-defined schema
        if('].[' in index_name):
            new_index_name = index_name.replace('[', '').replace(']', '').replace('.', '_')
            return new_index_name
        return index_name

    def _unique_supported(
        self,
        condition=None,
        deferrable=None,
        include=None,
        expressions=None,
        nulls_distinct=None,
    ):
        return (
            (not condition or self.connection.features.supports_partial_indexes)
            and (
                not deferrable
                or self.connection.features.supports_deferrable_unique_constraints
            )
            and (not include or self.connection.features.supports_covering_indexes)
            and (
                not expressions or self.connection.features.supports_expression_indexes
            )
            and (
                nulls_distinct is None
                or self.connection.features.supports_nulls_distinct_unique_constraints
            )
        )