Repository: sqlalchemy/sqlalchemy

Issue:
The runtime issue related to the join operation in the association example
### Describe the bug

sqlalchemy.exc.ArgumentError
```
(smart-meeting-nlp) (base) ➜  sqlalchemy git:(main) ✗ /home/tangzhifeng/miniconda3/envs/smart-meeting
-nlp/bin/python /home/tangzhifeng/sqlalchemy/examples/association/basic_association.py
/home/tangzhifeng/sqlalchemy/examples/association/basic_association.py:28: MovedIn20Warning: The ``declarative_base()`` function is now available as sqlalchemy.orm.declarative_base(). (deprecated since: 2.0) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  Base = declarative_base()
[('SA Mug', 6.5), ('SA Hat', 8.99), ('MySQL Crowbar', 10.99)]
Traceback (most recent call last):
  File "/home/tangzhifeng/sqlalchemy/examples/association/basic_association.py", line 108, in <module>
    q = session.query(Order).join("order_items", "item")
  File "<string>", line 2, in join
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/sql/base.py", line 280, in _generative
    x = fn(self, *args, **kw)
  File "<string>", line 2, in join
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/orm/base.py", line 304, in generate
    fn(self, *args, **kw)
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/orm/query.py", line 2422, in join
    join_target = coercions.expect(
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/sql/coercions.py", line 396, in expect
    resolved = impl._literal_coercion(
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/sql/coercions.py", line 1219, in _literal_coercion
    self._raise_for_expected(element, argname)
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/sql/coercions.py", line 518, in _raise_for_expected
    raise exc.ArgumentError(msg, code=code) from err
sqlalchemy.exc.ArgumentError: Join target, typically a FROM expression, or ORM relationship attribute expected, got 'order_items'.
```


### Optional link from https://docs.sqlalchemy.org which documents the behavior that is expected

_No response_

### SQLAlchemy Version in Use

2.0.2

### DBAPI (i.e. the database driver)

pysqlite

### Database Vendor and Major Version

SQLite

### Python Version

3.9

### Operating system

Linux

### To Reproduce

```python
"""Illustrate a many-to-many relationship between an
"Order" and a collection of "Item" objects, associating a purchase price
with each via an association object called "OrderItem"

The association object pattern is a form of many-to-many which
associates additional data with each association between parent/child.

The example illustrates an "order", referencing a collection
of "items", with a particular price paid associated with each "item".

"""

from datetime import datetime

from sqlalchemy import and_
from sqlalchemy import Column
from sqlalchemy import create_engine
from sqlalchemy import DateTime
from sqlalchemy import Float
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session


Base = declarative_base()


class Order(Base):
    __tablename__ = "order"

    order_id = Column(Integer, primary_key=True)
    customer_name = Column(String(30), nullable=False)
    order_date = Column(DateTime, nullable=False, default=datetime.now())
    order_items = relationship(
        "OrderItem", cascade="all, delete-orphan", backref="order"
    )

    def __init__(self, customer_name):
        self.customer_name = customer_name


class Item(Base):
    __tablename__ = "item"
    item_id = Column(Integer, primary_key=True)
    description = Column(String(30), nullable=False)
    price = Column(Float, nullable=False)

    def __init__(self, description, price):
        self.description = description
        self.price = price

    def __repr__(self):
        return "Item(%r, %r)" % (self.description, self.price)


class OrderItem(Base):
    __tablename__ = "orderitem"
    order_id = Column(Integer, ForeignKey("order.order_id"), primary_key=True)
    item_id = Column(Integer, ForeignKey("item.item_id"), primary_key=True)
    price = Column(Float, nullable=False)

    def __init__(self, item, price=None):
        self.item = item
        self.price = price or item.price

    item = relationship(Item, lazy="joined")


if __name__ == "__main__":
    engine = create_engine("sqlite://")
    Base.metadata.create_all(engine)

    session = Session(engine)

    # create catalog
    tshirt, mug, hat, crowbar = (
        Item("SA T-Shirt", 10.99),
        Item("SA Mug", 6.50),
        Item("SA Hat", 8.99),
        Item("MySQL Crowbar", 16.99),
    )
    session.add_all([tshirt, mug, hat, crowbar])
    session.commit()

    # create an order
    order = Order("john smith")

    # add three OrderItem associations to the Order and save
    order.order_items.append(OrderItem(mug))
    order.order_items.append(OrderItem(crowbar, 10.99))
    order.order_items.append(OrderItem(hat))
    session.add(order)
    session.commit()

    # query the order, print items
    order = session.query(Order).filter_by(customer_name="john smith").one()
    print(
        [
            (order_item.item.description, order_item.price)
            for order_item in order.order_items
        ]
    )

    # print customers who bought 'MySQL Crowbar' on sale
    q = session.query(Order).join("order_items", "item")
    q = q.filter(
        and_(Item.description == "MySQL Crowbar", Item.price > OrderItem.price)
    )

    print([order.customer_name for order in q])
```


### Error

```
/home/tangzhifeng/sqlalchemy/examples/association/basic_association.py:28: MovedIn20Warning: The ``declarative_base()`` function is now available as sqlalchemy.orm.declarative_base(). (deprecated since: 2.0) (Background on SQLAlchemy 2.0 at: https://sqlalche.me/e/b8d9)
  Base = declarative_base()
[('SA Mug', 6.5), ('SA Hat', 8.99), ('MySQL Crowbar', 10.99)]
Traceback (most recent call last):
  File "/home/tangzhifeng/sqlalchemy/examples/association/basic_association.py", line 108, in <module>
    q = session.query(Order).join("order_items", "item")
  File "<string>", line 2, in join
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/sql/base.py", line 280, in _generative
    x = fn(self, *args, **kw)
  File "<string>", line 2, in join
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/orm/base.py", line 304, in generate
    fn(self, *args, **kw)
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/orm/query.py", line 2422, in join
    join_target = coercions.expect(
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/sql/coercions.py", line 396, in expect
    resolved = impl._literal_coercion(
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/sql/coercions.py", line 1219, in _literal_coercion
    self._raise_for_expected(element, argname)
  File "/home/tangzhifeng/miniconda3/envs/smart-meeting-nlp/lib/python3.9/site-packages/sqlalchemy/sql/coercions.py", line 518, in _raise_for_expected
    raise exc.ArgumentError(msg, code=code) from err
sqlalchemy.exc.ArgumentError: Join target, typically a FROM expression, or ORM relationship attribute expected, got 'order_items'.
``` 

### Additional context

_No response_

Hints from issue comments:
@CaselIT: Hi,

Thanks for reporting and the PR!

Solution patch:
diff --git a/.github/ISSUE_TEMPLATE/bug_report.yaml b/.github/ISSUE_TEMPLATE/bug_report.yaml
index 8c68d52fdc1..d72ed558b93 100644
--- a/.github/ISSUE_TEMPLATE/bug_report.yaml
+++ b/.github/ISSUE_TEMPLATE/bug_report.yaml
@@ -9,13 +9,25 @@ body:
     attributes:
       value: "
 
