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
(cherry picked from commit 3efc9e1df378be8046d4b1f1b624968a62eb100f)

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".

lib/sqlalchemy/orm/session.py

@@ -1639,6 +1639,8 @@ class Session(_SessionClassMethods):
                 bind_arguments.update(kw)
         elif not bind_arguments:
             bind_arguments = {}
+        else:
+            bind_arguments = dict(bind_arguments)
 
         if (
             statement._propagate_attrs.get("compile_state_plugin", None)

test/orm/test_bind.py

@@ -290,6 +290,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,

test/orm/test_update_delete.py

@@ -22,6 +22,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_
@@ -1460,6 +1463,42 @@ class UpdateDeleteTest(fixtures.MappedTest):
         ]
         eq_(["name", "age_int"], cols)
 
+    @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 not connection.dialect.full_returning:
+                expected.insert(0, Select)
+
+        with RoutingSession(bind=connection) as sess:
+            sess.execute(stmt)
+
+        eq_(received, expected)
+
 
 class UpdateDeleteIgnoresLoadersTest(fixtures.MappedTest):
     @classmethod