Merge "apply render_schema_translates to identity insert directives"

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):