-**If you are writing new SQLAlchemy code and are observing a behavior that you did not expect,
-or if you are new to SQLAlchemy overall, please open a
-[discussion](https://github.com/sqlalchemy/sqlalchemy/discussions/new?category=Usage-Questions)
-instead of an issue report.  The VAST MAJORITY of issues are converted to discussions
-as they are not bugs.**
+**STOP**
+
+**We would really prefer if you DONT open a bug report.**
+
+**Please open a** [discussion](https://github.com/sqlalchemy/sqlalchemy/discussions/new?category=Usage-Questions) **instead of a bug report**.
+
+**Why?**
+
+**First, because the vast majority of issues reported are not bugs but either expected behaviors that
+are misunderstood by the user, or sometimes undefined behaviors that aren't supported.  These bugs are CLOSED**.
+
+**Secondly, because when there IS a bug, often it's not clear what the bug is or where it is, or
+if the thing is even expected, and we would much rather make a clean bug report once we've discussed
+the issue**.
+
+**Given the above, if you DO open a bug report anyway, we're probably going to assume you didn't read these instructions.**
 
-[START A NEW USAGE QUESTIONS DISCUSSION HERE](https://github.com/sqlalchemy/sqlalchemy/discussions/new?category=Usage-Questions)
+So since you are by definition reading this,
+[START A NEW USAGE QUESTIONS DISCUSSION HERE!](https://github.com/sqlalchemy/sqlalchemy/discussions/new?category=Usage-Questions)
 
 "
   - type: markdown
diff --git a/doc/build/changelog/changelog_20.rst b/doc/build/changelog/changelog_20.rst
index 0fa618f4a21..cca32ca1faf 100644
--- a/doc/build/changelog/changelog_20.rst
+++ b/doc/build/changelog/changelog_20.rst
@@ -9,9 +9,162 @@
 
 
 .. changelog::
-    :version: 2.0.33
+    :version: 2.0.35
     :include_notes_from: unreleased_20
 
+.. changelog::
+    :version: 2.0.34
+    :released: September 4, 2024
+
+    .. change::
+        :tags: bug, orm
+        :tickets: 11831
+
+        Fixed regression caused by issue :ticket:`11814` which broke support for
+        certain flavors of :pep:`593` ``Annotated`` in the type_annotation_map when
+        builtin types such as ``list``, ``dict`` were used without an element type.
+        While this is an incomplete style of typing, these types nonetheless
+        previously would be located in the type_annotation_map correctly.
+
+    .. change::
+        :tags: bug, sqlite
+        :tickets: 11832
+
+        Fixed regression in SQLite reflection caused by :ticket:`11677` which
+        interfered with reflection for CHECK constraints that were followed
+        by other kinds of constraints within the same table definition.   Pull
+        request courtesy Harutaka Kawamura.
+
+
+.. changelog::
+    :version: 2.0.33
+    :released: September 3, 2024
+
+    .. change::
+        :tags: bug, sqlite
+        :tickets: 11677
+
+        Improvements to the regex used by the SQLite dialect to reflect the name
+        and contents of a CHECK constraint.  Constraints with newline, tab, or
+        space characters in either or both the constraint text and constraint name
+        are now properly reflected.   Pull request courtesy Jeff Horemans.
+
+
+
+    .. change::
+        :tags: bug, engine
+        :tickets: 11687
+
+        Fixed issue in internal reflection cache where particular reflection
+        scenarios regarding same-named quoted_name() constructs would not be
+        correctly cached.  Pull request courtesy Felix Lüdin.
+
+    .. change::
+        :tags: bug, sql, regression
+        :tickets: 11703
+
+        Fixed regression in :meth:`_sql.Select.with_statement_hint` and others
+        where the generative behavior of the method stopped producing a copy of the
+        object.
+
+    .. change::
+        :tags: bug, mysql
+        :tickets: 11731
+
+        Fixed issue in MySQL dialect where using INSERT..FROM SELECT in combination
+        with ON DUPLICATE KEY UPDATE would erroneously render on MySQL 8 and above
+        the "AS new" clause, leading to syntax failures.  This clause is required
+        on MySQL 8 to follow the VALUES clause if use of the "new" alias is
+        present, however is not permitted to follow a FROM SELECT clause.
+
+
+    .. change::
+        :tags: bug, sqlite
+        :tickets: 11746
+
+        Improvements to the regex used by the SQLite dialect to reflect the name
+        and contents of a UNIQUE constraint that is defined inline within a column
+        definition inside of a SQLite CREATE TABLE statement, accommodating for tab
+        characters present within the column / constraint line. Pull request
+        courtesy John A Stevenson.
+
+
+
+
+    .. change::
+        :tags: bug, typing
+        :tickets: 11782
+
+        Fixed typing issue with :meth:`_sql.Select.with_only_columns`.
+
+    .. change::
+        :tags: bug, orm
+        :tickets: 11788
+
+        Correctly cleanup the internal top-level module registry when no
+        inner modules or classes are registered into it.
+
+    .. change::
+        :tags: bug, schema
+        :tickets: 11802
+
+        Fixed bug where the ``metadata`` element of an ``Enum`` datatype would not
+        be transferred to the new :class:`.MetaData` object when the type had been
+        copied via a :meth:`.Table.to_metadata` operation, leading to inconsistent
+        behaviors within create/drop sequences.
+
+    .. change::
+        :tags: bug, orm
+        :tickets: 11814
+
+        Improvements to the ORM annotated declarative type map lookup dealing with
+        composed types such as ``dict[str, Any]`` linking to JSON (or others) with
+        or without "future annotations" mode.
+
+
+
+    .. change::
+        :tags: change, general
+        :tickets: 11818
+
+        The pin for ``setuptools<69.3`` in ``pyproject.toml`` has been removed.
+        This pin was to prevent a sudden change in setuptools to use :pep:`625`
+        from taking place, which would change the file name of SQLAlchemy's source
+        distribution on pypi to be an all lower case name, which is likely to cause
+        problems with various build environments that expected the previous naming
+        style.  However, the presence of this pin is holding back environments that
+        otherwise want to use a newer setuptools, so we've decided to move forward
+        with this change, with the assumption that build environments will have
+        largely accommodated the setuptools change by now.
+
+
+
+    .. change::
+        :tags: bug, postgresql
+        :tickets: 11821
+
+        Revising the asyncpg ``terminate()`` fix first made in :ticket:`10717`
+        which improved the resiliency of this call under all circumstances, adding
+        ``asyncio.CancelledError`` to the list of exceptions that are intercepted
+        as failing for a graceful ``.close()`` which will then proceed to call
+        ``.terminate()``.
+
+    .. change::
+        :tags: bug, mssql
+        :tickets: 11822
+
+        Added error "The server failed to resume the transaction" to the list of
+        error strings for the pymssql driver in determining a disconnect scenario,
+        as observed by one user using pymssql under otherwise unknown conditions as
+        leaving an unusable connection in the connection pool which fails to ping
+        cleanly.
+
+    .. change::
+        :tags: bug, tests
+
+        Added missing ``array_type`` property to the testing suite
+        ``SuiteRequirements`` class.
+
 .. changelog::
     :version: 2.0.32
     :released: August 5, 2024
diff --git a/doc/build/changelog/unreleased_14/11728.rst b/doc/build/changelog/unreleased_14/11728.rst
new file mode 100644
index 00000000000..b27aa3333d7
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/11728.rst
@@ -0,0 +1,9 @@
+.. change::
+    :tags: bug, regression, orm
+    :tickets: 11728
+    :versions: 2.0.33
+
+    Fixed regression from 1.3 where the column key used for a hybrid property
+    might be populated with that of the underlying column that it returns, for
+    a property that returns an ORM mapped column directly, rather than the key
+    used by the hybrid property itself.
diff --git a/doc/build/changelog/unreleased_14/11819.rst b/doc/build/changelog/unreleased_14/11819.rst
new file mode 100644
index 00000000000..6211eb487ee
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/11819.rst
@@ -0,0 +1,14 @@
+.. change::
+    :tags: bug, postgresql
+    :tickets: 11819
+    :versions: 2.0.33, 1.4.54
+
+    Fixed critical issue in the asyncpg driver where a rollback or commit that
+    fails specifically for the ``MissingGreenlet`` condition or any other error
+    that is not raised by asyncpg itself would discard the asyncpg transaction
+    in any case, even though the transaction were still idle, leaving to a
+    server side condition with an idle transaction that then goes back into the
+    connection pool.   The flags for "transaction closed" are now not reset for
+    errors that are raised outside of asyncpg itself.  When asyncpg itself
+    raises an error for ``.commit()`` or ``.rollback()``, asyncpg does then
+    discard of this transaction.
diff --git a/doc/build/changelog/unreleased_20/11703.rst b/doc/build/changelog/unreleased_20/11703.rst
deleted file mode 100644
index 5c703138a14..00000000000
--- a/doc/build/changelog/unreleased_20/11703.rst
+++ /dev/null
@@ -1,7 +0,0 @@
-.. change::
-    :tags: bug, sql, regression
-    :tickets: 11703
-
-    Fixed regression in :meth:`_sql.Select.with_statement_hint` and others
-    where the generative behavior of the method stopped producing a copy of the
-    object.
diff --git a/doc/build/changelog/unreleased_20/array_type.rst b/doc/build/changelog/unreleased_20/array_type.rst
deleted file mode 100644
index 9b0801faf5b..00000000000
--- a/doc/build/changelog/unreleased_20/array_type.rst
+++ /dev/null
@@ -1,5 +0,0 @@
-.. change::
-    :tags: bug, test
-
-    Added missing ``array_type`` property to the testing suite
-    ``SuiteRequirements`` class.
diff --git a/doc/build/dialects/index.rst b/doc/build/dialects/index.rst
index b0064a09e9a..1a230481961 100644
--- a/doc/build/dialects/index.rst
+++ b/doc/build/dialects/index.rst
@@ -109,7 +109,7 @@ Currently maintained external dialect projects for SQLAlchemy include:
 +------------------------------------------------+---------------------------------------+
 | Microsoft Access (via pyodbc)                  | sqlalchemy-access_                    |
 +------------------------------------------------+---------------------------------------+
-| Microsoft SQL Server (via python-tds)          | sqlalchemy-tds_                       |
+| Microsoft SQL Server (via python-tds)          | sqlalchemy-pytds_                     |
 +------------------------------------------------+---------------------------------------+
 | Microsoft SQL Server (via turbodbc)            | sqlalchemy-turbodbc_                  |
 +------------------------------------------------+---------------------------------------+
@@ -151,7 +151,7 @@ Currently maintained external dialect projects for SQLAlchemy include:
 .. _sqlalchemy-sqlany: https://github.com/sqlanywhere/sqlalchemy-sqlany
 .. _sqlalchemy-monetdb: https://github.com/gijzelaerr/sqlalchemy-monetdb
 .. _snowflake-sqlalchemy: https://github.com/snowflakedb/snowflake-sqlalchemy
-.. _sqlalchemy-tds: https://github.com/m32/sqlalchemy-tds
+.. _sqlalchemy-pytds: https://pypi.org/project/sqlalchemy-pytds/
 .. _sqlalchemy-cratedb: https://github.com/crate/sqlalchemy-cratedb
 .. _sqlalchemy-access: https://pypi.org/project/sqlalchemy-access/
 .. _elasticsearch-dbapi: https://github.com/preset-io/elasticsearch-dbapi/
diff --git a/examples/association/basic_association.py b/examples/association/basic_association.py
index d2271ad430e..7a5b46097e3 100644
--- a/examples/association/basic_association.py
+++ b/examples/association/basic_association.py
@@ -105,7 +105,7 @@ def __init__(self, item, price=None):
     )
 
     # print customers who bought 'MySQL Crowbar' on sale
-    q = session.query(Order).join("order_items", "item")
+    q = session.query(Order).join(OrderItem).join(Item)
     q = q.filter(
         and_(Item.description == "MySQL Crowbar", Item.price > OrderItem.price)
     )
diff --git a/examples/association/proxied_association.py b/examples/association/proxied_association.py
index 0ec8fa899ac..65dcd6c0b66 100644
--- a/examples/association/proxied_association.py
+++ b/examples/association/proxied_association.py
@@ -112,7 +112,8 @@ def __init__(self, item, price=None):
     # print customers who bought 'MySQL Crowbar' on sale
     orders = (
         session.query(Order)
-        .join("order_items", "item")
+        .join(OrderItem)
+        .join(Item)
         .filter(Item.description == "MySQL Crowbar")
         .filter(Item.price > OrderItem.price)
     )
diff --git a/lib/sqlalchemy/dialects/mssql/provision.py b/lib/sqlalchemy/dialects/mssql/provision.py
index 143d386c45e..1c684b1dfef 100644
--- a/lib/sqlalchemy/dialects/mssql/provision.py
+++ b/lib/sqlalchemy/dialects/mssql/provision.py
@@ -22,10 +22,17 @@
 from ...testing.provision import get_temp_table_name
 from ...testing.provision import log
 from ...testing.provision import normalize_sequence
+from ...testing.provision import post_configure_engine
 from ...testing.provision import run_reap_dbs
 from ...testing.provision import temp_table_keyword_args
 
 
+@post_configure_engine.for_db("mssql")
+def post_configure_engine(url, engine, follower_ident):
+    if engine.driver == "pyodbc":
+        engine.dialect.dbapi.pooling = False
+
+
 @generate_driver_url.for_db("mssql")
 def generate_driver_url(url, driver, query_str):
     backend = url.get_backend_name()
diff --git a/lib/sqlalchemy/dialects/mssql/pymssql.py b/lib/sqlalchemy/dialects/mssql/pymssql.py
index ea1f9bd3a7e..c4207987bcd 100644
--- a/lib/sqlalchemy/dialects/mssql/pymssql.py
+++ b/lib/sqlalchemy/dialects/mssql/pymssql.py
@@ -103,6 +103,7 @@ def is_disconnect(self, e, connection, cursor):
             "message 20006",  # Write to the server failed
             "message 20017",  # Unexpected EOF from the server
             "message 20047",  # DBPROCESS is dead or not enabled
+            "The server failed to resume the transaction",
         ):
             if msg in str(e):
                 return True
diff --git a/lib/sqlalchemy/dialects/mysql/base.py b/lib/sqlalchemy/dialects/mysql/base.py
index d5db02d2781..aa99bf4d684 100644
--- a/lib/sqlalchemy/dialects/mysql/base.py
+++ b/lib/sqlalchemy/dialects/mysql/base.py
@@ -1349,7 +1349,7 @@ def visit_on_duplicate_key_update(self, on_duplicate, **kw):
 
         clauses = []
 
-        requires_mysql8_alias = (
+        requires_mysql8_alias = statement.select is None and (
             self.dialect._requires_alias_for_on_duplicate_key
         )
 
diff --git a/lib/sqlalchemy/dialects/oracle/cx_oracle.py b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
index 873d943371d..ed9b02d3fb1 100644
--- a/lib/sqlalchemy/dialects/oracle/cx_oracle.py
+++ b/lib/sqlalchemy/dialects/oracle/cx_oracle.py
@@ -62,7 +62,7 @@
 
 Users of Oracle Cloud should use this syntax and also configure the cloud
 wallet as shown in cx_Oracle documentation `Connecting to Autononmous Databases
-<https://cx-oracle.readthedocs.io/en/latest/user_guide/connection_handling.html#connecting-to-autononmous-databases>`_.
+<https://cx-oracle.readthedocs.io/en/latest/user_guide/connection_handling.html#autonomousdb>`_.
 
 SID Connections
 ^^^^^^^^^^^^^^^
diff --git a/lib/sqlalchemy/dialects/postgresql/asyncpg.py b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
index cb6b75154f3..48acbdceed3 100644
--- a/lib/sqlalchemy/dialects/postgresql/asyncpg.py
+++ b/lib/sqlalchemy/dialects/postgresql/asyncpg.py
@@ -865,27 +865,47 @@ async def _start_transaction(self):
         else:
             self._started = True
 
+    async def _rollback_and_discard(self):
+        try:
+            await self._transaction.rollback()
+        finally:
+            # if asyncpg .rollback() was actually called, then whether or
+            # not it raised or succeeded, the transation is done, discard it
+            self._transaction = None
+            self._started = False
+
+    async def _commit_and_discard(self):
+        try:
+            await self._transaction.commit()
+        finally:
+            # if asyncpg .commit() was actually called, then whether or
+            # not it raised or succeeded, the transation is done, discard it
+            self._transaction = None
+            self._started = False
+
     def rollback(self):
         if self._started:
             assert self._transaction is not None
             try:
-                await_(self._transaction.rollback())
-            except Exception as error:
-                self._handle_exception(error)
-            finally:
+                await_(self._rollback_and_discard())
                 self._transaction = None
                 self._started = False
+            except Exception as error:
+                # don't dereference asyncpg transaction if we didn't
+                # actually try to call rollback() on it
+                self._handle_exception(error)
 
     def commit(self):
         if self._started:
             assert self._transaction is not None
             try:
-                await_(self._transaction.commit())
-            except Exception as error:
-                self._handle_exception(error)
-            finally:
+                await_(self._commit_and_discard())
                 self._transaction = None
                 self._started = False
+            except Exception as error:
+                # don't dereference asyncpg transaction if we didn't
+                # actually try to call commit() on it
+                self._handle_exception(error)
 
     def close(self):
         self.rollback()
@@ -902,6 +922,7 @@ def terminate(self):
                 await_(self._connection.close(timeout=2))
             except (
                 asyncio.TimeoutError,
+                asyncio.CancelledError,
                 OSError,
                 self.dbapi.asyncpg.PostgresError,
             ):
diff --git a/lib/sqlalchemy/dialects/postgresql/json.py b/lib/sqlalchemy/dialects/postgresql/json.py
index 3790fa359b1..1cdafbd03d9 100644
--- a/lib/sqlalchemy/dialects/postgresql/json.py
+++ b/lib/sqlalchemy/dialects/postgresql/json.py
@@ -256,22 +256,27 @@ class Comparator(JSON.Comparator):
         """Define comparison operations for :class:`_types.JSON`."""
 
         def has_key(self, other):
-            """Boolean expression.  Test for presence of a key.  Note that the
-            key may be a SQLA expression.
+            """Boolean expression.  Test for presence of a key (equivalent of
+            the ``?`` operator).  Note that the key may be a SQLA expression.
             """
             return self.operate(HAS_KEY, other, result_type=sqltypes.Boolean)
 
         def has_all(self, other):
-            """Boolean expression.  Test for presence of all keys in jsonb"""
+            """Boolean expression.  Test for presence of all keys in jsonb
+            (equivalent of the ``?&`` operator)
+            """
             return self.operate(HAS_ALL, other, result_type=sqltypes.Boolean)
 
         def has_any(self, other):
-            """Boolean expression.  Test for presence of any key in jsonb"""
+            """Boolean expression.  Test for presence of any key in jsonb
+            (equivalent of the ``?|`` operator)
+            """
             return self.operate(HAS_ANY, other, result_type=sqltypes.Boolean)
 
         def contains(self, other, **kwargs):
             """Boolean expression.  Test if keys (or array) are a superset
-            of/contained the keys of the argument jsonb expression.
+            of/contained the keys of the argument jsonb expression
+            (equivalent of the ``@>`` operator).
 
             kwargs may be ignored by this operator but are required for API
             conformance.
@@ -280,7 +285,8 @@ def contains(self, other, **kwargs):
 
         def contained_by(self, other):
             """Boolean expression.  Test if keys are a proper subset of the
-            keys of the argument jsonb expression.
+            keys of the argument jsonb expression
+            (equivalent of the ``<@`` operator).
             """
             return self.operate(
                 CONTAINED_BY, other, result_type=sqltypes.Boolean
@@ -288,7 +294,7 @@ def contained_by(self, other):
 
         def delete_path(self, array):
             """JSONB expression. Deletes field or array element specified in
-            the argument array.
+            the argument array (equivalent of the ``#-`` operator).
 
             The input may be a list of strings that will be coerced to an
             ``ARRAY`` or an instance of :meth:`_postgres.array`.
@@ -302,7 +308,7 @@ def delete_path(self, array):
 
         def path_exists(self, other):
             """Boolean expression. Test for presence of item given by the
-            argument JSONPath expression.
+            argument JSONPath expression (equivalent of the ``@?`` operator).
 
             .. versionadded:: 2.0
             """
@@ -312,7 +318,8 @@ def path_exists(self, other):
 
         def path_match(self, other):
             """Boolean expression. Test if JSONPath predicate given by the
-            argument JSONPath expression matches.
+            argument JSONPath expression matches
+            (equivalent of the ``@@`` operator).
 
             Only the first item of the result is taken into account.
 
diff --git a/lib/sqlalchemy/dialects/sqlite/base.py b/lib/sqlalchemy/dialects/sqlite/base.py
index 04e84a68d2e..5e32e2fbb06 100644
--- a/lib/sqlalchemy/dialects/sqlite/base.py
+++ b/lib/sqlalchemy/dialects/sqlite/base.py
@@ -2588,8 +2588,8 @@ def parse_uqs():
                 return
             UNIQUE_PATTERN = r'(?:CONSTRAINT "?(.+?)"? +)?UNIQUE *\((.+?)\)'
             INLINE_UNIQUE_PATTERN = (
-                r'(?:(".+?")|(?:[\[`])?([a-z0-9_]+)(?:[\]`])?) '
-                r"+[a-z0-9_ ]+? +UNIQUE"
+                r'(?:(".+?")|(?:[\[`])?([a-z0-9_]+)(?:[\]`])?)[\t ]'
+                r"+[a-z0-9_ ]+?[\t ]+UNIQUE"
             )
 
             for match in re.finditer(UNIQUE_PATTERN, table_data, re.I):
@@ -2624,15 +2624,69 @@ def get_check_constraints(self, connection, table_name, schema=None, **kw):
             connection, table_name, schema=schema, **kw
         )
 
-        CHECK_PATTERN = r"(?:CONSTRAINT (.+) +)?" r"CHECK *\( *(.+) *\),? *"
+        # Notes:
+        # * The pattern currently matches any character for the name of the
+        #   constraint, including newline characters (re.S flag) as long as
+        #   none of the SQLite's table constraints keywords are encountered
+        #   by a negative lookahead.
+        #   This prevents the pattern from matching subsequent constraints
+        #   as part of the name.
+        #   This is only done for those keywords if seperated by spaces, to
+        #   support constraint names that contains them e.g. "check_value".
+        #
+        # * Because check constraint definitions can also contain newline
+        #   or tab characters, the pattern matches any character untill either
+        #   the beginning of the next constraint statement using a
+        #   non-capturing and non-consuming group, allowing the next one
+        #   to match, or the end of the table definition
+        #   e.g. newline and closing ')'.
+        CHECK_PATTERN = r"""
+        # Non-capturing group for the name part of named check constraints.
+        # This group is optional as unnamed check constraints can exist.
+        (?:
+        # Match beginning of constraint definition seperated by whitespace.
+        CONSTRAINT\s
+
+        # First capturing group that matches the actual name of the constraint.
+        # Any characters is allowed, as long as none of the reserved table
+        # constraint keywords are encountered using a negative lookahead.
+        ((?:(?!\sPRIMARY\s|\sFOREIGN\sKEY|\sUNIQUE\s|\sCHECK\s).)+)
+
+        # End of optional non-capturing name group seperated by whitespace.
+        \s)?
+
+        # Match beginning of the check expression with starting parenthesis
+        # and optional whitespace.
+        CHECK\s?\(
+
+        # Match actual expression, which can be any character.
+        (.+?)
+
+        # End parenthesis of the check expression.
+        \)
+
+        # Non-capturing group that helps denote the end of the check
+        # expression part.
+        # This can either be (1) the beginning of the next constraint,
+        # or (2) the end of the table definition.
+        (?:
+
+        # (1) Matches end of check constraint with trailing comma,
+        # optional whitespace (including newline), and the beginning
+        # of the next constraint (either named or unnamed).
+        ,[\s\n]*(?=CONSTRAINT|CHECK|UNIQUE|FOREIGN|PRIMARY)
+        # OR operator, seperating (1) & (2)
+        |
+        # (2) Matches end parenthesis of table definition, seperated by
+        # newline.
+        \n\)
+        # End of non-capturing group.
+        )
+        """
         cks = []
-        # NOTE: we aren't using re.S here because we actually are
-        # taking advantage of each CHECK constraint being all on one
-        # line in the table definition in order to delineate.  This
-        # necessarily makes assumptions as to how the CREATE TABLE
-        # was emitted.
-
-        for match in re.finditer(CHECK_PATTERN, table_data or "", re.I):
+        for match in re.finditer(
+            CHECK_PATTERN, table_data or "", re.I | re.S | re.VERBOSE
+        ):
             name = match.group(1)
 
             if name:
diff --git a/lib/sqlalchemy/dialects/sqlite/pysqlite.py b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
index 43ce043a97d..ab6ce6dc436 100644
--- a/lib/sqlalchemy/dialects/sqlite/pysqlite.py
+++ b/lib/sqlalchemy/dialects/sqlite/pysqlite.py
@@ -52,11 +52,11 @@
     e = create_engine('sqlite:///C:\\path\\to\\database.db')
 
 To use sqlite ``:memory:`` database specify it as the filename using
-``sqlite://:memory:``. It's also the default if no filepath is
+``sqlite:///:memory:``. It's also the default if no filepath is
 present, specifying only ``sqlite://`` and nothing else::
 
-    # in-memory database
-    e = create_engine('sqlite://:memory:')
+    # in-memory database (note three slashes)
+    e = create_engine('sqlite:///:memory:')
     # also in-memory database
     e2 = create_engine('sqlite://')
 
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 02a757379a8..58e3aa390fc 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -56,6 +56,7 @@
 from ..sql import operators
 from ..sql import schema as sa_schema
 from ..sql.cache_key import _ad_hoc_cache_key_from_args
+from ..sql.elements import quoted_name
 from ..sql.elements import TextClause
 from ..sql.type_api import TypeEngine
 from ..sql.visitors import InternalTraversal
@@ -89,8 +90,16 @@ def cache(
     exclude = {"info_cache", "unreflectable"}
     key = (
         fn.__name__,
-        tuple(a for a in args if isinstance(a, str)),
-        tuple((k, v) for k, v in kw.items() if k not in exclude),
+        tuple(
+            (str(a), a.quote) if isinstance(a, quoted_name) else a
+            for a in args
+            if isinstance(a, str)
+        ),
+        tuple(
+            (k, (str(v), v.quote) if isinstance(v, quoted_name) else v)
+            for k, v in kw.items()
+            if k not in exclude
+        ),
     )
     ret: _R = info_cache.get(key)
     if ret is None:
diff --git a/lib/sqlalchemy/orm/clsregistry.py b/lib/sqlalchemy/orm/clsregistry.py
index 26113d8b24d..382d6aef9be 100644
--- a/lib/sqlalchemy/orm/clsregistry.py
+++ b/lib/sqlalchemy/orm/clsregistry.py
@@ -287,8 +287,9 @@ def __getitem__(self, name: str) -> ClsRegistryToken:
 
     def _remove_item(self, name: str) -> None:
         self.contents.pop(name, None)
-        if not self.contents and self.parent is not None:
-            self.parent._remove_item(self.name)
+        if not self.contents:
+            if self.parent is not None:
+                self.parent._remove_item(self.name)
             _registries.discard(self)
 
     def resolve_attr(self, key: str) -> Union[_ModNS, Type[Any]]:
diff --git a/lib/sqlalchemy/orm/decl_base.py b/lib/sqlalchemy/orm/decl_base.py
index 90396403c2b..271c61a8b6e 100644
--- a/lib/sqlalchemy/orm/decl_base.py
+++ b/lib/sqlalchemy/orm/decl_base.py
@@ -431,7 +431,7 @@ def _setup_inheritance(self, mapper_kw: _MapperKwArgs) -> None:
 class _CollectedAnnotation(NamedTuple):
     raw_annotation: _AnnotationScanType
     mapped_container: Optional[Type[Mapped[Any]]]
-    extracted_mapped_annotation: Union[Type[Any], str]
+    extracted_mapped_annotation: Union[_AnnotationScanType, str]
     is_dataclass: bool
     attr_value: Any
     originating_module: str
diff --git a/lib/sqlalchemy/orm/util.py b/lib/sqlalchemy/orm/util.py
index d4dff11e454..6d6fc147151 100644
--- a/lib/sqlalchemy/orm/util.py
+++ b/lib/sqlalchemy/orm/util.py
@@ -90,6 +90,7 @@
     de_stringify_union_elements as _de_stringify_union_elements,
 )
 from ..util.typing import eval_name_only as _eval_name_only
+from ..util.typing import fixup_container_fwd_refs
 from ..util.typing import is_origin_of_cls
 from ..util.typing import Literal
 from ..util.typing import TupleAny
@@ -2323,7 +2324,7 @@ def _extract_mapped_subtype(
     is_dataclass_field: bool,
     expect_mapped: bool = True,
     raiseerr: bool = True,
-) -> Optional[Tuple[Union[type, str], Optional[type]]]:
+) -> Optional[Tuple[Union[_AnnotationScanType, str], Optional[type]]]:
     """given an annotation, figure out if it's ``Mapped[something]`` and if
     so, return the ``something`` part.
 
@@ -2409,7 +2410,11 @@ def _extract_mapped_subtype(
                 "Expected sub-type for Mapped[] annotation"
             )
 
-        return annotated.__args__[0], annotated.__origin__
+        return (
+            # fix dict/list/set args to be ForwardRef, see #11814
+            fixup_container_fwd_refs(annotated.__args__[0]),
+            annotated.__origin__,
+        )
 
 
 def _mapper_property_as_plain_name(prop: Type[Any]) -> str:
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 64b686fc037..2b6835a6a2b 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -5347,7 +5347,14 @@ def __init__(self, element, values):
 
     def _with_annotations(self, values):
         clone = super()._with_annotations(values)
-        clone.__dict__.pop("comparator", None)
+        for attr in (
+            "comparator",
+            "_proxy_key",
+            "_tq_key_label",
+            "_tq_label",
+            "_non_anon_label",
+        ):
+            clone.__dict__.pop(attr, None)
         return clone
 
     @util.memoized_property
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 1ecb680e446..21c44d8170a 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -1435,7 +1435,7 @@ def referred_schema_fn(table, to_schema,
 
         args = []
         for col in self.columns:
-            args.append(col._copy(schema=actual_schema))
+            args.append(col._copy(schema=actual_schema, _to_metadata=metadata))
         table = Table(
             name,
             metadata,
@@ -2477,6 +2477,8 @@ def _copy(self, **kw: Any) -> Column[Any]:
         server_onupdate = self.server_onupdate
         if isinstance(server_default, (Computed, Identity)):
             # TODO: likely should be copied in all cases
+            # TODO: if a Sequence, we would need to transfer the Sequence
+            # .metadata as well
             args.append(server_default._copy(**kw))
             server_default = server_onupdate = None
 
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index f38e6cea0a5..f4bc0986d08 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -4165,7 +4165,7 @@ def slice(
 
         For example, ::
 
-            stmt = select(User).order_by(User).id.slice(1, 3)
+            stmt = select(User).order_by(User.id).slice(1, 3)
 
         renders as
 
@@ -5838,22 +5838,35 @@ def reduce_columns(
         )
         return woc
 
-    # START OVERLOADED FUNCTIONS self.with_only_columns Select 8
+    # START OVERLOADED FUNCTIONS self.with_only_columns Select 1-8 ", *, maintain_column_froms: bool =..." # noqa: E501
 
     # code within this block is **programmatically,
-    # statically generated** by tools/generate_sel_v1_overloads.py
+    # statically generated** by tools/generate_tuple_map_overloads.py
 
     @overload
-    def with_only_columns(self, __ent0: _TCCA[_T0]) -> Select[_T0]: ...
+    def with_only_columns(
+        self, __ent0: _TCCA[_T0], /, *, maintain_column_froms: bool = ...
+    ) -> Select[_T0]: ...
 
     @overload
     def with_only_columns(
-        self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1]
+        self,
+        __ent0: _TCCA[_T0],
+        __ent1: _TCCA[_T1],
+        /,
+        *,
+        maintain_column_froms: bool = ...,
     ) -> Select[_T0, _T1]: ...
 
     @overload
     def with_only_columns(
-        self, __ent0: _TCCA[_T0], __ent1: _TCCA[_T1], __ent2: _TCCA[_T2]
+        self,
+        __ent0: _TCCA[_T0],
+        __ent1: _TCCA[_T1],
+        __ent2: _TCCA[_T2],
+        /,
+        *,
+        maintain_column_froms: bool = ...,
     ) -> Select[_T0, _T1, _T2]: ...
 
     @overload
@@ -5863,6 +5876,9 @@ def with_only_columns(
         __ent1: _TCCA[_T1],
         __ent2: _TCCA[_T2],
         __ent3: _TCCA[_T3],
+        /,
+        *,
+        maintain_column_froms: bool = ...,
     ) -> Select[_T0, _T1, _T2, _T3]: ...
 
     @overload
@@ -5873,6 +5889,9 @@ def with_only_columns(
         __ent2: _TCCA[_T2],
         __ent3: _TCCA[_T3],
         __ent4: _TCCA[_T4],
+        /,
+        *,
+        maintain_column_froms: bool = ...,
     ) -> Select[_T0, _T1, _T2, _T3, _T4]: ...
 
     @overload
@@ -5884,6 +5903,9 @@ def with_only_columns(
         __ent3: _TCCA[_T3],
         __ent4: _TCCA[_T4],
         __ent5: _TCCA[_T5],
+        /,
+        *,
+        maintain_column_froms: bool = ...,
     ) -> Select[_T0, _T1, _T2, _T3, _T4, _T5]: ...
 
     @overload
@@ -5896,6 +5918,9 @@ def with_only_columns(
         __ent4: _TCCA[_T4],
         __ent5: _TCCA[_T5],
         __ent6: _TCCA[_T6],
+        /,
+        *,
+        maintain_column_froms: bool = ...,
     ) -> Select[_T0, _T1, _T2, _T3, _T4, _T5, _T6]: ...
 
     @overload
@@ -5909,7 +5934,10 @@ def with_only_columns(
         __ent5: _TCCA[_T5],
         __ent6: _TCCA[_T6],
         __ent7: _TCCA[_T7],
-    ) -> Select[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7]: ...
+        /,
+        *entities: _ColumnsClauseArgument[Any],
+        maintain_column_froms: bool = ...,
+    ) -> Select[_T0, _T1, _T2, _T3, _T4, _T5, _T6, _T7, Unpack[TupleAny]]: ...
 
     # END OVERLOADED FUNCTIONS self.with_only_columns
 
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 0a411ce349d..145fce2fb40 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -1086,6 +1086,11 @@ def copy(self, **kw):
         return self.adapt(
             cast("Type[TypeEngine[Any]]", self.__class__),
             _create_events=True,
+            metadata=(
+                kw.get("_to_metadata", self.metadata)
+                if self.metadata is not None
+                else None
+            ),
         )
 
     @overload
@@ -1909,6 +1914,13 @@ def __init__(
         if _adapted_from:
             self.dispatch = self.dispatch._join(_adapted_from.dispatch)
 
+    def copy(self, **kw):
+        # override SchemaType.copy() to not include to_metadata logic
+        return self.adapt(
+            cast("Type[TypeEngine[Any]]", self.__class__),
+            _create_events=True,
+        )
+
     def _should_create_constraint(self, compiler, **kw):
         if not self._is_impl_for_variant(compiler.dialect, kw):
             return False
diff --git a/lib/sqlalchemy/testing/provision.py b/lib/sqlalchemy/testing/provision.py
index 74cdb0c73d9..b57ec1afb52 100644
--- a/lib/sqlalchemy/testing/provision.py
+++ b/lib/sqlalchemy/testing/provision.py
@@ -361,7 +361,7 @@ def update_db_opts(db_url, db_opts, options):
 def post_configure_engine(url, engine, follower_ident):
     """Perform extra steps after configuring an engine for testing.
 
-    (For the internal dialects, currently only used by sqlite, oracle)
+    (For the internal dialects, currently only used by sqlite, oracle, mssql)
     """
 
 
diff --git a/lib/sqlalchemy/util/typing.py b/lib/sqlalchemy/util/typing.py
index cfc3a26a971..7be6589e03d 100644
--- a/lib/sqlalchemy/util/typing.py
+++ b/lib/sqlalchemy/util/typing.py
@@ -190,9 +190,51 @@ def de_stringify_annotation(
         )
 
         return _copy_generic_annotation_with(annotation, elements)
+
     return annotation  # type: ignore
 
 
+def fixup_container_fwd_refs(
+    type_: _AnnotationScanType,
+) -> _AnnotationScanType:
+    """Correct dict['x', 'y'] into dict[ForwardRef('x'), ForwardRef('y')]
+    and similar for list, set
+
+    """
+
+    if (
+        is_generic(type_)
+        and typing_get_origin(type_)
+        in (
+            dict,
+            set,
+            list,
+            collections_abc.MutableSet,
+            collections_abc.MutableMapping,
+            collections_abc.MutableSequence,
+            collections_abc.Mapping,
+            collections_abc.Sequence,
+        )
+        # fight, kick and scream to struggle to tell the difference between
+        # dict[] and typing.Dict[] which DO NOT compare the same and DO NOT
+        # behave the same yet there is NO WAY to distinguish between which type
+        # it is using public attributes
+        and not re.match(
+            "typing.(?:Dict|List|Set|.*Mapping|.*Sequence|.*Set)", repr(type_)
+        )
+    ):
+        # compat with py3.10 and earlier
+        return typing_get_origin(type_).__class_getitem__(  # type: ignore
+            tuple(
+                [
+                    ForwardRef(elem) if isinstance(elem, str) else elem
+                    for elem in typing_get_args(type_)
+                ]
+            )
+        )
+    return type_
+
+
 def _copy_generic_annotation_with(
     annotation: GenericProtocol[_T], elements: Tuple[_AnnotationScanType, ...]
 ) -> Type[_T]:
diff --git a/pyproject.toml b/pyproject.toml
index 075254645d6..bcbf8599dd7 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,9 +1,7 @@
 [build-system]
 build-backend = "setuptools.build_meta"
 requires = [
-    # avoid moving to https://github.com/pypa/setuptools/issues/3593
-    # until we're ready
-    "setuptools>=61.0,<69.3",
+    "setuptools>=61.0",
     "cython>=3; platform_python_implementation == 'CPython'", # Skip cython when using pypy
 ]
 
diff --git a/tools/format_docs_code.py b/tools/format_docs_code.py
index 7bae0126b02..8d24a9163af 100644
--- a/tools/format_docs_code.py
+++ b/tools/format_docs_code.py
@@ -6,6 +6,7 @@
 .. versionadded:: 2.0
 
 """
+
 # mypy: ignore-errors
 
 from argparse import ArgumentParser
@@ -316,11 +317,13 @@ def main(
             print(
                 f"{to_reformat} file(s) would be reformatted;",
                 (
-                    f"{sum(formatting_error_counts)} formatting errors "
-                    f"reported in {len(formatting_error_counts)} files"
-                )
-                if formatting_error_counts
-                else "no formatting errors reported",
+                    (
+                        f"{sum(formatting_error_counts)} formatting errors "
+                        f"reported in {len(formatting_error_counts)} files"
+                    )
+                    if formatting_error_counts
+                    else "no formatting errors reported"
+                ),
             )
 
             exit(1)
@@ -388,9 +391,11 @@ def main(
             for val in config.get("target_version", [])
             if val != "py27"
         },
-        line_length=config.get("line_length", DEFAULT_LINE_LENGTH)
-        if args.project_line_length
-        else DEFAULT_LINE_LENGTH,
+        line_length=(
+            config.get("line_length", DEFAULT_LINE_LENGTH)
+            if args.project_line_length
+            else DEFAULT_LINE_LENGTH
+        ),
     )
     REPORT_ONLY_DOCTEST = args.report_doctest
 
diff --git a/tools/generate_proxy_methods.py b/tools/generate_proxy_methods.py
index 9881d26426f..31832ae8bfa 100644
--- a/tools/generate_proxy_methods.py
+++ b/tools/generate_proxy_methods.py
@@ -40,6 +40,7 @@
 .. versionadded:: 2.0
 
 """
+
 # mypy: ignore-errors
 
 from __future__ import annotations
@@ -85,9 +86,9 @@ def __repr__(self) -> str:
         return self.sym
 
 
-classes: collections.defaultdict[
-    str, Dict[str, Tuple[Any, ...]]
-] = collections.defaultdict(dict)
+classes: collections.defaultdict[str, Dict[str, Tuple[Any, ...]]] = (
+    collections.defaultdict(dict)
+)
 
 _T = TypeVar("_T", bound="Any")
 
@@ -214,18 +215,22 @@ def instrument(buf: TextIO, name: str, clslevel: bool = False) -> None:
 
             if spec.defaults:
                 new_defaults = tuple(
-                    _repr_sym("util.EMPTY_DICT")
-                    if df is util.EMPTY_DICT
-                    else df
+                    (
+                        _repr_sym("util.EMPTY_DICT")
+                        if df is util.EMPTY_DICT
+                        else df
+                    )
                     for df in spec.defaults
                 )
                 elem[3] = new_defaults
 
             if spec.kwonlydefaults:
                 new_kwonlydefaults = {
-                    name: _repr_sym("util.EMPTY_DICT")
-                    if df is util.EMPTY_DICT
-                    else df
+                    name: (
+                        _repr_sym("util.EMPTY_DICT")
+                        if df is util.EMPTY_DICT
+                        else df
+                    )
                     for name, df in spec.kwonlydefaults.items()
                 }
                 elem[5] = new_kwonlydefaults
@@ -415,9 +420,9 @@ def main(cmd: code_writer_cmd) -> None:
     from sqlalchemy import util
     from sqlalchemy.util import langhelpers
 
-    util.create_proxy_methods = (
-        langhelpers.create_proxy_methods
-    ) = create_proxy_methods
+    util.create_proxy_methods = langhelpers.create_proxy_methods = (
+        create_proxy_methods
+    )
 
     for entry in entries:
         if cmd.args.module in {"all", entry}:
diff --git a/tools/generate_sql_functions.py b/tools/generate_sql_functions.py
index 411cfed7219..b777ae406a2 100644
--- a/tools/generate_sql_functions.py
+++ b/tools/generate_sql_functions.py
@@ -1,6 +1,7 @@
 """Generate inline stubs for generic functions on func
 
 """
+
 # mypy: ignore-errors
 
 from __future__ import annotations
diff --git a/tools/generate_tuple_map_overloads.py b/tools/generate_tuple_map_overloads.py
index 9ca648333cd..a7a2eb5f430 100644
--- a/tools/generate_tuple_map_overloads.py
+++ b/tools/generate_tuple_map_overloads.py
@@ -16,6 +16,7 @@
 .. versionadded:: 2.0
 
 """
+
 # mypy: ignore-errors
 
 from __future__ import annotations
@@ -36,10 +37,13 @@
 sys.path.append(str(Path(__file__).parent.parent))
 
 
-def process_module(modname: str, filename: str, cmd: code_writer_cmd) -> str:
+def process_module(
+    modname: str, filename: str, expected_number: int, cmd: code_writer_cmd
+) -> str:
     # use tempfile in same path as the module, or at least in the
     # current working directory, so that black / zimports use
     # local pyproject.toml
+    found = 0
     with NamedTemporaryFile(
         mode="w",
         delete=False,
@@ -54,6 +58,7 @@ def process_module(modname: str, filename: str, cmd: code_writer_cmd) -> str:
                 line,
             )
             if m:
+                found += 1
                 indent = m.group(1)
                 given_fnname = current_fnname = m.group(2)
                 if current_fnname.startswith("self."):
@@ -116,16 +121,20 @@ def {current_fnname}(
 
             if not in_block:
                 buf.write(line)
+    if found != expected_number:
+        raise Exception(
+            f"{modname} processed {found}. expected {expected_number}"
+        )
     return buf.name
 
 
-def run_module(modname: str, cmd: code_writer_cmd) -> None:
+def run_module(modname: str, count: int, cmd: code_writer_cmd) -> None:
     cmd.write_status(f"importing module {modname}\n")
     mod = importlib.import_module(modname)
     destination_path = mod.__file__
     assert destination_path is not None
 
-    tempfile = process_module(modname, destination_path, cmd)
+    tempfile = process_module(modname, destination_path, count, cmd)
 
     cmd.run_zimports(tempfile)
     cmd.run_black(tempfile)
@@ -133,17 +142,17 @@ def run_module(modname: str, cmd: code_writer_cmd) -> None:
 
 
 def main(cmd: code_writer_cmd) -> None:
-    for modname in entries:
+    for modname, count in entries:
         if cmd.args.module in {"all", modname}:
-            run_module(modname, cmd)
+            run_module(modname, count, cmd)
 
 
 entries = [
-    "sqlalchemy.sql._selectable_constructors",
-    "sqlalchemy.orm.session",
-    "sqlalchemy.orm.query",
-    "sqlalchemy.sql.selectable",
-    "sqlalchemy.sql.dml",
+    ("sqlalchemy.sql._selectable_constructors", 1),
+    ("sqlalchemy.orm.session", 1),
+    ("sqlalchemy.orm.query", 1),
+    ("sqlalchemy.sql.selectable", 1),
+    ("sqlalchemy.sql.dml", 3),
 ]
 
 if __name__ == "__main__":
@@ -152,7 +161,7 @@ def main(cmd: code_writer_cmd) -> None:
     with cmd.add_arguments() as parser:
         parser.add_argument(
             "--module",
-            choices=entries + ["all"],
+            choices=[n for n, _ in entries] + ["all"],
             default="all",
             help="Which file to generate. Default is to regenerate all files",
         )
diff --git a/tools/trace_orm_adapter.py b/tools/trace_orm_adapter.py
index de8098bcb8f..966705690de 100644
--- a/tools/trace_orm_adapter.py
+++ b/tools/trace_orm_adapter.py
@@ -23,6 +23,7 @@
 
 
 """  # noqa: E501
+
 # mypy: ignore-errors
 
 


Test patch:
diff --git a/test/dialect/mssql/test_engine.py b/test/dialect/mssql/test_engine.py
index 557341aa6a4..0e9d2fdcf03 100644
--- a/test/dialect/mssql/test_engine.py
+++ b/test/dialect/mssql/test_engine.py
@@ -375,6 +375,7 @@ def test_pymssql_disconnect(self):
             "message 20006",  # Write to the server failed
             "message 20017",  # Unexpected EOF from the server
             "message 20047",  # DBPROCESS is dead or not enabled
+            "The server failed to resume the transaction",
         ]:
             eq_(dialect.is_disconnect(error, None, None), True)
 
diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py
index 6712300aa40..189390659ad 100644
--- a/test/dialect/mysql/test_compiler.py
+++ b/test/dialect/mysql/test_compiler.py
@@ -1127,6 +1127,31 @@ def test_from_values(self, version: Variation):
 
         self.assert_compile(stmt, expected_sql, dialect=dialect)
 
+    @testing.variation("version", ["mysql8", "all_others"])
+    def test_from_select(self, version: Variation):
+        stmt = insert(self.table).from_select(
+            ["id", "bar"],
+            select(self.table.c.id, literal("bar2")),
+        )
+        stmt = stmt.on_duplicate_key_update(
+            bar=stmt.inserted.bar, baz=stmt.inserted.baz
+        )
+
+        expected_sql = (
+            "INSERT INTO foos (id, bar) SELECT foos.id, %s AS anon_1 "
+            "FROM foos "
+            "ON DUPLICATE KEY UPDATE bar = VALUES(bar), baz = VALUES(baz)"
+        )
+        if version.all_others:
+            dialect = None
+        elif version.mysql8:
+            dialect = mysql.dialect()
+            dialect._requires_alias_for_on_duplicate_key = True
+        else:
+            version.fail()
+
+        self.assert_compile(stmt, expected_sql, dialect=dialect)
+
     def test_from_literal(self):
         stmt = insert(self.table).values(
             [{"id": 1, "bar": "ab"}, {"id": 2, "bar": "b"}]
diff --git a/test/dialect/mysql/test_on_duplicate.py b/test/dialect/mysql/test_on_duplicate.py
index 5a4e6ca8d5c..35aebb470c3 100644
--- a/test/dialect/mysql/test_on_duplicate.py
+++ b/test/dialect/mysql/test_on_duplicate.py
@@ -3,6 +3,8 @@
 from sqlalchemy import exc
 from sqlalchemy import func
 from sqlalchemy import Integer
+from sqlalchemy import literal
+from sqlalchemy import select
 from sqlalchemy import String
 from sqlalchemy import Table
 from sqlalchemy.dialects.mysql import insert
@@ -63,6 +65,22 @@ def test_on_duplicate_key_update_multirow(self, connection):
             [(1, "ab", "bz", False)],
         )
 
+    def test_on_duplicate_key_from_select(self, connection):
+        foos = self.tables.foos
+        conn = connection
+        conn.execute(insert(foos).values(dict(id=1, bar="b", baz="bz")))
+        stmt = insert(foos).from_select(
+            ["id", "bar", "baz"],
+            select(foos.c.id, literal("bar2"), literal("baz2")),
+        )
+        stmt = stmt.on_duplicate_key_update(bar=stmt.inserted.bar)
+
+        conn.execute(stmt)
+        eq_(
+            conn.execute(foos.select().where(foos.c.id == 1)).fetchall(),
+            [(1, "bar2", "bz", False)],
+        )
+
     def test_on_duplicate_key_update_singlerow(self, connection):
         foos = self.tables.foos
         conn = connection
diff --git a/test/dialect/postgresql/test_async_pg_py3k.py b/test/dialect/postgresql/test_async_pg_py3k.py
index c09acf5b472..feff60c5789 100644
--- a/test/dialect/postgresql/test_async_pg_py3k.py
+++ b/test/dialect/postgresql/test_async_pg_py3k.py
@@ -13,6 +13,7 @@
 from sqlalchemy.dialects.postgresql import ENUM
 from sqlalchemy.testing import async_test
 from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import mock
 
@@ -165,6 +166,54 @@ async def async_setup(engine, enums):
                 ],
             )
 
+    @testing.variation("trans", ["commit", "rollback"])
+    @async_test
+    async def test_dont_reset_open_transaction(
+        self, trans, async_testing_engine
+    ):
+        """test for #11819"""
+
+        engine = async_testing_engine()
+
+        control_conn = await engine.connect()
+        await control_conn.execution_options(isolation_level="AUTOCOMMIT")
+
+        conn = await engine.connect()
+        txid_current = (
+            await conn.exec_driver_sql("select txid_current()")
+        ).scalar()
+
+        with expect_raises(exc.MissingGreenlet):
+            if trans.commit:
+                conn.sync_connection.connection.dbapi_connection.commit()
+            elif trans.rollback:
+                conn.sync_connection.connection.dbapi_connection.rollback()
+            else:
+                trans.fail()
+
+        trans_exists = (
+            await control_conn.exec_driver_sql(
+                f"SELECT count(*) FROM pg_stat_activity "
+                f"where backend_xid={txid_current}"
+            )
+        ).scalar()
+        eq_(trans_exists, 1)
+
+        if trans.commit:
+            await conn.commit()
+        elif trans.rollback:
+            await conn.rollback()
+        else:
+            trans.fail()
+
+        trans_exists = (
+            await control_conn.exec_driver_sql(
+                f"SELECT count(*) FROM pg_stat_activity "
+                f"where backend_xid={txid_current}"
+            )
+        ).scalar()
+        eq_(trans_exists, 0)
+
     @async_test
     async def test_failed_commit_recover(self, metadata, async_testing_engine):
         Table("t1", metadata, Column("id", Integer, primary_key=True))
diff --git a/test/dialect/test_sqlite.py b/test/dialect/test_sqlite.py
index 8dedadbde9d..ebd5312177e 100644
--- a/test/dialect/test_sqlite.py
+++ b/test/dialect/test_sqlite.py
@@ -1816,6 +1816,29 @@ def setup_test_class(cls):
 
             Table("q", meta, Column("id", Integer), PrimaryKeyConstraint("id"))
 
+            # intentional new line
+            Table(
+                "r",
+                meta,
+                Column("id", Integer),
+                Column("value", Integer),
+                Column("prefix", String),
+                CheckConstraint("id > 0"),
+                UniqueConstraint("prefix", name="prefix_named"),
+                # Constraint definition with newline and tab characters
+                CheckConstraint(
+                    """((value > 0) AND \n\t(value < 100) AND \n\t
+                      (value != 50))""",
+                    name="ck_r_value_multiline",
+                ),
+                UniqueConstraint("value"),
+                # Constraint name with special chars and 'check' in the name
+                CheckConstraint("value IS NOT NULL", name="^check-r* #\n\t"),
+                PrimaryKeyConstraint("id", name="pk_name"),
+                # Constraint definition with special characters.
+                CheckConstraint("prefix NOT GLOB '*[^-. /#,]*'"),
+            )
+
             meta.create_all(conn)
 
             # will contain an "autoindex"
@@ -1911,6 +1934,7 @@ def teardown_test_class(cls):
                 "b",
                 "a1",
                 "a2",
+                "r",
             ]:
                 conn.exec_driver_sql("drop table %s" % name)
 
@@ -2426,6 +2450,27 @@ def test_unique_constraint_unnamed_normal_temporary(
             [{"column_names": ["x"], "name": None}],
         )
 
+    def test_unique_constraint_mixed_into_ck(self, connection):
+        """test #11832"""
+
+        inspector = inspect(connection)
+        eq_(
+            inspector.get_unique_constraints("r"),
+            [
+                {"name": "prefix_named", "column_names": ["prefix"]},
+                {"name": None, "column_names": ["value"]},
+            ],
+        )
+
+    def test_primary_key_constraint_mixed_into_ck(self, connection):
+        """test #11832"""
+
+        inspector = inspect(connection)
+        eq_(
+            inspector.get_pk_constraint("r"),
+            {"constrained_columns": ["id"], "name": "pk_name"},
+        )
+
     def test_primary_key_constraint_named(self):
         inspector = inspect(testing.db)
         eq_(
@@ -2456,6 +2501,21 @@ def test_check_constraint(self):
                 {"sqltext": "q > 1 AND q < 6", "name": None},
             ],
         )
