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

4 files changed, 140 insertions, 4 deletions

diff --git a/doc/build/changelog/unreleased_13/4717.rst b/doc/build/changelog/unreleased_13/4717.rst
new file mode 100644
index 000000000..70d9a7e68
--- /dev/null
+++ b/doc/build/changelog/unreleased_13/4717.rst
@@ -0,0 +1,8 @@
+.. change::
+    :tags: usecase, postgresql
+    :tickets: 4717
+
+    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.
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,
diff --git a/test/dialect/postgresql/test_reflection.py b/test/dialect/postgresql/test_reflection.py
index f2e491167..9580018be 100644
--- a/test/dialect/postgresql/test_reflection.py
+++ b/test/dialect/postgresql/test_reflection.py
@@ -907,6 +907,83 @@ class ReflectionTest(fixtures.TestBase):
             ],
         )
 
+    @testing.fails_if("postgresql < 8.3", "index ordering not supported")
+    @testing.provide_metadata
+    def test_index_reflection_with_sorting(self):
+        """reflect indexes with sorting options set"""
+
+        t1 = Table(
+            "party",
+            self.metadata,
+            Column("id", String(10), nullable=False),
+            Column("name", String(20)),
+            Column("aname", String(20)),
+        )
+
+        with testing.db.connect() as conn:
+
+            t1.create(conn)
+
+            # check ASC, DESC options alone
+            conn.execute(
+                """
+                create index idx1 on party
+                    (id, name ASC, aname DESC)
+            """
+            )
+
+            # check DESC w/ NULLS options
+            conn.execute(
+                """
+              create index idx2 on party
+                    (name DESC NULLS FIRST, aname DESC NULLS LAST)
+            """
+            )
+
+            # check ASC w/ NULLS options
+            conn.execute(
+                """
+              create index idx3 on party
+                    (name ASC NULLS FIRST, aname ASC NULLS LAST)
+            """
+            )
+
+        # reflect data
+        with testing.db.connect() as conn:
+            m2 = MetaData(conn)
+            t2 = Table("party", m2, autoload=True)
+
+        eq_(len(t2.indexes), 3)
+
+        # Make sure indexes are in the order we expect them in
+        r1, r2, r3 = sorted(t2.indexes, key=lambda idx: idx.name)
+
+        eq_(r1.name, "idx1")
+        eq_(r2.name, "idx2")
+        eq_(r3.name, "idx3")
+
+        # "ASC NULLS LAST" is implicit default for indexes,
+        # and "NULLS FIRST" is implicit default for "DESC".
+        # (https://www.postgresql.org/docs/11/indexes-ordering.html)
+
+        def compile_exprs(exprs):
+            return list(map(str, exprs))
+
+        eq_(
+            compile_exprs([t2.c.id, t2.c.name, t2.c.aname.desc()]),
+            compile_exprs(r1.expressions),
+        )
+
+        eq_(
+            compile_exprs([t2.c.name.desc(), t2.c.aname.desc().nullslast()]),
+            compile_exprs(r2.expressions),
+        )
+
+        eq_(
+            compile_exprs([t2.c.name.nullsfirst(), t2.c.aname]),
+            compile_exprs(r3.expressions),
+        )
+
     @testing.provide_metadata
     def test_index_reflection_modified(self):
         """reflect indexes when a column name has changed - PG 9