Improved support for expression indexes

Added support for autogenerate comparison of indexes on PostgreSQL which
include SQL expressions; the previous warning that such indexes were
skipped is now removed. This functionality requires SQLAlchemy 2.0.
For older SQLAlchemy versions, these indexes are still skipped.

Fixed issue where indexes on SQLite which include SQL expressions would not
compare against themselves correctly, generating false positives.
SQLAlchemy as of version 2 has no support for reflecting expression based
indexes on SQLite; so for now, the behavior is that SQLite expression-based
indexes are ignored for autogenerate compare, in the same way that
PostgreSQL expression-based indexes were ignored for the time that
SQLAlchemy did not support reflection of such indexes (which is now
supported in SQLAlchemy 2.0 as well as this release of Alembic).

Fixed issue in index detection where autogenerate change detection would
consider indexes with the same columns but with different order as equal,
while in general they are not equivalent in how a database will use them.

Fixes: #1165
Fixes: #1166
Change-Id: I226408eed855b923172e5df0bdab005ed2cc9f53

17 files changed, 521 insertions, 104 deletions

diff --git a/alembic/autogenerate/compare.py b/alembic/autogenerate/compare.py
index 828a4cd..d97eb9c 100644
--- a/alembic/autogenerate/compare.py
+++ b/alembic/autogenerate/compare.py
@@ -17,6 +17,7 @@ from typing import Union
 from sqlalchemy import event
 from sqlalchemy import inspect
 from sqlalchemy import schema as sa_schema
+from sqlalchemy import text
 from sqlalchemy import types as sqltypes
 from sqlalchemy.util import OrderedSet
 
@@ -30,6 +31,7 @@ if TYPE_CHECKING:
 
     from sqlalchemy.engine.reflection import Inspector
     from sqlalchemy.sql.elements import quoted_name
+    from sqlalchemy.sql.elements import TextClause
     from sqlalchemy.sql.schema import Column
     from sqlalchemy.sql.schema import ForeignKeyConstraint
     from sqlalchemy.sql.schema import Index
@@ -37,6 +39,7 @@ if TYPE_CHECKING:
     from sqlalchemy.sql.schema import UniqueConstraint
 
     from alembic.autogenerate.api import AutogenContext
+    from alembic.ddl.impl import DefaultImpl
     from alembic.operations.ops import AlterColumnOp
     from alembic.operations.ops import MigrationScript
     from alembic.operations.ops import ModifyTableOps