+        eq_(
+            inspector.get_check_constraints("r"),
+            [
+                {"sqltext": "value IS NOT NULL", "name": "^check-r* #\n\t"},
+                # Triple-quote multi-line definition should have added a
+                # newline and whitespace:
+                {
+                    "sqltext": "((value > 0) AND \n\t(value < 100) AND \n\t\n"
+                    "                      (value != 50))",
+                    "name": "ck_r_value_multiline",
+                },
+                {"sqltext": "id > 0", "name": None},
+                {"sqltext": "prefix NOT GLOB '*[^-. /#,]*'", "name": None},
+            ],
+        )
 
     @testing.combinations(
         ("plain_name", "plain_name"),
@@ -2466,17 +2526,27 @@ def test_check_constraint(self):
         argnames="colname,expected",
     )
     @testing.combinations(
-        "uq", "uq_inline", "pk", "ix", argnames="constraint_type"
+        "uq",
+        "uq_inline",
+        "uq_inline_tab_before",  # tab before column params
+        "uq_inline_tab_within",  # tab within column params
+        "pk",
+        "ix",
+        argnames="constraint_type",
     )
     def test_constraint_cols(
         self, colname, expected, constraint_type, connection, metadata
     ):
-        if constraint_type == "uq_inline":
+        if constraint_type.startswith("uq_inline"):
+            inline_create_sql = {
+                "uq_inline": "CREATE TABLE t (%s INTEGER UNIQUE)",
+                "uq_inline_tab_before": "CREATE TABLE t (%s\tINTEGER UNIQUE)",
+                "uq_inline_tab_within": "CREATE TABLE t (%s INTEGER\tUNIQUE)",
+            }
+
             t = Table("t", metadata, Column(colname, Integer))
             connection.exec_driver_sql(
-                """
-            CREATE TABLE t (%s INTEGER UNIQUE)
-            """
+                inline_create_sql[constraint_type]
                 % connection.dialect.identifier_preparer.quote(colname)
             )
         else:
