include mssql_clustered dialect_options when reflecting - issue #8288

Implemented reflection of the "clustered index" flag ``mssql_clustered``
for the SQL Server dialect. Pull request courtesy John Lennox.

Fixes: #8288
Closes: #8289
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/8289
Pull-request-sha: 1bb57352e3e31d8fb7de69ab5e60e5464949f640

Change-Id: Ife367066328f9e47ad823e4098647964a18e21e8

1 files changed, 21 insertions, 2 deletions

diff --git a/lib/sqlalchemy/dialects/mssql/base.py b/lib/sqlalchemy/dialects/mssql/base.py
index c85d21ef7..94e082611 100644
--- a/lib/sqlalchemy/dialects/mssql/base.py
+++ b/lib/sqlalchemy/dialects/mssql/base.py
@@ -3218,6 +3218,9 @@ class MSDialect(default.DefaultDialect):
         rp = connection.execution_options(future_result=True).execute(
             sql.text(
                 "select ind.index_id, ind.is_unique, ind.name, "
+                "case when ind.index_id = 1 "
+                "then cast(1 as bit) "
+                "else cast(0 as bit) end as is_clustered, "
                 f"{filter_definition} "
                 "from sys.indexes as ind join sys.tables as tab on "
                 "ind.object_id=tab.object_id "
@@ -3240,6 +3243,7 @@ class MSDialect(default.DefaultDialect):
                 "unique": row["is_unique"] == 1,
                 "column_names": [],
                 "include_columns": [],
+                "dialect_options": {"mssql_clustered": row["is_clustered"]},
             }
 
             if row["filter_definition"] is not None:
@@ -3566,7 +3570,15 @@ class MSDialect(default.DefaultDialect):
         # Primary key constraints
         s = (
             sql.select(
-                C.c.column_name, TC.c.constraint_type, C.c.constraint_name
+                C.c.column_name,
+                TC.c.constraint_type,
+                C.c.constraint_name,
+                func.objectproperty(
+                    func.object_id(
+                        C.c.table_schema + "." + C.c.constraint_name
+                    ),
+                    "CnstIsClustKey",
+                ).label("is_clustered"),
             )
             .where(
                 sql.and_(
@@ -3580,13 +3592,20 @@ class MSDialect(default.DefaultDialect):
         )
         c = connection.execution_options(future_result=True).execute(s)
         constraint_name = None
+        is_clustered = None
         for row in c.mappings():
             if "PRIMARY" in row[TC.c.constraint_type.name]:
                 pkeys.append(row["COLUMN_NAME"])
                 if constraint_name is None:
                     constraint_name = row[C.c.constraint_name.name]
+                if is_clustered is None:
+                    is_clustered = row["is_clustered"]
         if pkeys:
-            return {"constrained_columns": pkeys, "name": constraint_name}
+            return {
+                "constrained_columns": pkeys,
+                "name": constraint_name,
+                "dialect_options": {"mssql_clustered": is_clustered},
+            }
         else:
             return self._default_or_error(
                 connection,