Reflect expression-based indexes on PostgreSQL

The PostgreSQL dialect now supports reflection of expression based indexes.
The reflection is supported both when using
:meth:`_engine.Inspector.get_indexes` and when reflecting a
:class:`_schema.Table` using :paramref:`_schema.Table.autoload_with`.
Thanks to immerrr and Aidan Kane for the help on this ticket.

Fixes: #7442
Change-Id: I3e36d557235286c0f7f6d8276272ff9225058d48

4 files changed, 83 insertions, 13 deletions

diff --git a/lib/sqlalchemy/testing/__init__.py b/lib/sqlalchemy/testing/__init__.py
index 8c3c4bc27..0c83cb469 100644
--- a/lib/sqlalchemy/testing/__init__.py
+++ b/lib/sqlalchemy/testing/__init__.py
@@ -18,6 +18,7 @@ from .assertions import assert_warns
 from .assertions import assert_warns_message
 from .assertions import AssertsCompiledSQL
 from .assertions import AssertsExecutionResults
+from .assertions import ComparesIndexes
 from .assertions import ComparesTables
 from .assertions import emits_warning
 from .assertions import emits_warning_on
diff --git a/lib/sqlalchemy/testing/assertions.py b/lib/sqlalchemy/testing/assertions.py
index 937706363..44e7e892f 100644
--- a/lib/sqlalchemy/testing/assertions.py
+++ b/lib/sqlalchemy/testing/assertions.py
@@ -838,3 +838,34 @@ class AssertsExecutionResults:
 
     def assert_statement_count(self, db, count):
         return self.assert_execution(db, assertsql.CountStatements(count))
+
+
+class ComparesIndexes:
+    def compare_table_index_with_expected(
+        self, table: schema.Table, expected: list, dialect_name: str
+    ):
+        eq_(len(table.indexes), len(expected))
+        idx_dict = {idx.name: idx for idx in table.indexes}
+        for exp in expected:
+            idx = idx_dict[exp["name"]]
+            eq_(idx.unique, exp["unique"])
+            cols = [c for c in exp["column_names"] if c is not None]
+            eq_(len(idx.columns), len(cols))
+            for c in cols:
+                is_true(c in idx.columns)
+            exprs = exp.get("expressions")
+            if exprs:
+                eq_(len(idx.expressions), len(exprs))
+                for idx_exp, expr, col in zip(
+                    idx.expressions, exprs, exp["column_names"]
+                ):
+                    if col is None:
+                        eq_(idx_exp.text, expr)
+            if (
+                exp.get("dialect_options")
+                and f"{dialect_name}_include" in exp["dialect_options"]
+            ):
+                eq_(
+                    idx.dialect_options[dialect_name]["include"],
+                    exp["dialect_options"][f"{dialect_name}_include"],
+                )
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index cb955ff3d..55b10bdd5 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -736,6 +736,12 @@ class SuiteRequirements(Requirements):
         return exclusions.closed()
 
     @property
+    def reflect_indexes_with_expressions(self):
+        """target database supports reflection of indexes with
+        SQL expressions."""
+        return exclusions.closed()
+
+    @property
     def unique_constraint_reflection(self):
         """target dialect supports reflection of unique constraints"""
         return exclusions.open()
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index 6c71696a0..a3737a91a 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -35,6 +35,7 @@ from ...schema import DDL
 from ...schema import Index
 from ...sql.elements import quoted_name
 from ...sql.schema import BLANK_SCHEMA
+from ...testing import ComparesIndexes
 from ...testing import ComparesTables
 from ...testing import is_false
 from ...testing import is_true
@@ -2254,7 +2255,7 @@ class TableNoColumnsTest(fixtures.TestBase):
         eq_(multi, {(None, "empty_v"): []})
 
 
-class ComponentReflectionTestExtra(fixtures.TestBase):
+class ComponentReflectionTestExtra(ComparesIndexes, fixtures.TestBase):
 
     __backend__ = True
 
@@ -2322,9 +2323,10 @@ class ComponentReflectionTestExtra(fixtures.TestBase):
             metadata,
             Column("x", String(30)),
             Column("y", String(30)),
+            Column("z", String(30)),
         )
 
-        Index("t_idx", func.lower(t.c.x), func.lower(t.c.y))
+        Index("t_idx", func.lower(t.c.x), t.c.z, func.lower(t.c.y))
 
         Index("t_idx_2", t.c.x)
 
@@ -2335,19 +2337,49 @@ class ComponentReflectionTestExtra(fixtures.TestBase):
         expected = [
             {"name": "t_idx_2", "column_names": ["x"], "unique": False}
         ]
-        if testing.requires.index_reflects_included_columns.enabled:
-            expected[0]["include_columns"] = []
-            expected[0]["dialect_options"] = {
-                "%s_include" % connection.engine.name: []
+
+        def completeIndex(entry):
+            if testing.requires.index_reflects_included_columns.enabled:
+                entry["include_columns"] = []
+                entry["dialect_options"] = {
+                    f"{connection.engine.name}_include": []
+                }
+
+        completeIndex(expected[0])
+
+        class filtering_str(str):
+            def __eq__(self, other):
+                # test that lower and x or y are in the string
+                return "lower" in other and ("x" in other or "y" in other)
+
+        if testing.requires.reflect_indexes_with_expressions.enabled:
+            expr_index = {
+                "name": "t_idx",
+                "column_names": [None, "z", None],
+                "expressions": [
+                    filtering_str("lower(x)"),
+                    "z",
+                    filtering_str("lower(y)"),
+                ],
+                "unique": False,
             }
+            completeIndex(expr_index)
+            expected.insert(0, expr_index)
+            eq_(insp.get_indexes("t"), expected)
+            m2 = MetaData()
+            t2 = Table("t", m2, autoload_with=connection)
+        else:
+            with expect_warnings(
+                "Skipped unsupported reflection of expression-based "
+                "index t_idx"
+            ):
+                eq_(insp.get_indexes("t"), expected)
+                m2 = MetaData()
+                t2 = Table("t", m2, autoload_with=connection)
 
-        with expect_warnings(
-            "Skipped unsupported reflection of expression-based index t_idx"
-        ):
-            eq_(
-                insp.get_indexes("t"),
-                expected,
-            )
+        self.compare_table_index_with_expected(
+            t2, expected, connection.engine.name
+        )
 
     @testing.requires.index_reflects_included_columns
     def test_reflect_covering_index(self, metadata, connection):