@@ -2494,7 +2564,12 @@ def test_constraint_cols(
 
             t.create(connection)
 
-        if constraint_type in ("uq", "uq_inline"):
+        if constraint_type in (
+            "uq",
+            "uq_inline",
+            "uq_inline_tab_before",
+            "uq_inline_tab_within",
+        ):
             const = inspect(connection).get_unique_constraints("t")[0]
             eq_(const["column_names"], [expected])
         elif constraint_type == "pk":
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py
index 003b457a51a..adb40370655 100644
--- a/test/engine/test_reflection.py
+++ b/test/engine/test_reflection.py
@@ -1,3 +1,4 @@
+import itertools
 import unicodedata
 
 import sqlalchemy as sa
@@ -19,6 +20,8 @@
 from sqlalchemy import testing
 from sqlalchemy import UniqueConstraint
 from sqlalchemy.engine import Inspector
+from sqlalchemy.engine.reflection import cache
+from sqlalchemy.sql.elements import quoted_name
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
@@ -2494,3 +2497,162 @@ def test_table_works_minus_fks(self, connection, tab_w_fks):
             "SELECT b_1.x, b_1.q, b_1.p, b_1.r, b_1.s, b_1.t "
             "FROM b AS b_1 JOIN a ON a.x = b_1.r",
         )
