apply render_schema_translates to identity insert directives

Fixed bug where the "schema_translate_map" feature would fail to function
correctly in conjunction with an INSERT into a table that has an IDENTITY
column, where the value of the IDENTITY column were specified in the values
of the INSERT thus triggering SQLAlchemy's feature of setting IDENTITY
INSERT to "on"; it's in this directive where the schema translate map would
fail to be honored.

Fixes: #6658
Change-Id: I8235aa639dd465d038a2ad48e7a669f3e5c5c37c

3 files changed, 42 insertions, 2 deletions

diff --git a/doc/build/changelog/unreleased_14/6658.rst b/doc/build/changelog/unreleased_14/6658.rst
new file mode 100644
index 000000000..c0d899517
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/6658.rst
@@ -0,0 +1,11 @@
+.. change::
+    :tags: bug, mssql
+    :tickets: 6658
+
+    Fixed bug where the "schema_translate_map" feature would fail to function
+    correctly in conjunction with an INSERT into a table that has an IDENTITY
+    column, where the value of the IDENTITY column were specified in the values
+    of the INSERT thus triggering SQLAlchemy's feature of setting IDENTITY
+    INSERT to "on"; it's in this directive where the schema translate map would
+    fail to be honored.
+
diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index 750f3743b..4ca83a697 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -1533,10 +1533,18 @@ class MSExecutionContext(default.DefaultExecutionContext):
     _result_strategy = None
 
     def _opt_encode(self, statement):
+
         if not self.dialect.supports_unicode_statements:
-            return self.dialect._encoder(statement)[0]
+            encoded = self.dialect._encoder(statement)[0]
         else:
-            return statement
+            encoded = statement
+
+        if self.compiled and self.compiled.schema_translate_map:
+
+            rst = self.compiled.preparer._render_schema_translates
+            encoded = rst(encoded, self.compiled.schema_translate_map)
+
+        return encoded
 
     def pre_exec(self):
         """Activate IDENTITY_INSERT if needed."""
diff --git a/test/dialect/mssql/test_query.py b/test/dialect/mssql/test_query.py
index d22016e8f..6952b26a9 100644
--- a/test/dialect/mssql/test_query.py
+++ b/test/dialect/mssql/test_query.py
@@ -18,6 +18,7 @@ from sqlalchemy import testing
 from sqlalchemy import util
 from sqlalchemy.dialects.mssql import base as mssql
 from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import config
 from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
@@ -118,6 +119,26 @@ class IdentityInsertTest(fixtures.TablesTest, AssertsCompiledSQL):
         conn.execute(cattable.insert().values({cattable.c.id: literal(5)}))
         eq_(conn.scalar(select(cattable.c.id)), 5)
 
+    @testing.requires.schemas
+    def test_insert_using_schema_translate(self, connection, metadata):
+
+        t = Table(
+            "t",
+            metadata,
+            Column("id", Integer),
+            Column("description", String(50)),
+            PrimaryKeyConstraint("id", name="PK_cattable"),
+            schema=None,
+        )
+        conn = connection.execution_options(
+            schema_translate_map={None: config.test_schema}
+        )
+        metadata.create_all(conn)
+
+        conn.execute(t.insert().values({"id": 1, "description": "descrip"}))
+
+        eq_(conn.execute(select(t)).first(), (1, "descrip"))
+
 
 class QueryUnicodeTest(fixtures.TestBase):