PostgreSQL now reflects per-column sort order on indexes.

Added support for column sorting flags when reflecting indexes for
PostgreSQL, including ASC, DESC, NULLSFIRST, NULLSLAST.  Also adds this
facility to the reflection system in general which can be applied to other
dialects in future releases.  Pull request courtesy Eli Collins.

Fixes: #4717
Closes: #4725
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/4725
Pull-request-sha: 3cbb067bd46776fdb125553ba0ac192cb45d060c

Change-Id: I8b0617d68580cfe4ff79d758a077263f33e852c2

2 files changed, 55 insertions, 4 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/base.py b/lib/sqlalchemy/dialects/postgresql/base.py
index 1363e81af..2bc48c53e 100644
--- a/lib/sqlalchemy/dialects/postgresql/base.py
+++ b/lib/sqlalchemy/dialects/postgresql/base.py
@@ -3196,7 +3196,7 @@ class PGDialect(default.DefaultDialect):
                   i.relname as relname,
                   ix.indisunique, ix.indexprs, ix.indpred,
                   a.attname, a.attnum, NULL, ix.indkey%s,
-                  %s, am.amname
+                  %s, %s, am.amname
               FROM
                   pg_class t
                         join pg_index ix on t.oid = ix.indrelid
@@ -3219,6 +3219,9 @@ class PGDialect(default.DefaultDialect):
                 # cast does not work in PG 8.2.4, does work in 8.3.0.
                 # nothing in PG changelogs regarding this.
                 "::varchar" if self.server_version_info >= (8, 3) else "",
+                "ix.indoption::varchar"
+                if self.server_version_info >= (8, 3)
+                else "NULL",
                 "i.reloptions"
                 if self.server_version_info >= (8, 2)
                 else "NULL",
@@ -3230,7 +3233,7 @@ class PGDialect(default.DefaultDialect):
                   i.relname as relname,
                   ix.indisunique, ix.indexprs, ix.indpred,
                   a.attname, a.attnum, c.conrelid, ix.indkey::varchar,
-                  i.reloptions, am.amname
+                  ix.indoption::varchar, i.reloptions, am.amname
               FROM
                   pg_class t
                         join pg_index ix on t.oid = ix.indrelid
@@ -3273,6 +3276,7 @@ class PGDialect(default.DefaultDialect):
                 col_num,
                 conrelid,
                 idx_key,
+                idx_option,
                 options,
                 amname,
             ) = row
@@ -3299,6 +3303,29 @@ class PGDialect(default.DefaultDialect):
                 index["cols"][col_num] = col
             if not has_idx:
                 index["key"] = [int(k.strip()) for k in idx_key.split()]
+
+                # (new in pg 8.3)
+                # "pg_index.indoption" is list of ints, one per column/expr.
+                # int acts as bitmask: 0x01=DESC, 0x02=NULLSFIRST
+                sorting = {}
+                for col_idx, col_flags in enumerate(
+                    (idx_option or "").split()
+                ):
+                    col_flags = int(col_flags.strip())
+                    col_sorting = ()
+                    # try to set flags only if they differ from PG defaults...
+                    if col_flags & 0x01:
+                        col_sorting += ("desc",)
+                        if not (col_flags & 0x02):
+                            col_sorting += ("nullslast",)
+                    else:
+                        if col_flags & 0x02:
+                            col_sorting += ("nullsfirst",)
+                    if col_sorting:
+                        sorting[col_idx] = col_sorting
+                if sorting:
+                    index["sorting"] = sorting
+
                 index["unique"] = unique
                 if conrelid is not None:
                     index["duplicates_constraint"] = idx_name
@@ -3323,6 +3350,11 @@ class PGDialect(default.DefaultDialect):
             }
             if "duplicates_constraint" in idx:
                 entry["duplicates_constraint"] = idx["duplicates_constraint"]
+            if "sorting" in idx:
+                entry["column_sorting"] = dict(
+                    (idx["cols"][idx["key"][i]], value)
+                    for i, value in idx["sorting"].items()
+                )
             if "options" in idx:
                 entry.setdefault("dialect_options", {})[
                     "postgresql_with"
diff --git a/lib/sqlalchemy/engine/reflection.py b/lib/sqlalchemy/engine/reflection.py
index 888149226..dcd1d0313 100644
--- a/lib/sqlalchemy/engine/reflection.py
+++ b/lib/sqlalchemy/engine/reflection.py
@@ -30,6 +30,7 @@ from .. import exc
 from .. import inspection
 from .. import sql
 from .. import util
+from ..sql import operators
 from ..sql import schema as sa_schema
 from ..sql.type_api import TypeEngine
 from ..util import deprecated
@@ -469,6 +470,12 @@ class Inspector(object):
         unique
           boolean
 
+        column_sorting
+          optional dict mapping column names to tuple of sort keywords,
+          which may include ``asc``, ``desc``, ``nullsfirst``, ``nullslast``.
+
+          .. versionadded:: 1.3.5
+
         dialect_options
           dict of dialect-specific index options.  May not be present
           for all dialects.
@@ -854,6 +861,13 @@ class Inspector(object):
                 )
             )
 
+    _index_sort_exprs = [
+        ("asc", operators.asc_op),
+        ("desc", operators.desc_op),
+        ("nullsfirst", operators.nullsfirst_op),
+        ("nullslast", operators.nullslast_op),
+    ]
+
     def _reflect_indexes(
         self,
         table_name,
@@ -869,6 +883,7 @@ class Inspector(object):
         for index_d in indexes:
             name = index_d["name"]
             columns = index_d["column_names"]
+            column_sorting = index_d.get("column_sorting", {})
             unique = index_d["unique"]
             flavor = index_d.get("type", "index")
             dialect_options = index_d.get("dialect_options", {})
@@ -897,8 +912,12 @@ class Inspector(object):
                         "%s key '%s' was not located in "
                         "columns for table '%s'" % (flavor, c, table_name)
                     )
-                else:
-                    idx_cols.append(idx_col)
+                    continue
+                c_sorting = column_sorting.get(c, ())
+                for k, op in self._index_sort_exprs:
+                    if k in c_sorting:
+                        idx_col = op(idx_col)
+                idx_cols.append(idx_col)
 
             sa_schema.Index(
                 name,