+
+
+class ReflectionCacheTest(fixtures.TestBase):
+    @testing.fixture(params=["arg", "kwarg"])
+    def cache(self, connection, request):
+        dialect = connection.dialect
+        info_cache = {}
+        counter = itertools.count(1)
+
+        @cache
+        def get_cached_name(self, connection, *args, **kw):
+            return next(counter)
+
+        def get_cached_name_via_arg(name):
+            return get_cached_name(
+                dialect, connection, name, info_cache=info_cache
+            )
+
+        def get_cached_name_via_kwarg(name):
+            return get_cached_name(
+                dialect, connection, name=name, info_cache=info_cache
+            )
+
+        if request.param == "arg":
+            yield get_cached_name_via_arg
+        elif request.param == "kwarg":
+            yield get_cached_name_via_kwarg
+        else:
+            assert False
+
+    @testing.fixture(params=[False, True])
+    def quote(self, request):
+        yield request.param
+
+    def test_single_string(self, cache):
+        # new value
+        eq_(cache("name1"), 1)
+
+        # same value, counter not incremented
+        eq_(cache("name1"), 1)
+
+    def test_multiple_string(self, cache):
+        # new value
+        eq_(cache("name1"), 1)
+        eq_(cache("name2"), 2)
+
+        # same values, counter not incremented
+        eq_(cache("name1"), 1)
+        eq_(cache("name2"), 2)
+
+    def test_single_quoted_name(self, cache, quote):
+        # new value
+        eq_(cache(quoted_name("name1", quote=quote)), 1)
+
+        # same value, counter not incremented
+        eq_(cache(quoted_name("name1", quote=quote)), 1)
+
+    def test_multiple_quoted_name(self, cache, quote):
+        # new value
+        eq_(cache(quoted_name("name1", quote=quote)), 1)
+        eq_(cache(quoted_name("name2", quote=quote)), 2)
+
+        # same values, counter not incremented
+        eq_(cache(quoted_name("name1", quote=quote)), 1)
+        eq_(cache(quoted_name("name2", quote=quote)), 2)
+
+    def test_single_quoted_name_and_string(self, cache, quote):
+        # new values
+        eq_(cache(quoted_name("n1", quote=quote)), 1)
+        eq_(cache("n1"), 2)
+
+        # same values, counter not incremented
+        eq_(cache(quoted_name("n1", quote=quote)), 1)
+        eq_(cache("n1"), 2)
+
+    def test_multiple_quoted_name_and_string(self, cache, quote):
+        # new values
+        eq_(cache(quoted_name("n1", quote=quote)), 1)
+        eq_(cache(quoted_name("n2", quote=quote)), 2)
+        eq_(cache("n1"), 3)
+        eq_(cache("n2"), 4)
+
+        # same values, counter not incremented
+        eq_(cache(quoted_name("n1", quote=quote)), 1)
+        eq_(cache(quoted_name("n2", quote=quote)), 2)
+        eq_(cache("n1"), 3)
+        eq_(cache("n2"), 4)
+
+    def test_single_quoted_name_false_true_and_string(self, cache, quote):
+        # new values
+        eq_(cache(quoted_name("n1", quote=quote)), 1)
+        eq_(cache(quoted_name("n1", quote=not quote)), 2)
+        eq_(cache("n1"), 3)
+
+        # same values, counter not incremented
+        eq_(cache(quoted_name("n1", quote=quote)), 1)
+        eq_(cache(quoted_name("n1", quote=not quote)), 2)
+        eq_(cache("n1"), 3)
+
+    def test_multiple_quoted_name_false_true_and_string(self, cache, quote):
+        # new values
+        eq_(cache(quoted_name("n1", quote=quote)), 1)
+        eq_(cache(quoted_name("n2", quote=quote)), 2)
+        eq_(cache(quoted_name("n1", quote=not quote)), 3)
+        eq_(cache(quoted_name("n2", quote=not quote)), 4)
+        eq_(cache("n1"), 5)
+        eq_(cache("n2"), 6)
+
+        # same values, counter not incremented
+        eq_(cache(quoted_name("n1", quote=quote)), 1)
+        eq_(cache(quoted_name("n2", quote=quote)), 2)
+        eq_(cache(quoted_name("n1", quote=not quote)), 3)
+        eq_(cache(quoted_name("n2", quote=not quote)), 4)
+        eq_(cache("n1"), 5)
+        eq_(cache("n2"), 6)
+
+    def test_multiple_quoted_name_false_true_and_string_arg_and_kwarg(
+        self, connection, quote
+    ):
+        dialect = connection.dialect
+        info_cache = {}
+        counter = itertools.count(1)
+
+        @cache
+        def get_cached_name(self, connection, *args, **kw):
+            return next(counter)
+
+        def cache_(*args, **kw):
+            return get_cached_name(
+                dialect, connection, *args, **kw, info_cache=info_cache
+            )
+
+        # new values
+        eq_(cache_(quoted_name("n1", quote=quote)), 1)
+        eq_(cache_(name=quoted_name("n1", quote=quote)), 2)
+        eq_(cache_(quoted_name("n2", quote=quote)), 3)
+        eq_(cache_(name=quoted_name("n2", quote=quote)), 4)
+        eq_(cache_(quoted_name("n1", quote=not quote)), 5)
+        eq_(cache_(name=quoted_name("n1", quote=not quote)), 6)
+        eq_(cache_(quoted_name("n2", quote=not quote)), 7)
+        eq_(cache_(name=quoted_name("n2", quote=not quote)), 8)
+        eq_(cache_("n1"), 9)
+        eq_(cache_(name="n1"), 10)
+        eq_(cache_("n2"), 11)
+        eq_(cache_(name="n2"), 12)
+
+        # same values, counter not incremented
+        eq_(cache_(quoted_name("n1", quote=quote)), 1)
+        eq_(cache_(name=quoted_name("n1", quote=quote)), 2)
+        eq_(cache_(quoted_name("n2", quote=quote)), 3)
+        eq_(cache_(name=quoted_name("n2", quote=quote)), 4)
+        eq_(cache_(quoted_name("n1", quote=not quote)), 5)
+        eq_(cache_(name=quoted_name("n1", quote=not quote)), 6)
+        eq_(cache_(quoted_name("n2", quote=not quote)), 7)
+        eq_(cache_(name=quoted_name("n2", quote=not quote)), 8)
+        eq_(cache_("n1"), 9)
+        eq_(cache_(name="n1"), 10)
+        eq_(cache_("n2"), 11)
+        eq_(cache_(name="n2"), 12)
diff --git a/test/ext/test_hybrid.py b/test/ext/test_hybrid.py
index 69e9c133515..8e3d7e9cd57 100644
--- a/test/ext/test_hybrid.py
+++ b/test/ext/test_hybrid.py
@@ -7,6 +7,7 @@
 from sqlalchemy import insert
 from sqlalchemy import inspect
 from sqlalchemy import Integer