@@ -276,14 +279,12 @@ def _compare_tables(
 
 
 def _make_index(params: Dict[str, Any], conn_table: Table) -> Optional[Index]:
-    exprs = []
-    for col_name in params["column_names"]:
+    exprs: list[Union[Column[Any], TextClause]] = []
+    for num, col_name in enumerate(params["column_names"]):
+        item: Union[Column[Any], TextClause]
         if col_name is None:
-            util.warn(
-                "Skipping reflected expression-based "
-                f"index {params['name']!r}"
-            )
-            return None
+            assert "expressions" in params
+            item = text(params["expressions"][num])
         else:
             item = conn_table.c[col_name]
         exprs.append(item)
@@ -439,10 +440,10 @@ class _uq_constraint_sig(_constraint_sig):
 class _ix_constraint_sig(_constraint_sig):
     is_index = True
 
-    def __init__(self, const: Index) -> None:
+    def __init__(self, const: Index, impl: DefaultImpl) -> None:
         self.const = const
         self.name = const.name
-        self.sig = tuple(sorted([col.name for col in const.columns]))
+        self.sig = impl.create_index_sig(const)
         self.is_unique = bool(const.unique)
 
     def md_name_to_sql_name(self, context: AutogenContext) -> Optional[str]:
@@ -624,11 +625,14 @@ def _compare_indexes_and_uniques(
         _uq_constraint_sig(uq) for uq in metadata_unique_constraints
     }
 
-    metadata_indexes_sig = {_ix_constraint_sig(ix) for ix in metadata_indexes}
+    impl = autogen_context.migration_context.impl
+    metadata_indexes_sig = {
+        _ix_constraint_sig(ix, impl) for ix in metadata_indexes
+    }
 
     conn_unique_constraints = {_uq_constraint_sig(uq) for uq in conn_uniques}
 
-    conn_indexes_sig = {_ix_constraint_sig(ix) for ix in conn_indexes}
+    conn_indexes_sig = {_ix_constraint_sig(ix, impl) for ix in conn_indexes}
 
     # 5. index things by name, for those objects that have names
     metadata_names = {
@@ -816,7 +820,7 @@ def _compare_indexes_and_uniques(
                 )
             if conn_obj.sig != metadata_obj.sig:
                 msg.append(
-                    " columns %r to %r" % (conn_obj.sig, metadata_obj.sig)
+                    " expression %r to %r" % (conn_obj.sig, metadata_obj.sig)
                 )
 
             if msg:
diff --git a/alembic/ddl/impl.py b/alembic/ddl/impl.py
index 728d1da..f11d1ed 100644
--- a/alembic/ddl/impl.py
+++ b/alembic/ddl/impl.py
@@ -665,6 +665,27 @@ class DefaultImpl(metaclass=ImplMeta):
             bool(diff) or bool(metadata_identity) != bool(inspector_identity),
         )
 
+    def create_index_sig(self, index: Index) -> Tuple[Any, ...]:
+        # order of col matters in an index
+        return tuple(col.name for col in index.columns)
+
+    def _skip_functional_indexes(self, metadata_indexes, conn_indexes):
+        conn_indexes_by_name = {c.name: c for c in conn_indexes}
+
+        for idx in list(metadata_indexes):
+            if idx.name in conn_indexes_by_name:
+                continue
+            iex = sqla_compat.is_expression_index(idx)
+            if iex:
+                util.warn(
+                    "autogenerate skipping metadata-specified "
+                    "expression-based index "
+                    f"{idx.name!r}; dialect {self.__dialect__!r} under "
+                    f"SQLAlchemy {sqla_compat.sqlalchemy_version} can't "
+                    "reflect these indexes so they can't be compared"
+                )
+                metadata_indexes.discard(idx)
+
 
 def _compare_identity_options(
     attributes, metadata_io, inspector_io, default_io
diff --git a/alembic/ddl/postgresql.py b/alembic/ddl/postgresql.py
index 32674d2..e7b0a48 100644
--- a/alembic/ddl/postgresql.py
+++ b/alembic/ddl/postgresql.py
@@ -12,6 +12,7 @@ from typing import TYPE_CHECKING
 from typing import Union
 
 from sqlalchemy import Column
+from sqlalchemy import Index
 from sqlalchemy import literal_column
 from sqlalchemy import Numeric
 from sqlalchemy import text
@@ -21,7 +22,6 @@ from sqlalchemy.dialects.postgresql import ExcludeConstraint
 from sqlalchemy.dialects.postgresql import INTEGER
 from sqlalchemy.schema import CreateIndex
 from sqlalchemy.sql.elements import ColumnClause
-from sqlalchemy.sql.elements import UnaryExpression
 from sqlalchemy.types import NULLTYPE
 
 from .base import alter_column
@@ -235,8 +235,6 @@ class PostgresqlImpl(DefaultImpl):
         metadata_indexes,
     ):
 
-        conn_indexes_by_name = {c.name: c for c in conn_indexes}
-
         doubled_constraints = {
             index
             for index in conn_indexes
@@ -246,23 +244,41 @@ class PostgresqlImpl(DefaultImpl):
         for ix in doubled_constraints:
             conn_indexes.remove(ix)
 
-        for idx in list(metadata_indexes):
-            if idx.name in conn_indexes_by_name:
-                continue
-            exprs = idx.expressions
-            for expr in exprs:
-                while isinstance(expr, UnaryExpression):
-                    expr = expr.element
-                if not isinstance(expr, Column):
-                    if sqla_compat.sqla_2:
-                        msg = ""
-                    else:
-                        msg = "; not supported by SQLAlchemy reflection"
-                    util.warn(
-                        "autogenerate skipping functional index "
-                        f"{idx.name!r}{msg}"
-                    )
-                    metadata_indexes.discard(idx)
+        if not sqla_compat.sqla_2:
+            self._skip_functional_indexes(metadata_indexes, conn_indexes)
+
+    def _cleanup_index_expr(self, index: Index, expr: str) -> str:
+        # start = expr
+        expr = expr.lower()
+        expr = expr.replace('"', "")
+        if index.table is not None:
+            expr = expr.replace(f"{index.table.name.lower()}.", "")
+
+        while expr and expr[0] == "(" and expr[-1] == ")":
+            expr = expr[1:-1]
+        if "::" in expr:
+            # strip :: cast. types can have spaces in them
+            expr = re.sub(r"(::[\w ]+\w)", "", expr)
+
+        # print(f"START: {start} END: {expr}")
+        return expr
+
+    def create_index_sig(self, index: Index) -> Tuple[Any, ...]:
+        if sqla_compat.is_expression_index(index):
+            return tuple(
+                self._cleanup_index_expr(
+                    index,
+                    e
+                    if isinstance(e, str)
+                    else e.compile(
+                        dialect=self.dialect,
+                        compile_kwargs={"literal_binds": True},
+                    ).string,
+                )
+                for e in index.expressions
+            )
+        else:
+            return super().create_index_sig(index)
 
     def render_type(
         self, type_: TypeEngine, autogen_context: AutogenContext
diff --git a/alembic/ddl/sqlite.py b/alembic/ddl/sqlite.py
index 6b8c293..302a877 100644
--- a/alembic/ddl/sqlite.py
+++ b/alembic/ddl/sqlite.py
@@ -186,6 +186,16 @@ class SQLiteImpl(DefaultImpl):
                 existing_transfer["expr"], new_type
             )
 
+    def correct_for_autogen_constraints(
+        self,
+        conn_unique_constraints,
+        conn_indexes,
+        metadata_unique_constraints,
+        metadata_indexes,
+    ):
+
+        self._skip_functional_indexes(metadata_indexes, conn_indexes)
+
 
 @compiles(RenameTable, "sqlite")
 def visit_rename_table(
diff --git a/alembic/testing/requirements.py b/alembic/testing/requirements.py
index 6896276..989c7cd 100644
--- a/alembic/testing/requirements.py
+++ b/alembic/testing/requirements.py
@@ -96,6 +96,13 @@ class SuiteRequirements(Requirements):
         )
 
     @property
+    def sqlalchemy_2(self):
+        return exclusions.skip_if(
+            lambda config: not util.sqla_2,
+            "SQLAlchemy 2.x test",
+        )
+
+    @property
     def comments(self):
         return exclusions.only_if(
             lambda config: config.db.dialect.supports_comments
diff --git a/alembic/testing/schemacompare.py b/alembic/testing/schemacompare.py
index 4409421..c063499 100644
--- a/alembic/testing/schemacompare.py
+++ b/alembic/testing/schemacompare.py
@@ -43,15 +43,19 @@ class CompareColumn:
 
 
 class CompareIndex:
-    def __init__(self, index):
+    def __init__(self, index, name_only=False):
         self.index = index
+        self.name_only = name_only
 
     def __eq__(self, other):
-        return (
-            str(schema.CreateIndex(self.index))
-            == str(schema.CreateIndex(other))
-            and self.index.dialect_kwargs == other.dialect_kwargs
-        )
+        if self.name_only:
+            return self.index.name == other.name
+        else:
+            return (
+                str(schema.CreateIndex(self.index))
+                == str(schema.CreateIndex(other))
+                and self.index.dialect_kwargs == other.dialect_kwargs
+            )
 
     def __ne__(self, other):
         return not self.__eq__(other)
diff --git a/alembic/testing/util.py b/alembic/testing/util.py
index a82690d..e65597d 100644
--- a/alembic/testing/util.py
+++ b/alembic/testing/util.py
@@ -10,6 +10,8 @@ import re
 import types
 from typing import Union
 
+from sqlalchemy.util import inspect_getfullargspec
+
 
 def flag_combinations(*combinations):
     """A facade around @testing.combinations() oriented towards boolean
@@ -69,10 +71,12 @@ def resolve_lambda(__fn, **kw):
 
     """
 
+    pos_args = inspect_getfullargspec(__fn)[0]
+    pass_pos_args = {arg: kw.pop(arg) for arg in pos_args}
     glb = dict(__fn.__globals__)
     glb.update(kw)
     new_fn = types.FunctionType(__fn.__code__, glb)
-    return new_fn()
+    return new_fn(**pass_pos_args)
 
 
 def metadata_fixture(ddl="function"):
diff --git a/alembic/util/__init__.py b/alembic/util/__init__.py
index 4374f46..c81d431 100644
--- a/alembic/util/__init__.py
+++ b/alembic/util/__init__.py
@@ -28,6 +28,7 @@ from .sqla_compat import has_computed
 from .sqla_compat import sqla_13
 from .sqla_compat import sqla_14
 from .sqla_compat import sqla_1x
+from .sqla_compat import sqla_2
 
 
 if not sqla_13:
diff --git a/alembic/util/sqla_compat.py b/alembic/util/sqla_compat.py
index 23255be..9534c99 100644
--- a/alembic/util/sqla_compat.py
+++ b/alembic/util/sqla_compat.py
@@ -23,8 +23,10 @@ from sqlalchemy.schema import Column
 from sqlalchemy.schema import ForeignKeyConstraint
 from sqlalchemy.sql import visitors
 from sqlalchemy.sql.elements import BindParameter
+from sqlalchemy.sql.elements import ColumnClause
 from sqlalchemy.sql.elements import quoted_name
 from sqlalchemy.sql.elements import TextClause
+from sqlalchemy.sql.elements import UnaryExpression
 from sqlalchemy.sql.visitors import traverse
 
 if TYPE_CHECKING:
@@ -37,7 +39,6 @@ if TYPE_CHECKING:
     from sqlalchemy.sql.base import ColumnCollection
     from sqlalchemy.sql.compiler import SQLCompiler
     from sqlalchemy.sql.dml import Insert
-    from sqlalchemy.sql.elements import ColumnClause
     from sqlalchemy.sql.elements import ColumnElement
     from sqlalchemy.sql.schema import Constraint
     from sqlalchemy.sql.schema import SchemaItem
@@ -60,7 +61,8 @@ _vers = tuple(
 sqla_13 = _vers >= (1, 3)
 sqla_14 = _vers >= (1, 4)
 sqla_14_26 = _vers >= (1, 4, 26)
-sqla_2 = _vers >= (1, 5)
+sqla_2 = _vers >= (2,)
+sqlalchemy_version = __version__
 
 
 if sqla_14:
@@ -556,3 +558,13 @@ else:
 
     def _select(*columns, **kw) -> Select:  # type: ignore[no-redef]
         return sql.select(list(columns), **kw)  # type: ignore[call-overload]
+
+
+def is_expression_index(index: Index) -> bool:
+    expr: Any
+    for expr in index.expressions:
+        while isinstance(expr, UnaryExpression):
+            expr = expr.element
+        if not isinstance(expr, ColumnClause) or expr.is_literal:
+            return True
+    return False
diff --git a/docs/build/unreleased/1165.rst b/docs/build/unreleased/1165.rst
new file mode 100644
index 0000000..c2e3c9a
--- /dev/null
+++ b/docs/build/unreleased/1165.rst
@@ -0,0 +1,12 @@
+.. change::
+    :tags: bug, autogenerate, sqlite
+    :tickets: 1165
+
+    Fixed issue where indexes on SQLite which include SQL expressions would not
+    compare against themselves correctly, generating false positives.
+    SQLAlchemy as of version 2 has no support for reflecting expression based
+    indexes on SQLite; so for now, the behavior is that SQLite expression-based
+    indexes are ignored for autogenerate compare, in the same way that
+    PostgreSQL expression-based indexes were ignored for the time that
+    SQLAlchemy did not support reflection of such indexes (which is now
+    supported in SQLAlchemy 2.0 as well as this release of Alembic).
diff --git a/docs/build/unreleased/1166.rst b/docs/build/unreleased/1166.rst
new file mode 100644
index 0000000..c9b64b9
--- /dev/null
+++ b/docs/build/unreleased/1166.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: bug, autogenerate
+    :tickets: 1166
+
+    Fixed issue in index detection where autogenerate change detection would
+    consider indexes with the same columns but with different order as equal,
+    while in general they are not equivalent in how a database will use them.
diff --git a/docs/build/unreleased/index_updates.rst b/docs/build/unreleased/index_updates.rst
new file mode 100644
index 0000000..a468c73
--- /dev/null
+++ b/docs/build/unreleased/index_updates.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: usecase, autogenerate, postgresql
+
+    Added support for autogenerate comparison of indexes on PostgreSQL which
+    include SQL expressions; the previous warning that such indexes were
+    skipped is now removed. This functionality requires SQLAlchemy 2.0.
+    For older SQLAlchemy versions, these indexes are still skipped.
diff --git a/setup.cfg b/setup.cfg
index 0d9ce1a..cc73baa 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -114,7 +114,7 @@ oracle8=oracle://scott:tiger@127.0.0.1:1521/?use_ansi=0
 
 
 [tool:pytest]
-addopts= --tb native -v -r sfxX -p no:warnings -p no:logging --maxfail=25
+addopts= --tb native -v -r sfxX -p no:warnings -p no:logging --maxfail=100
 python_files=tests/test_*.py
 markers =
     backend: tests that should run on all backends; typically dialect-sensitive
diff --git a/tests/requirements.py b/tests/requirements.py
index c774e67..1a100dd 100644
--- a/tests/requirements.py
+++ b/tests/requirements.py
@@ -410,3 +410,17 @@ class DefaultRequirements(SuiteRequirements):
         )
 
         return imports + version + sqlalchemy
+
+    @property
+    def reflect_indexes_with_expressions(self):
+        sqlalchemy = exclusions.only_if(
+            lambda _: sqla_compat.sqla_2, "sqlalchemy 2 is required"
+        )
+
+        postgresql = exclusions.only_on(["postgresql"])
+
+        return sqlalchemy + postgresql
+
+    @property
+    def indexes_with_expressions(self):
+        return exclusions.only_on(["postgresql", "sqlite>=3.9.0"])
diff --git a/tests/test_autogen_indexes.py b/tests/test_autogen_indexes.py
index 68a6bd6..6faba04 100644
--- a/tests/test_autogen_indexes.py
+++ b/tests/test_autogen_indexes.py
@@ -1,6 +1,11 @@
+from contextlib import nullcontext
+import itertools
+
 from sqlalchemy import Column
+from sqlalchemy import Float
 from sqlalchemy import ForeignKey
 from sqlalchemy import ForeignKeyConstraint
+from sqlalchemy import func
 from sqlalchemy import Index
 from sqlalchemy import Integer
 from sqlalchemy import MetaData
@@ -9,16 +14,20 @@ from sqlalchemy import PrimaryKeyConstraint
 from sqlalchemy import String
 from sqlalchemy import Table
 from sqlalchemy import UniqueConstraint
+from sqlalchemy.dialects.postgresql import DOUBLE_PRECISION
 from sqlalchemy.sql.expression import column
 from sqlalchemy.sql.expression import desc
 
+from alembic import testing
 from alembic.testing import combinations
 from alembic.testing import config
 from alembic.testing import eq_
 from alembic.testing import exclusions
+from alembic.testing import resolve_lambda
 from alembic.testing import schemacompare
 from alembic.testing import TestBase
 from alembic.testing import util
+from alembic.testing.assertions import expect_warnings
 from alembic.testing.env import staging_env
 from alembic.testing.suite._autogen_fixtures import AutogenFixtureTest
 from alembic.util import sqla_compat
@@ -1175,6 +1184,287 @@ class AutogenerateIndexTest(AutogenFixtureTest, TestBase):
         diffs = self._fixture(m1, m2)
         eq_(diffs, [])
 
+    def test_column_order_changed(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        old = Index("SomeIndex", "x", "y")
+        Table(
+            "order_change",
+            m1,
+            Column("id", Integer, primary_key=True),
+            Column("x", Integer),
+            Column("y", Integer),
+            old,
+        )
+
+        new = Index("SomeIndex", "y", "x")
+        Table(
+            "order_change",
+            m2,
+            Column("id", Integer, primary_key=True),
+            Column("x", Integer),
+            Column("y", Integer),
+            new,
+        )
+        diffs = self._fixture(m1, m2)
+        eq_(
+            diffs,
+            [
+                ("remove_index", schemacompare.CompareIndex(old)),
+                ("add_index", schemacompare.CompareIndex(new)),
+            ],
+        )
+
+
+class AutogenerateExpressionIndexTest(AutogenFixtureTest, TestBase):
+    """tests involving indexes with expression"""
+
+    __requires__ = ("indexes_with_expressions",)
+
+    __backend__ = True
+
+    @property
+    def has_reflection(self):
+        return config.requirements.reflect_indexes_with_expressions.enabled
+
+    def test_expression_indexes_add(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        Table(
+            "exp_index",
+            m1,
+            Column("id", Integer, primary_key=True),
+            Column("x", Integer),
+            Column("y", Integer),
+        )
+
+        idx = Index("SomeIndex", "y", func.lower(column("x")))  # noqa
+        Table(
+            "exp_index",
+            m2,
+            Column("id", Integer, primary_key=True),
+            Column("x", Integer),
+            Column("y", Integer),
+            idx,
+        )
+
+        if self.has_reflection:
+            diffs = self._fixture(m1, m2)
+            eq_(diffs, [("add_index", schemacompare.CompareIndex(idx))])
+        else:
+            with expect_warnings(
+                r"autogenerate skipping metadata-specified expression-based "
+                r"index 'SomeIndex'; dialect '.*' under SQLAlchemy .* "
+                r"can't reflect these "
+                r"indexes so they can't be compared",
+            ):
+                diffs = self._fixture(m1, m2)
+            eq_(diffs, [])
+
+    def _lots_of_indexes(flatten: bool = False):
+        diff_pairs = [
+            (
+                lambda t: Index("SomeIndex", "y", func.lower(t.c.x)),
+                lambda t: Index("SomeIndex", func.lower(t.c.x)),
+            ),
+            (
+                lambda CapT: Index("SomeIndex", "y", func.lower(CapT.c.XCol)),
+                lambda CapT: Index("SomeIndex", func.lower(CapT.c.XCol)),
+            ),
+            (
+                lambda t: Index(
+                    "SomeIndex", "y", func.lower(column("x")), _table=t
+                ),
+                lambda t: Index(
+                    "SomeIndex", func.lower(column("x")), _table=t
+                ),
+            ),
+            (
+                lambda t: Index("SomeIndex", t.c.y, func.lower(t.c.x)),
+                lambda t: Index("SomeIndex", func.lower(t.c.x), t.c.y),
+            ),
+            (
+                lambda t: Index("SomeIndex", t.c.y, func.lower(t.c.x)),
+                lambda t: Index("SomeIndex", t.c.y, func.lower(t.c.q)),
+            ),
+            (
+                lambda t: Index("SomeIndex", t.c.z, func.lower(t.c.x)),
+                lambda t: Index("SomeIndex", t.c.y, func.lower(t.c.x)),
+            ),
+            (
+                lambda t: Index("SomeIndex", func.lower(t.c.x)),
+                lambda t: Index("SomeIndex", t.c.y, func.lower(t.c.x)),
+            ),
+            (
+                lambda t: Index("SomeIndex", t.c.y, func.upper(t.c.x)),
+                lambda t: Index("SomeIndex", t.c.y, func.lower(t.c.x)),
+            ),
+            (
+                lambda t: Index("SomeIndex", t.c.y, t.c.ff + 1),
+                lambda t: Index("SomeIndex", t.c.y, t.c.ff + 3),
+            ),
+            (
+                lambda t: Index("SomeIndex", t.c.y, func.ceil(t.c.ff)),
+                lambda t: Index("SomeIndex", t.c.y, func.floor(t.c.ff)),
+            ),
+            (
+                lambda t: Index("SomeIndex", t.c.y, func.lower(t.c.x)),
+                lambda t: Index("SomeIndex", t.c.y, func.lower(t.c.x + t.c.q)),
+            ),
+            (
+                lambda t: Index("SomeIndex", t.c.y, t.c.z + 3),
+                lambda t: Index("SomeIndex", t.c.y, t.c.z * 3),
+            ),
+            (
+                lambda t: Index("SomeIndex", func.lower(t.c.x), t.c.q + "42"),
+                lambda t: Index("SomeIndex", func.lower(t.c.q), t.c.x + "42"),
+            ),
+            (
+                lambda t: Index("SomeIndex", func.lower(t.c.x), t.c.z + 42),
+                lambda t: Index("SomeIndex", t.c.z + 42, func.lower(t.c.q)),
+            ),
+            (
+                lambda t: Index("SomeIndex", t.c.ff + 42),
+                lambda t: Index("SomeIndex", 42 + t.c.ff),
+            ),
+        ]
+        if flatten:
+            return list(itertools.chain.from_iterable(diff_pairs))
+        else:
+            return diff_pairs
+
+    @testing.fixture
+    def index_changed_tables(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        t_old = Table(
+            "exp_index",
+            m1,
+            Column("id", Integer, primary_key=True),
+            Column("x", String(100)),
+            Column("y", String(100)),
+            Column("q", String(100)),
+            Column("z", Integer),
+            Column("ff", Float().with_variant(DOUBLE_PRECISION, "postgresql")),
+        )
+
+        t_new = Table(
+            "exp_index",
+            m2,
+            Column("id", Integer, primary_key=True),
+            Column("x", String(100)),
+            Column("y", String(100)),
+            Column("q", String(100)),
+            Column("z", Integer),
+            Column("ff", Float().with_variant(DOUBLE_PRECISION, "postgresql")),
+        )
+
+        CapT_old = Table(
+            "CapT table",
+            m1,
+            Column("id", Integer, primary_key=True),
+            Column("XCol", String(100)),
+            Column("y", String(100)),
+        )
+
+        CapT_new = Table(
+            "CapT table",
+            m2,
+            Column("id", Integer, primary_key=True),
+            Column("XCol", String(100)),
+            Column("y", String(100)),
+        )
+
+        return (
+            m1,
+            m2,
+            {"t": t_old, "CapT": CapT_old},
+            {"t": t_new, "CapT": CapT_new},
+        )
+
+    @combinations(*_lots_of_indexes(), argnames="old_fn, new_fn")
+    def test_expression_indexes_changed(
+        self, index_changed_tables, old_fn, new_fn
+    ):
+        m1, m2, old_fixture_tables, new_fixture_tables = index_changed_tables
+
+        old, new = resolve_lambda(
+            old_fn, **old_fixture_tables
+        ), resolve_lambda(new_fn, **new_fixture_tables)
+
+        if self.has_reflection:
+            diffs = self._fixture(m1, m2)
+            eq_(
+                diffs,
+                [
+                    ("remove_index", schemacompare.CompareIndex(old, True)),
+                    ("add_index", schemacompare.CompareIndex(new)),
+                ],
+            )
+        else:
+            with expect_warnings(
+                r"Skipped unsupported reflection of expression-based index "
+                r"SomeIndex",
+                r"autogenerate skipping metadata-specified expression-based "
+                r"index 'SomeIndex'; dialect '.*' under SQLAlchemy .* "
+                r"can't reflect these "
+                r"indexes so they can't be compared",
+            ):
+                diffs = self._fixture(m1, m2)
+            eq_(diffs, [])
+
+    @combinations(*_lots_of_indexes(flatten=True), argnames="fn")
+    def test_expression_indexes_no_change(self, index_changed_tables, fn):
+        m1, m2, old_fixture_tables, new_fixture_tables = index_changed_tables
+
+        resolve_lambda(fn, **old_fixture_tables)
+        resolve_lambda(fn, **new_fixture_tables)
+
+        if self.has_reflection:
+            ctx = nullcontext()
+        else:
+            ctx = expect_warnings()
+
+        with ctx:
+            diffs = self._fixture(m1, m2)
+        eq_(diffs, [])
+
+    def test_expression_indexes_remove(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        idx = Index("SomeIndex", "y", func.lower(column("x")))
+        Table(
+            "exp_index",
+            m1,
+            Column("id", Integer, primary_key=True),
+            Column("x", String(100)),
+            Column("y", Integer),
+            idx,
+        )
+
+        Table(
+            "exp_index",
+            m2,
+            Column("id", Integer, primary_key=True),
+            Column("x", String(100)),
+            Column("y", Integer),
+        )
+
+        if self.has_reflection:
+            diffs = self._fixture(m1, m2)
+            eq_(
+                diffs,
+                [("remove_index", schemacompare.CompareIndex(idx, True))],
+            )
+        else:
+            with expect_warnings():
+                diffs = self._fixture(m1, m2)
+            eq_(diffs, [])
+
 
 class NoUqReflectionIndexTest(NoUqReflection, AutogenerateUniqueIndexTest):
     __only_on__ = "sqlite"
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index 6a67e0b..c9860c2 100644
--- a/tests/test_postgresql.py
+++ b/tests/test_postgresql.py
@@ -39,7 +39,6 @@ from alembic.migration import MigrationContext
 from alembic.operations import ops
 from alembic.script import ScriptDirectory
 from alembic.testing import assert_raises_message
-from alembic.testing import assertions
 from alembic.testing import combinations
 from alembic.testing import config
 from alembic.testing import eq_
@@ -1305,64 +1304,3 @@ class PGUniqueIndexAutogenerateTest(AutogenFixtureTest, TestBase):
         eq_(diffs[0][0], "remove_constraint")
         eq_(diffs[0][1].name, "uq_name")
         eq_(len(diffs), 1)
-
-    def _functional_index_warn(self):
-        return (r"Skip.*refl",)
-
-    def test_functional_ix_one(self):
-        m1 = MetaData()
-        m2 = MetaData()
-
-        t1 = Table(
-            "foo",
-            m1,
-            Column("id", Integer, primary_key=True),
-            Column("email", String(50)),
-        )
-        Index("email_idx", func.lower(t1.c.email), unique=True)
-
-        t2 = Table(
-            "foo",
-            m2,
-            Column("id", Integer, primary_key=True),
-            Column("email", String(50)),
-        )
-        Index("email_idx", func.lower(t2.c.email), unique=True)
-
-        with assertions.expect_warnings(*self._functional_index_warn()):
-            diffs = self._fixture(m1, m2)
-        eq_(diffs, [])
-
-    def test_functional_ix_two(self):
-        m1 = MetaData()
-        m2 = MetaData()
-
-        t1 = Table(
-            "foo",
-            m1,
-            Column("id", Integer, primary_key=True),
-            Column("email", String(50)),
-            Column("name", String(50)),
-        )
-        Index(
-            "email_idx",
-            func.coalesce(t1.c.email, t1.c.name).desc(),
-            unique=True,
-        )
-
-        t2 = Table(
-            "foo",
-            m2,
-            Column("id", Integer, primary_key=True),
-            Column("email", String(50)),
-            Column("name", String(50)),
-        )
-        Index(
-            "email_idx",
-            func.coalesce(t2.c.email, t2.c.name).desc(),
-            unique=True,
-        )
-
-        with assertions.expect_warnings(*self._functional_index_warn()):
-            diffs = self._fixture(m1, m2)
-        eq_(diffs, [])
diff --git a/tests/test_sqlite.py b/tests/test_sqlite.py
index b8edeaa..4c60ce4 100644
--- a/tests/test_sqlite.py
+++ b/tests/test_sqlite.py
@@ -3,6 +3,7 @@ from sqlalchemy import Column
 from sqlalchemy import DateTime
 from sqlalchemy import Float
 from sqlalchemy import func
+from sqlalchemy import Index
 from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import MetaData
@@ -21,10 +22,12 @@ from alembic.testing import assert_raises_message
 from alembic.testing import config
 from alembic.testing import eq_
 from alembic.testing import eq_ignore_whitespace
+from alembic.testing.assertions import expect_warnings
 from alembic.testing.env import clear_staging_env
 from alembic.testing.env import staging_env
 from alembic.testing.fixtures import op_fixture
 from alembic.testing.fixtures import TestBase
+from alembic.testing.suite._autogen_fixtures import AutogenFixtureTest
 
 
 class SQLiteTest(TestBase):
@@ -274,3 +277,70 @@ class SQLiteAutogenRenderTest(TestBase):
             "sa.Column('int_value', sa.Integer(), "
             "nullable=True, sqlite_on_conflict_not_null='FAIL')",
         )
+
+
+class SQLiteAutogenIndexTest(AutogenFixtureTest, TestBase):
+    __requires__ = ("indexes_with_expressions",)
+    __only_on__ = "sqlite"
+    __backend__ = True
+
+    def _functional_index_warn(self):
+        return (r"Skip.*refl",)
+
+    def test_functional_ix_one(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        t1 = Table(
+            "foo",
+            m1,
+            Column("id", Integer, primary_key=True),
+            Column("email", String(50)),
+        )
+        Index("email_idx", func.lower(t1.c.email), unique=True)
+
+        t2 = Table(
+            "foo",
+            m2,
+            Column("id", Integer, primary_key=True),
+            Column("email", String(50)),
+        )
+        Index("email_idx", func.lower(t2.c.email), unique=True)
+
+        with expect_warnings(*self._functional_index_warn()):
+            diffs = self._fixture(m1, m2)
+        eq_(diffs, [])
+
+    def test_functional_ix_two(self):
+        m1 = MetaData()
+        m2 = MetaData()
+
+        t1 = Table(
+            "foo",
+            m1,
+            Column("id", Integer, primary_key=True),
+            Column("email", String(50)),
+            Column("name", String(50)),
+        )
+        Index(
+            "email_idx",
+            func.coalesce(t1.c.email, t1.c.name).desc(),
+            unique=True,
+        )
+
+        t2 = Table(
+            "foo",
+            m2,
+            Column("id", Integer, primary_key=True),
+            Column("email", String(50)),
+            Column("name", String(50)),
+        )
+        Index(
+            "email_idx",
+            func.coalesce(t2.c.email, t2.c.name).desc(),
+            unique=True,
+        )
+
+        with expect_warnings(*self._functional_index_warn()):
+            diffs = self._fixture(m1, m2)
+        eq_(diffs, [])