Improve oracle index reflection

Added reflection support in the Oracle dialect to expression based indexes
and the ordering direction of index expressions.

Fixes: #9597
Change-Id: I40e163496789774e9930f46823d2208c35eab6f8

6 files changed, 107 insertions, 31 deletions

diff --git a/lib/sqlalchemy/dialects/oracle/base.py b/lib/sqlalchemy/dialects/oracle/base.py
index a3e724cbe..d20175b0a 100644
--- a/lib/sqlalchemy/dialects/oracle/base.py
+++ b/lib/sqlalchemy/dialects/oracle/base.py
@@ -2304,6 +2304,8 @@ class OracleDialect(default.DefaultDialect):
             else:
                 return value
 
+        remove_size = re.compile(r"\(\d+\)")
+
         for row_dict in result:
             table_name = self.normalize_name(row_dict["table_name"])
             orig_colname = row_dict["column_name"]
@@ -2339,7 +2341,7 @@ class OracleDialect(default.DefaultDialect):
             elif "WITH LOCAL TIME ZONE" in coltype:
                 coltype = TIMESTAMP(local_timezone=True)
             else:
-                coltype = re.sub(r"\(\d+\)", "", coltype)
+                coltype = re.sub(remove_size, "", coltype)
                 try:
                     coltype = self.ischema_names[coltype]
                 except KeyError:
@@ -2557,6 +2559,8 @@ class OracleDialect(default.DefaultDialect):
                 dictionary.all_indexes.c.uniqueness,
                 dictionary.all_indexes.c.compression,
                 dictionary.all_indexes.c.prefix_length,
+                dictionary.all_ind_columns.c.descend,
+                dictionary.all_ind_expressions.c.column_expression,
             )
             .select_from(dictionary.all_ind_columns)
             .join(
@@ -2564,17 +2568,30 @@ class OracleDialect(default.DefaultDialect):
                 sql.and_(
                     dictionary.all_ind_columns.c.index_name
                     == dictionary.all_indexes.c.index_name,
-                    dictionary.all_ind_columns.c.table_owner
-                    == dictionary.all_indexes.c.table_owner,
-                    # NOTE: this condition on table_name is not required
-                    # but it improves the query performance noticeably
-                    dictionary.all_ind_columns.c.table_name
-                    == dictionary.all_indexes.c.table_name,
+                    dictionary.all_ind_columns.c.index_owner
+                    == dictionary.all_indexes.c.owner,
+                ),
+            )
+            .outerjoin(
+                # NOTE: this adds about 20% to the query time. Using a
+                # case expression with a scalar subquery only when needed
+                # with the assumption that most indexes are not expression
+                # would be faster but oracle does not like that with
+                # LONG datatype. It errors with:
+                # ORA-00997: illegal use of LONG datatype
+                dictionary.all_ind_expressions,
+                sql.and_(
+                    dictionary.all_ind_expressions.c.index_name
+                    == dictionary.all_ind_columns.c.index_name,
+                    dictionary.all_ind_expressions.c.index_owner
+                    == dictionary.all_ind_columns.c.index_owner,
+                    dictionary.all_ind_expressions.c.column_position
+                    == dictionary.all_ind_columns.c.column_position,
                 ),
             )
             .where(
-                dictionary.all_ind_columns.c.table_owner == owner,
-                dictionary.all_ind_columns.c.table_name.in_(
+                dictionary.all_indexes.c.table_owner == owner,
+                dictionary.all_indexes.c.table_name.in_(
                     bindparam("all_objects")
                 ),
             )
@@ -2604,11 +2621,12 @@ class OracleDialect(default.DefaultDialect):
             if row_dict["constraint_type"] == "P"
         }
 
+        # all_ind_expressions.column_expression is LONG
         result = self._run_batches(
             connection,
             query,
             dblink,
-            returns_long=False,
+            returns_long=True,
             mappings=True,
             all_objects=all_objects,
         )
@@ -2642,8 +2660,6 @@ class OracleDialect(default.DefaultDialect):
         enabled = {"DISABLED": False, "ENABLED": True}
         is_bitmap = {"BITMAP", "FUNCTION-BASED BITMAP"}
 