+from sqlalchemy import LABEL_STYLE_DISAMBIGUATE_ONLY
 from sqlalchemy import LABEL_STYLE_TABLENAME_PLUS_COL
 from sqlalchemy import literal_column
 from sqlalchemy import Numeric
@@ -423,6 +424,21 @@ def name(self):
 
         return A
 
+    @testing.fixture
+    def _unnamed_expr_matches_col_fixture(self):
+        Base = declarative_base()
+
+        class A(Base):
+            __tablename__ = "a"
+            id = Column(Integer, primary_key=True)
+            foo = Column(String)
+
+            @hybrid.hybrid_property
+            def bar(self):
+                return self.foo
+
+        return A
+
     def test_access_from_unmapped(self):
         """test #9519"""
 
@@ -497,6 +513,39 @@ def test_labeling_for_unnamed(self, _unnamed_expr_fixture):
             "a.lastname AS name FROM a) AS anon_1",
         )
 
+    @testing.variation("pre_populate_col_proxy", [True, False])
+    def test_labeling_for_unnamed_matches_col(
+        self, _unnamed_expr_matches_col_fixture, pre_populate_col_proxy
+    ):
+        """test #11728"""
+
+        A = _unnamed_expr_matches_col_fixture
+
+        if pre_populate_col_proxy:
+            pre_stmt = select(A.id, A.foo)
+            pre_stmt.subquery().c
+
+        stmt = select(A.id, A.bar)
+        self.assert_compile(
+            stmt,
+            "SELECT a.id, a.foo FROM a",
+        )
+
+        compile_state = stmt._compile_state_factory(stmt, None)
+        eq_(
+            compile_state._column_naming_convention(
+                LABEL_STYLE_DISAMBIGUATE_ONLY, legacy=False
+            )(list(stmt.inner_columns)[1]),
+            "bar",
+        )
+        eq_(stmt.subquery().c.keys(), ["id", "bar"])
+
+        self.assert_compile(
+            select(stmt.subquery()),
+            "SELECT anon_1.id, anon_1.foo FROM "
+            "(SELECT a.id AS id, a.foo AS foo FROM a) AS anon_1",
+        )
+
     def test_labeling_for_unnamed_tablename_plus_col(
         self, _unnamed_expr_fixture
     ):
