dont mutate bind_arguments incoming dictionary

The :paramref:`_orm.Session.execute.bind_arguments` dictionary is no longer
mutated when passed to :meth:`_orm.Session.execute` and similar; instead,
it's copied to an internal dictionary for state changes. Among other
things, this fixes and issue where the "clause" passed to the
:meth:`_orm.Session.get_bind` method would be incorrectly referring to the
:class:`_sql.Select` construct used for the "fetch" synchronization
strategy, when the actual query being emitted was a :class:`_dml.Delete` or
:class:`_dml.Update`. This would interfere with recipes for "routing
sessions".

Fixes: #8614
Change-Id: I8d237449485c9bbf41db2b29a34b6136aa43b7bc

5 files changed, 84 insertions, 1 deletions

diff --git a/doc/build/changelog/unreleased_14/8614.rst b/doc/build/changelog/unreleased_14/8614.rst
new file mode 100644
index 000000000..b975dbc1f
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/8614.rst
@@ -0,0 +1,13 @@
+.. change::
+    :tags: bug, orm
+    :tickets: 8614
+
+    The :paramref:`_orm.Session.execute.bind_arguments` dictionary is no longer
+    mutated when passed to :meth:`_orm.Session.execute` and similar; instead,
+    it's copied to an internal dictionary for state changes. Among other
+    things, this fixes and issue where the "clause" passed to the
+    :meth:`_orm.Session.get_bind` method would be incorrectly referring to the
+    :class:`_sql.Select` construct used for the "fetch" synchronization
+    strategy, when the actual query being emitted was a :class:`_dml.Delete` or
+    :class:`_dml.Update`. This would interfere with recipes for "routing
+    sessions".
diff --git a/lib/sqlalchemy/orm/bulk_persistence.py b/lib/sqlalchemy/orm/bulk_persistence.py
index b407fcdca..af5bf6b6a 100644
--- a/lib/sqlalchemy/orm/bulk_persistence.py
+++ b/lib/sqlalchemy/orm/bulk_persistence.py
@@ -599,7 +599,6 @@ class BulkUDCompileState(ORMDMLState):
             execution_options,
             statement._execution_options,
         )
-
         bind_arguments["clause"] = statement
         try:
             plugin_subject = statement._propagate_attrs["plugin_subject"]
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index 9577b4d26..324ab7b25 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -1823,6 +1823,8 @@ class Session(_SessionClassMethods, EventTarget):
 
         if not bind_arguments:
             bind_arguments = {}
+        else:
+            bind_arguments = dict(bind_arguments)
 
         if (
             statement._propagate_attrs.get("compile_state_plugin", None)
diff --git a/test/orm/dml/test_update_delete_where.py b/test/orm/dml/test_update_delete_where.py
index 3250cb3f9..c8e56e3c1 100644
--- a/test/orm/dml/test_update_delete_where.py
+++ b/test/orm/dml/test_update_delete_where.py
@@ -26,6 +26,9 @@ from sqlalchemy.orm import Session
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.orm import synonym
 from sqlalchemy.orm import with_loader_criteria
+from sqlalchemy.sql.dml import Delete
+from sqlalchemy.sql.dml import Update
+from sqlalchemy.sql.selectable import Select
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import eq_
@@ -1886,6 +1889,50 @@ class UpdateDeleteTest(fixtures.MappedTest):
         ):
             session.execute(stmt)
 
+    @testing.combinations(("update",), ("delete",), argnames="stmt_type")
+    @testing.combinations(
+        ("evaluate",), ("fetch",), (None,), argnames="sync_type"
+    )
+    def test_routing_session(self, stmt_type, sync_type, connection):
+        User = self.classes.User
+
+        if stmt_type == "update":
+            stmt = update(User).values(age=123)
+            expected = [Update]
+        elif stmt_type == "delete":
+            stmt = delete(User)
+            expected = [Delete]
+        else:
+            assert False
+
+        received = []
+
+        class RoutingSession(Session):
+            def get_bind(self, **kw):
+                received.append(type(kw["clause"]))
+                return super(RoutingSession, self).get_bind(**kw)
+
+        stmt = stmt.execution_options(synchronize_session=sync_type)
+
+        if sync_type == "fetch":
+            expected.insert(0, Select)
+
+            if (
+                stmt_type == "update"
+                and not connection.dialect.update_returning
+            ):
+                expected.insert(0, Select)
+            elif (
+                stmt_type == "delete"
+                and not connection.dialect.delete_returning
+            ):
+                expected.insert(0, Select)
+
+        with RoutingSession(bind=connection) as sess:
+            sess.execute(stmt)
+
+        eq_(received, expected)
+
 
 class UpdateDeleteIgnoresLoadersTest(fixtures.MappedTest):
     @classmethod
diff --git a/test/orm/test_bind.py b/test/orm/test_bind.py
index 2f392cf6e..409c6244f 100644
--- a/test/orm/test_bind.py
+++ b/test/orm/test_bind.py
@@ -291,6 +291,28 @@ class BindIntegrationTest(_fixtures.FixtureTest):
 
         sess.close()
 
+    @testing.combinations(True, False)
+    def test_dont_mutate_binds(self, empty_dict):
+        users, User = (
+            self.tables.users,
+            self.classes.User,
+        )
+
+        mp = self.mapper_registry.map_imperatively(User, users)
+
+        sess = fixture_session()
+
+        if empty_dict:
+            bind_arguments = {}
+        else:
+            bind_arguments = {"mapper": mp}
+        sess.execute(select(1), bind_arguments=bind_arguments)
+
+        if empty_dict:
+            eq_(bind_arguments, {})
+        else:
+            eq_(bind_arguments, {"mapper": mp})
+
     @testing.combinations(
         (
             lambda session, Address: session.query(Address).statement,