-        oracle_sys_col = re.compile(r"SYS_NC\d+\$", re.IGNORECASE)
-
         indexes = defaultdict(dict)
 
         for row_dict in self._get_indexes_rows(
@@ -2669,13 +2685,25 @@ class OracleDialect(default.DefaultDialect):
             else:
                 index_dict = table_indexes[index_name]
 
-            # filter out Oracle SYS_NC names.  could also do an outer join
-            # to the all_tab_columns table and check for real col names
-            # there.
-            if not oracle_sys_col.match(row_dict["column_name"]):
-                index_dict["column_names"].append(
-                    self.normalize_name(row_dict["column_name"])
-                )
+            expr = row_dict["column_expression"]
+            if expr is not None:
+                index_dict["column_names"].append(None)
+                if "expressions" in index_dict:
+                    index_dict["expressions"].append(expr)
+                else:
+                    index_dict["expressions"] = index_dict["column_names"][:-1]
+                    index_dict["expressions"].append(expr)
+
+                if row_dict["descend"].lower() != "asc":
+                    assert row_dict["descend"].lower() == "desc"
+                    cs = index_dict.setdefault("column_sorting", {})
+                    cs[expr] = ("desc",)
+            else:
+                assert row_dict["descend"].lower() == "asc"
+                cn = self.normalize_name(row_dict["column_name"])
+                index_dict["column_names"].append(cn)
+                if "expressions" in index_dict:
+                    index_dict["expressions"].append(cn)
 
         default = ReflectionDefaults.indexes
 
diff --git a/lib/sqlalchemy/dialects/oracle/dictionary.py b/lib/sqlalchemy/dialects/oracle/dictionary.py
index f79501072..fdf47ef31 100644
--- a/lib/sqlalchemy/dialects/oracle/dictionary.py
+++ b/lib/sqlalchemy/dialects/oracle/dictionary.py
@@ -393,6 +393,17 @@ all_indexes = Table(
     Column("auto", VARCHAR2(3)),
 ).alias("a_indexes")
 
+all_ind_expressions = Table(
+    "all_ind_expressions" + DB_LINK_PLACEHOLDER,
+    dictionary_meta,
+    Column("index_owner", VARCHAR2(128), nullable=False),
+    Column("index_name", VARCHAR2(128), nullable=False),
+    Column("table_owner", VARCHAR2(128), nullable=False),
+    Column("table_name", VARCHAR2(128), nullable=False),
+    Column("column_expression", LONG),
+    Column("column_position", NUMBER, nullable=False),
+).alias("a_ind_expressions")
+
 all_constraints = Table(
     "all_constraints" + DB_LINK_PLACEHOLDER,
     dictionary_meta,
diff --git a/lib/sqlalchemy/engine/interfaces.py b/lib/sqlalchemy/engine/interfaces.py
index e4914551c..e9e1d8ced 100644
--- a/lib/sqlalchemy/engine/interfaces.py
+++ b/lib/sqlalchemy/engine/interfaces.py
@@ -549,8 +549,9 @@ class ReflectedIndex(TypedDict):
     """
 
     column_sorting: NotRequired[Dict[str, Tuple[str]]]
-    """optional dict mapping column names to tuple of sort keywords,
-    which may include ``asc``, ``desc``, ``nulls_first``, ``nulls_last``.
+    """optional dict mapping column names or expressions to tuple of sort
+    keywords, which may include ``asc``, ``desc``, ``nulls_first``,
+    ``nulls_last``.
 
     .. versionadded:: 1.3.5
     """
diff --git a/lib/sqlalchemy/sql/base.py b/lib/sqlalchemy/sql/base.py
index 309555338..ee80b0514 100644
--- a/lib/sqlalchemy/sql/base.py
+++ b/lib/sqlalchemy/sql/base.py
@@ -1962,13 +1962,16 @@ class DedupeColumnCollection(ColumnCollection[str, _NAMEDCOL]):
             # in a _make_proxy operation
             util.memoized_property.reset(named_column, "proxy_set")
         else:
-            l = len(self._collection)
-            self._collection.append(
-                (key, named_column, _ColumnMetrics(self, named_column))
-            )
-            self._colset.add(named_column._deannotate())
-            self._index[l] = (key, named_column)
-            self._index[key] = (key, named_column)
+            self._append_new_column(key, named_column)
+
+    def _append_new_column(self, key: str, named_column: _NAMEDCOL) -> None:
+        l = len(self._collection)
+        self._collection.append(
+            (key, named_column, _ColumnMetrics(self, named_column))
+        )
+        self._colset.add(named_column._deannotate())
+        self._index[l] = (key, named_column)
+        self._index[key] = (key, named_column)
 
     def _populate_separate_keys(
         self, iter_: Iterable[Tuple[str, _NAMEDCOL]]
@@ -2057,6 +2060,9 @@ class DedupeColumnCollection(ColumnCollection[str, _NAMEDCOL]):
         if column.key in self._index:
             remove_col.add(self._index[column.key][1])
 
+        if not remove_col:
+            self._append_new_column(column.key, column)
+            return
         new_cols: List[Tuple[str, _NAMEDCOL, _ColumnMetrics[_NAMEDCOL]]] = []
         replaced = False
         for k, col, metrics in self._collection:
diff --git a/lib/sqlalchemy/testing/requirements.py b/lib/sqlalchemy/testing/requirements.py
index b59cce374..7286cd81b 100644
--- a/lib/sqlalchemy/testing/requirements.py
+++ b/lib/sqlalchemy/testing/requirements.py
@@ -763,6 +763,12 @@ class SuiteRequirements(Requirements):
         return exclusions.open()
 
     @property
+    def reflect_indexes_with_ascdesc_as_expression(self):
+        """target database supports reflecting INDEX with per-column
+        ASC/DESC but reflects them as expressions (like oracle)."""
+        return exclusions.closed()
+
+    @property
     def indexes_with_expressions(self):
         """target database supports CREATE INDEX against SQL expressions."""
         return exclusions.closed()
diff --git a/lib/sqlalchemy/testing/suite/test_reflection.py b/lib/sqlalchemy/testing/suite/test_reflection.py
index 5927df065..8c26c265b 100644
--- a/lib/sqlalchemy/testing/suite/test_reflection.py
+++ b/lib/sqlalchemy/testing/suite/test_reflection.py
@@ -1109,6 +1109,10 @@ class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
         ):
             fk_req = testing.requires.foreign_keys_reflect_as_index
             dup_req = testing.requires.unique_constraints_reflect_as_index
+            sorting_expression = (
+                testing.requires.reflect_indexes_with_ascdesc_as_expression
+            )
+
             if (fk and not fk_req.enabled) or (
                 duplicates and not dup_req.enabled
             ):
@@ -1121,7 +1125,13 @@ class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
                 "include_columns": [],
             }
             if column_sorting:
-                res["column_sorting"] = {"q": ("desc",)}
+                res["column_sorting"] = column_sorting
+                if sorting_expression.enabled:
+                    res["expressions"] = orig = res["column_names"]
+                    res["column_names"] = [
+                        None if c in column_sorting else c for c in orig
+                    ]
+
             if duplicates:
                 res["duplicates_constraint"] = name
             return [res]
@@ -2065,6 +2075,15 @@ class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
             insp.clear_cache()
             eq_(insp.get_multi_table_comment(**kw), exp)
 
+    def _check_expressions(self, result, exp, err_msg):
+        def _clean(text: str):
+            return re.sub(r"['\" ]", "", text).lower()
+
+        if isinstance(exp, dict):
+            eq_({_clean(e): v for e, v in result.items()}, exp, err_msg)
+        else:
+            eq_([_clean(e) for e in result], exp, err_msg)
+
     def _check_list(self, result, exp, req_keys=None, msg=None):
         if req_keys is None:
             eq_(result, exp, msg)
@@ -2073,7 +2092,11 @@ class ComponentReflectionTest(ComparesTables, OneConnectionTablesTest):
             for r, e in zip(result, exp):
                 for k in set(r) | set(e):
                     if k in req_keys or (k in r and k in e):
-                        eq_(r[k], e[k], f"{msg} - {k} - {r}")
+                        err_msg = f"{msg} - {k} - {r}"
+                        if k in ("expressions", "column_sorting"):
+                            self._check_expressions(r[k], e[k], err_msg)
+                        else:
+                            eq_(r[k], e[k], err_msg)
 
     def _check_table_dict(self, result, exp, req_keys=None, make_lists=False):
         eq_(set(result.keys()), set(exp.keys()))
@@ -2427,8 +2450,9 @@ class ComponentReflectionTestExtra(ComparesIndexes, fixtures.TestBase):
 
         class lower_index_str(str):
             def __eq__(self, other):
+                ol = other.lower()
                 # test that lower and x or y are in the string
-                return "lower" in other and ("x" in other or "y" in other)
+                return "lower" in ol and ("x" in ol or "y" in ol)
 
         class coalesce_index_str(str):
             def __eq__(self, other):