diff --git a/test/orm/declarative/test_clsregistry.py b/test/orm/declarative/test_clsregistry.py
index ffc8528125c..0cf775e4d27 100644
--- a/test/orm/declarative/test_clsregistry.py
+++ b/test/orm/declarative/test_clsregistry.py
@@ -230,7 +230,7 @@ def test_dupe_classes_cleanout(self):
             del f2
             gc_collect()
 
-            eq_(len(clsregistry._registries), 1)
+            eq_(len(clsregistry._registries), 0)
 
     def test_dupe_classes_name_race(self):
         """test the race condition that the class was garbage "
diff --git a/test/orm/declarative/test_tm_future_annotations_sync.py b/test/orm/declarative/test_tm_future_annotations_sync.py
index 5dca5e246c3..eb1e605d10e 100644
--- a/test/orm/declarative/test_tm_future_annotations_sync.py
+++ b/test/orm/declarative/test_tm_future_annotations_sync.py
@@ -1419,22 +1419,68 @@ class RefElementTwo(decl_base):
             Dict,
             (str, str),
         ),
+        (list, None, testing.requires.python310),
+        (
+            List,
+            None,
+        ),
+        (dict, None, testing.requires.python310),
+        (
+            Dict,
+            None,
+        ),
         id_="sa",
+        argnames="container_typ,args",
     )
-    def test_extract_generic_from_pep593(self, container_typ, args):
-        """test #9099"""
+    @testing.variation("style", ["pep593", "alias", "direct"])
+    def test_extract_composed(self, container_typ, args, style):
+        """test #9099 (pep593)
+
+        test #11814
+
+        test #11831, regression from #11814
+        """
 
         global TestType
