Allow custom sorting of column in the ORM.

To accommodate a change in column ordering used by ORM Declarative in
SQLAlchemy 2.0, a new parameter :paramref:`_orm.mapped_column.sort_order`
has been added that can be used to control the order of the columns defined
in the table by the ORM, for common use cases such as mixins with primary
key columns that should appear first in tables. The change notes at
:ref:`change_9297` illustrate the default change in ordering behavior
(which is part of all SQLAlchemy 2.0 releases) as well as use of the
:paramref:`_orm.mapped_column.sort_order` to control column ordering when
using mixins and multiple classes (new in 2.0.4).

Fixes: #9297
Change-Id: Ic7163d64efdc0eccb53d6ae0dd89ec83427fb675

7 files changed, 51 insertions, 20 deletions

diff --git a/lib/sqlalchemy/orm/_orm_constructors.py b/lib/sqlalchemy/orm/_orm_constructors.py
index 424385e1b..3bd1db79d 100644
--- a/lib/sqlalchemy/orm/_orm_constructors.py
+++ b/lib/sqlalchemy/orm/_orm_constructors.py
@@ -62,6 +62,7 @@ if TYPE_CHECKING:
     from .relationships import _RelationshipJoinConditionArgument
     from .relationships import ORMBackrefArgument
     from .session import _SessionBind
+    from ..sql._typing import _AutoIncrementType
     from ..sql._typing import _ColumnExpressionArgument
     from ..sql._typing import _FromClauseArgument
     from ..sql._typing import _InfoType
