turn off deduping for col expressions

Fixed ORM issue where column expressions passed to ``query()`` or
ORM-enabled ``select()`` would be deduplicated on the identity of the
object, such as a phrase like ``select(A.id, null(), null())`` would
produce only one "NULL" expression, which previously was not the case in
1.3. However, the change also allows for ORM expressions to render as given
as well, such as ``select(A.data, A.data)`` will produce a result row with
two columns.

Fixes: #6979
Change-Id: I4dd59d4c7b1baa711b686379eb959f87c44841c4

2 files changed, 19 insertions, 12 deletions

diff --git a/lib/sqlalchemy/orm/context.py b/lib/sqlalchemy/orm/context.py
index 9318bb163..85c736e12 100644
--- a/lib/sqlalchemy/orm/context.py
+++ b/lib/sqlalchemy/orm/context.py
@@ -187,6 +187,12 @@ class ORMCompileState(CompileState):
     def __init__(self, *arg, **kw):
         raise NotImplementedError()
 
+    def _append_dedupe_col_collection(self, obj, col_collection):
+        dedupe = self.dedupe_columns
+        if obj not in dedupe:
+            dedupe.add(obj)
+            col_collection.append(obj)
+
     @classmethod
     def _column_naming_convention(cls, label_style, legacy):
 
@@ -443,6 +449,7 @@ class ORMFromStatementCompileState(ORMCompileState):
 
         self.primary_columns = []
         self.secondary_columns = []
+        self.dedupe_columns = set()
         self.create_eager_joins = []
         self._fallback_from_clauses = []
 
@@ -645,6 +652,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
 
         self.primary_columns = []
         self.secondary_columns = []
+        self.dedupe_columns = set()
         self.eager_joins = {}
         self.extra_criteria_entities = {}
         self.create_eager_joins = []
@@ -765,8 +773,6 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
 
         # PART II
 
-        self.dedupe_cols = True
-
         self._for_update_arg = query._for_update_arg
 
         for entity in self._entities:
@@ -1036,9 +1042,8 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
         # put FOR UPDATE on the inner query, where MySQL will honor it,
         # as well as if it has an OF so PostgreSQL can use it.
         inner = self._select_statement(
-            util.unique_list(self.primary_columns + order_by_col_expr)
-            if self.dedupe_cols
-            else (self.primary_columns + order_by_col_expr),
+            self.primary_columns
+            + [c for c in order_by_col_expr if c not in self.dedupe_columns],
             self.from_clauses,
             self._where_criteria,
             self._having_criteria,
@@ -1116,9 +1121,7 @@ class ORMSelectCompileState(ORMCompileState, SelectState):
             self.primary_columns += to_add
 
         statement = self._select_statement(
-            util.unique_list(self.primary_columns + self.secondary_columns)
-            if self.dedupe_cols
-            else (self.primary_columns + self.secondary_columns),
+            self.primary_columns + self.secondary_columns,
             tuple(self.from_clauses) + tuple(self.eager_joins.values()),
             self._where_criteria,
             self._having_criteria,
@@ -2822,6 +2825,7 @@ class _RawColumnEntity(_ColumnEntity):
             # result due to the __eq__() method, so use deannotated
             column = column._deannotate()
 
+        compile_state.dedupe_columns.add(column)
         compile_state.primary_columns.append(column)
         self._fetch_column = column
 
@@ -2949,6 +2953,7 @@ class _ORMColumnEntity(_ColumnEntity):
         ):
             compile_state._fallback_from_clauses.append(ezero.selectable)
 
+        compile_state.dedupe_columns.add(column)
         compile_state.primary_columns.append(column)
         self._fetch_column = column
 
diff --git a/lib/sqlalchemy/orm/strategies.py b/lib/sqlalchemy/orm/strategies.py
index 069e5e667..4f361be2c 100644
--- a/lib/sqlalchemy/orm/strategies.py
+++ b/lib/sqlalchemy/orm/strategies.py
@@ -157,7 +157,7 @@ class UninstrumentedColumnLoader(LoaderStrategy):
         for c in self.columns:
             if adapter:
                 c = adapter.columns[c]
-            column_collection.append(c)
+            compile_state._append_dedupe_col_collection(c, column_collection)
 
     def create_row_processor(
         self,
@@ -206,7 +206,7 @@ class ColumnLoader(LoaderStrategy):
                 else:
                     c = adapter.columns[c]
 
-            column_collection.append(c)
+            compile_state._append_dedupe_col_collection(c, column_collection)
 
         fetch = self.columns[0]
         if adapter:
@@ -296,7 +296,7 @@ class ExpressionColumnLoader(ColumnLoader):
         for c in columns:
             if adapter:
                 c = adapter.columns[c]
-            column_collection.append(c)
+            compile_state._append_dedupe_col_collection(c, column_collection)
 
         fetch = columns[0]
         if adapter:
@@ -2335,7 +2335,9 @@ class JoinedLoader(AbstractRelationshipLoader):
                 if localparent.persist_selectable.c.contains_column(col):
                     if adapter:
                         col = adapter.columns[col]
-                    compile_state.primary_columns.append(col)
+                    compile_state._append_dedupe_col_collection(
+                        col, compile_state.primary_columns
+                    )
 
         if self.parent_property.order_by:
             compile_state.eager_order_by += tuple(