-        TestType = Annotated[container_typ[args], 0]
+
+        if style.pep593:
+            if args is None:
+                TestType = Annotated[container_typ, 0]
+            else:
+                TestType = Annotated[container_typ[args], 0]
+        elif style.alias:
+            if args is None:
+                TestType = container_typ
+            else:
+                TestType = container_typ[args]
+        elif style.direct:
+            TestType = container_typ
 
         class Base(DeclarativeBase):
-            type_annotation_map = {TestType: JSON()}
+            if style.direct:
+                if args == (str, str):
+                    type_annotation_map = {TestType[str, str]: JSON()}
+                elif args is None:
+                    type_annotation_map = {TestType: JSON()}
+                else:
+                    type_annotation_map = {TestType[str]: JSON()}
+            else:
+                type_annotation_map = {TestType: JSON()}
 
         class MyClass(Base):
             __tablename__ = "my_table"
 
             id: Mapped[int] = mapped_column(primary_key=True)
-            data: Mapped[TestType] = mapped_column()
+
+            if style.direct:
+                if args == (str, str):
+                    data: Mapped[TestType[str, str]] = mapped_column()
+                elif args is None:
+                    data: Mapped[TestType] = mapped_column()
+                else:
+                    data: Mapped[TestType[str]] = mapped_column()
+            else:
+                data: Mapped[TestType] = mapped_column()
 
         is_(MyClass.__table__.c.data.type._type_affinity, JSON)
 
diff --git a/test/orm/declarative/test_typed_mapping.py b/test/orm/declarative/test_typed_mapping.py
index 25200514dc3..c9eacbae7da 100644
--- a/test/orm/declarative/test_typed_mapping.py
+++ b/test/orm/declarative/test_typed_mapping.py
@@ -1410,22 +1410,68 @@ class RefElementTwo(decl_base):
             Dict,
             (str, str),
         ),
+        (list, None, testing.requires.python310),
+        (
+            List,
+            None,
+        ),
+        (dict, None, testing.requires.python310),
+        (
+            Dict,
+            None,
+        ),
         id_="sa",
+        argnames="container_typ,args",
     )
-    def test_extract_generic_from_pep593(self, container_typ, args):
-        """test #9099"""
+    @testing.variation("style", ["pep593", "alias", "direct"])
+    def test_extract_composed(self, container_typ, args, style):
+        """test #9099 (pep593)
+
+        test #11814
+
+        test #11831, regression from #11814
+        """
 
         global TestType
-        TestType = Annotated[container_typ[args], 0]
+
+        if style.pep593:
+            if args is None:
+                TestType = Annotated[container_typ, 0]
+            else:
+                TestType = Annotated[container_typ[args], 0]
+        elif style.alias:
+            if args is None:
+                TestType = container_typ
+            else:
+                TestType = container_typ[args]
+        elif style.direct:
+            TestType = container_typ
 
         class Base(DeclarativeBase):
-            type_annotation_map = {TestType: JSON()}
+            if style.direct:
+                if args == (str, str):
+                    type_annotation_map = {TestType[str, str]: JSON()}
+                elif args is None:
+                    type_annotation_map = {TestType: JSON()}
+                else:
+                    type_annotation_map = {TestType[str]: JSON()}
+            else:
+                type_annotation_map = {TestType: JSON()}
 
         class MyClass(Base):
             __tablename__ = "my_table"
 
             id: Mapped[int] = mapped_column(primary_key=True)
-            data: Mapped[TestType] = mapped_column()
+
+            if style.direct:
+                if args == (str, str):
+                    data: Mapped[TestType[str, str]] = mapped_column()
+                elif args is None:
+                    data: Mapped[TestType] = mapped_column()
+                else:
+                    data: Mapped[TestType[str]] = mapped_column()
+            else:
+                data: Mapped[TestType] = mapped_column()
 
         is_(MyClass.__table__.c.data.type._type_affinity, JSON)
 
diff --git a/test/sql/test_metadata.py b/test/sql/test_metadata.py
index 97c2f086458..1b068c02f7f 100644
--- a/test/sql/test_metadata.py
+++ b/test/sql/test_metadata.py
@@ -2395,17 +2395,27 @@ def test_inherit_schema_enum(self):
         t1 = Table("x", m, Column("y", type_), schema="z")
         eq_(t1.c.y.type.schema, "z")
 
-    def test_to_metadata_copy_type(self):
+    @testing.variation("assign_metadata", [True, False])
+    def test_to_metadata_copy_type(self, assign_metadata):
         m1 = MetaData()
 
-        type_ = self.MyType()
+        if assign_metadata:
+            type_ = self.MyType(metadata=m1)
+        else:
+            type_ = self.MyType()
+
         t1 = Table("x", m1, Column("y", type_))
 
         m2 = MetaData()
         t2 = t1.to_metadata(m2)
 
-        # metadata isn't set
-        is_(t2.c.y.type.metadata, None)
+        if assign_metadata:
+            # metadata was transferred
+            # issue #11802
+            is_(t2.c.y.type.metadata, m2)
+        else:
+            # metadata isn't set
+            is_(t2.c.y.type.metadata, None)
 
         # our test type sets table, though
         is_(t2.c.y.type.table, t2)
@@ -2435,11 +2445,34 @@ def test_to_metadata_independent_schema(self):
 
         eq_(t2.c.y.type.schema, None)
 
-    def test_to_metadata_inherit_schema(self):
+    @testing.combinations(
+        ("name", "foobar", "name"),
+        ("schema", "someschema", "schema"),
+        ("inherit_schema", True, "inherit_schema"),
+        ("metadata", MetaData(), "metadata"),
+    )
+    def test_copy_args(self, argname, value, attrname):
+        kw = {argname: value}
+        e1 = self.MyType(**kw)
+
+        e1_copy = e1.copy()
+
+        eq_(getattr(e1_copy, attrname), value)
+
+    @testing.variation("already_has_a_schema", [True, False])
+    def test_to_metadata_inherit_schema(self, already_has_a_schema):
         m1 = MetaData()
 
-        type_ = self.MyType(inherit_schema=True)
+        if already_has_a_schema:
+            type_ = self.MyType(schema="foo", inherit_schema=True)
+            eq_(type_.schema, "foo")
+        else:
+            type_ = self.MyType(inherit_schema=True)
+
         t1 = Table("x", m1, Column("y", type_))
+        # note that inherit_schema means the schema mutates to be that
+        # of the table
+        is_(type_.schema, None)
 
         m2 = MetaData()
         t2 = t1.to_metadata(m2, schema="bar")