@@ -116,7 +117,7 @@ def mapped_column(
     use_existing_column: bool = False,
     name: Optional[str] = None,
     type_: Optional[_TypeEngineArgument[Any]] = None,
-    autoincrement: Union[bool, Literal["auto", "ignore_fk"]] = "auto",
+    autoincrement: _AutoIncrementType = "auto",
     doc: Optional[str] = None,
     key: Optional[str] = None,
     index: Optional[bool] = None,
@@ -129,7 +130,8 @@ def mapped_column(
     quote: Optional[bool] = None,
     system: bool = False,
     comment: Optional[str] = None,
-    **dialect_kwargs: Any,
+    sort_order: int = 0,
+    **kw: Any,
 ) -> MappedColumn[Any]:
     r"""declare a new ORM-mapped :class:`_schema.Column` construct
     for use within :ref:`Declarative Table <orm_declarative_table>`
@@ -248,6 +250,15 @@ def mapped_column(
      :paramref:`_orm.mapped_column.default` will always apply to the
      constructor default for a dataclasses mapping.
 
+    :param sort_order: An integer that indicates how this mapped column
+     should be sorted compared to the others when the ORM is creating a
+     :class:`_schema.Table`. Among mapped columns that have the same
+     value the default ordering is used, placing first the mapped columns
+     defined in the main class, then the ones in the super classes.
+     Defaults to 0. The sort is ascending.
+
+     .. versionadded:: 2.0.4
+
     :param init: Specific to :ref:`orm_declarative_native_dataclasses`,
      specifies if the mapped attribute should be part of the ``__init__()``
      method as generated by the dataclass process.
@@ -270,7 +281,7 @@ def mapped_column(
      :ref:`orm_declarative_native_dataclasses`, indicates if this field
      should be marked as keyword-only when generating the ``__init__()``.
 
-    :param \**kw: All remaining keyword argments are passed through to the
+    :param \**kw: All remaining keyword arguments are passed through to the
      constructor for the :class:`_schema.Column`.
 
     """
@@ -303,7 +314,8 @@ def mapped_column(
         deferred=deferred,
         deferred_group=deferred_group,
         deferred_raiseload=deferred_raiseload,
-        **dialect_kwargs,
+        sort_order=sort_order,
+        **kw,
     )
 
 
@@ -2300,7 +2312,7 @@ def join(
 
     :func:`_orm.join` is an extension to the core join interface
     provided by :func:`_expression.join()`, where the
-    left and right selectables may be not only core selectable
+    left and right selectable may be not only core selectable
     objects such as :class:`_schema.Table`, but also mapped classes or
     :class:`.AliasedClass` instances.   The "on" clause can
     be a SQL expression or an ORM mapped attribute
diff --git a/lib/sqlalchemy/orm/decl_base.py b/lib/sqlalchemy/orm/decl_base.py
index f0be55b89..29d748596 100644
--- a/lib/sqlalchemy/orm/decl_base.py
+++ b/lib/sqlalchemy/orm/decl_base.py
@@ -211,13 +211,13 @@ def _get_immediate_cls_attr(
         return getattr(cls, attrname)
 
     for base in cls.__mro__[1:]:
-        _is_classicial_inherits = _dive_for_cls_manager(base) is not None
+        _is_classical_inherits = _dive_for_cls_manager(base) is not None
 
         if attrname in base.__dict__ and (
             base is cls
             or (
                 (base in cls.__bases__ if strict else True)
-                and not _is_classicial_inherits
+                and not _is_classical_inherits
             )
         ):
             return getattr(base, attrname)
@@ -451,6 +451,7 @@ class _ClassScanMapperConfig(_MapperConfig):
         "local_table",
         "persist_selectable",
         "declared_columns",
+        "column_ordering",
         "column_copies",
         "table_args",
         "tablename",
@@ -471,6 +472,7 @@ class _ClassScanMapperConfig(_MapperConfig):
     local_table: Optional[FromClause]
     persist_selectable: Optional[FromClause]
     declared_columns: util.OrderedSet[Column[Any]]
+    column_ordering: Dict[Column[Any], int]
     column_copies: Dict[
         Union[MappedColumn[Any], Column[Any]],
         Union[MappedColumn[Any], Column[Any]],
@@ -522,6 +524,7 @@ class _ClassScanMapperConfig(_MapperConfig):
         self.collected_attributes = {}
         self.collected_annotations = {}
         self.declared_columns = util.OrderedSet()
+        self.column_ordering = {}
         self.column_copies = {}
 
         self.dataclass_setup_arguments = dca = getattr(
@@ -1557,6 +1560,7 @@ class _ClassScanMapperConfig(_MapperConfig):
 
         # extract columns from the class dict
         declared_columns = self.declared_columns
+        column_ordering = self.column_ordering
         name_to_prop_key = collections.defaultdict(set)
 
         for key, c in list(our_stuff.items()):
@@ -1570,10 +1574,12 @@ class _ClassScanMapperConfig(_MapperConfig):
                     # this is a MappedColumn that will produce a Column for us
                     del our_stuff[key]
 
-                for col in c.columns_to_assign:
+                for col, sort_order in c.columns_to_assign:
                     if not isinstance(c, CompositeProperty):
                         name_to_prop_key[col.name].add(key)
                     declared_columns.add(col)
+                    assert col not in column_ordering
+                    column_ordering[col] = sort_order
 
                     # if this is a MappedColumn and the attribute key we
                     # have is not what the column has for its key, map the
@@ -1613,6 +1619,7 @@ class _ClassScanMapperConfig(_MapperConfig):
         table_args = self.table_args
         clsdict_view = self.clsdict_view
         declared_columns = self.declared_columns
+        column_ordering = self.column_ordering
 
         manager = attributes.manager_of_class(cls)
 
@@ -1647,12 +1654,17 @@ class _ClassScanMapperConfig(_MapperConfig):
                 if autoload:
                     table_kw["autoload"] = True
 
+                sorted_columns = sorted(
+                    declared_columns,
+                    key=lambda c: column_ordering.get(c, 0),
+                )
                 table = self.set_cls_attribute(
                     "__table__",
                     table_cls(
                         tablename,
                         self._metadata_for_cls(manager),
-                        *(tuple(declared_columns) + tuple(args)),
+                        *sorted_columns,
+                        *args,
                         **table_kw,
                     ),
                 )
@@ -1998,7 +2010,7 @@ def _add_attribute(
             mapped_cls.__mapper__.add_property(key, value)
         elif isinstance(value, _MapsColumns):
             mp = value.mapper_property_to_assign
-            for col in value.columns_to_assign:
+            for col, _ in value.columns_to_assign:
                 _undefer_column_name(key, col)
                 _table_or_raise(mapped_cls).append_column(
                     col, replace_existing=True
diff --git a/lib/sqlalchemy/orm/descriptor_props.py b/lib/sqlalchemy/orm/descriptor_props.py
index 3bab78123..b65171c9d 100644
--- a/lib/sqlalchemy/orm/descriptor_props.py
+++ b/lib/sqlalchemy/orm/descriptor_props.py
@@ -502,8 +502,8 @@ class CompositeProperty(
         return self
 
     @property
-    def columns_to_assign(self) -> List[schema.Column[Any]]:
-        return [c for c in self.columns if c.table is None]
+    def columns_to_assign(self) -> List[Tuple[schema.Column[Any], int]]:
+        return [(c, 0) for c in self.columns if c.table is None]
 
     @util.preload_module("orm.properties")
     def _setup_arguments_on_columns(self) -> None:
diff --git a/lib/sqlalchemy/orm/interfaces.py b/lib/sqlalchemy/orm/interfaces.py
index 4a8b72441..866749139 100644
--- a/lib/sqlalchemy/orm/interfaces.py
+++ b/lib/sqlalchemy/orm/interfaces.py
@@ -306,7 +306,7 @@ class _MapsColumns(_DCAttributeOptions, _MappedAttribute[_T]):
         raise NotImplementedError()
 
     @property
-    def columns_to_assign(self) -> List[Column[_T]]:
+    def columns_to_assign(self) -> List[Tuple[Column[_T], int]]:
         """A list of Column objects that should be declaratively added to the
         new Table object.
 
diff --git a/lib/sqlalchemy/orm/properties.py b/lib/sqlalchemy/orm/properties.py
index 60b1611ac..e736e4fd2 100644
--- a/lib/sqlalchemy/orm/properties.py
+++ b/lib/sqlalchemy/orm/properties.py
@@ -21,6 +21,7 @@ from typing import List
 from typing import Optional
 from typing import Sequence
 from typing import Set
+from typing import Tuple
 from typing import Type
 from typing import TYPE_CHECKING
 from typing import TypeVar
@@ -214,10 +215,10 @@ class ColumnProperty(
         return self
 
     @property
-    def columns_to_assign(self) -> List[Column[Any]]:
+    def columns_to_assign(self) -> List[Tuple[Column[Any], int]]:
         # mypy doesn't care about the isinstance here
         return [
-            c  # type: ignore
+            (c, 0)  # type: ignore
             for c in self.columns
             if isinstance(c, Column) and c.table is None
         ]
@@ -524,6 +525,7 @@ class MappedColumn(
     __slots__ = (
         "column",
         "_creation_order",
+        "_sort_order",
         "foreign_keys",
         "_has_nullable",
         "_has_insert_default",
@@ -578,6 +580,7 @@ class MappedColumn(
                 self.deferred_group or self.deferred_raiseload
             )
 
+        self._sort_order = kw.pop("sort_order", 0)
         self.column = cast("Column[_T]", Column(*arg, **kw))
         self.foreign_keys = self.column.foreign_keys
         self._has_nullable = "nullable" in kw and kw.get("nullable") not in (
@@ -597,6 +600,7 @@ class MappedColumn(
         new._has_insert_default = self._has_insert_default
         new._has_dataclass_arguments = self._has_dataclass_arguments
         new._use_existing_column = self._use_existing_column
+        new._sort_order = self._sort_order
         util.set_creation_order(new)
         return new
 
@@ -618,8 +622,8 @@ class MappedColumn(
             return None
 
     @property
-    def columns_to_assign(self) -> List[Column[Any]]:
-        return [self.column]
+    def columns_to_assign(self) -> List[Tuple[Column[Any], int]]:
+        return [(self.column, self._sort_order)]
 
     def __clause_element__(self) -> Column[_T]:
         return self.column
diff --git a/lib/sqlalchemy/sql/_typing.py b/lib/sqlalchemy/sql/_typing.py
index ab124103f..6bf9a5a1f 100644
--- a/lib/sqlalchemy/sql/_typing.py
+++ b/lib/sqlalchemy/sql/_typing.py
@@ -264,6 +264,8 @@ _EquivalentColumnMap = Dict["ColumnElement[Any]", Set["ColumnElement[Any]"]]
 
 _LimitOffsetType = Union[int, _ColumnExpressionArgument[int], None]
 
+_AutoIncrementType = Union[bool, Literal["auto", "ignore_fk"]]
+
 if TYPE_CHECKING:
 
     def is_sql_compiler(c: Compiled) -> TypeGuard[SQLCompiler]:
diff --git a/lib/sqlalchemy/sql/schema.py b/lib/sqlalchemy/sql/schema.py
index 2a713fea6..20c0341ad 100644
--- a/lib/sqlalchemy/sql/schema.py
+++ b/lib/sqlalchemy/sql/schema.py
@@ -84,6 +84,7 @@ from ..util.typing import Self
 from ..util.typing import TypeGuard
 
 if typing.TYPE_CHECKING:
+    from ._typing import _AutoIncrementType
     from ._typing import _DDLColumnArgument
     from ._typing import _InfoType
     from ._typing import _TextCoercedExpressionArgument
@@ -1375,7 +1376,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause[_T]):
         *args: SchemaEventTarget,
         name: Optional[str] = None,
         type_: Optional[_TypeEngineArgument[_T]] = None,
-        autoincrement: Union[bool, Literal["auto", "ignore_fk"]] = "auto",
+        autoincrement: _AutoIncrementType = "auto",
         default: Optional[Any] = None,
         doc: Optional[str] = None,
         key: Optional[str] = None,
@@ -1949,7 +1950,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause[_T]):
 
         self.system = system
         self.doc = doc
-        self.autoincrement = autoincrement
+        self.autoincrement: _AutoIncrementType = autoincrement
         self.constraints = set()
         self.foreign_keys = set()
         self.comment = comment
@@ -2278,7 +2279,7 @@ class Column(DialectKWArgs, SchemaItem, ColumnClause[_T]):
     def _copy(self, **kw: Any) -> Column[Any]:
         """Create a copy of this ``Column``, uninitialized.
 
-        This is used in :meth:`_schema.Table.to_metadata`.
+        This is used in :meth:`_schema.Table.to_metadata` and by the ORM.
 
         """