- [feature] Added reduce_columns() method

    to select() construct, replaces columns inline
    using the util.reduce_columns utility function
    to remove equivalent columns.  reduce_columns()
    also adds "with_only_synonyms" to limit the
    reduction just to those columns which have the same
    name.  The deprecated fold_equivalents() feature is
    removed [ticket:1729].
  - [feature] Added with_labels and
    reduce_columns keyword arguments to
    Query.subquery(), to provide two alternate
    strategies for producing queries with uniquely-
    named columns.  [ticket:1729].

8 files changed, 158 insertions, 80 deletions

diff --git a/CHANGES b/CHANGES
index 251167392..14216fdc1 100644
--- a/CHANGES
+++ b/CHANGES
@@ -179,6 +179,12 @@ underneath "0.7.xx".
     when dereferenced by a unit test.
     [ticket:2526]
 
+  - [feature] Added with_labels and
+    reduce_columns keyword arguments to
+    Query.subquery(), to provide two alternate
+    strategies for producing queries with uniquely-
+    named columns.  [ticket:1729].
+
   - [removed] Deprecated identifiers removed:
 
     * allow_null_pks mapper() argument
@@ -304,6 +310,15 @@ underneath "0.7.xx".
     method, auto correlates all selectables except those
     passed.
 
+  - [feature] Added reduce_columns() method
+    to select() construct, replaces columns inline
+    using the util.reduce_columns utility function
+    to remove equivalent columns.  reduce_columns()
+    also adds "with_only_synonyms" to limit the
+    reduction just to those columns which have the same
+    name.  The deprecated fold_equivalents() feature is
+    removed [ticket:1729].
+
   - [bug] All of UniqueConstraint, ForeignKeyConstraint,
     CheckConstraint, and PrimaryKeyConstraint will
     attach themselves to their parent table automatically
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index 0a345f284..46ee59298 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -434,25 +434,37 @@ class Query(object):
         # the annotation not being there
         return stmt._annotate({'no_replacement_traverse': True})
 
-    def subquery(self, name=None):
+    def subquery(self, name=None, with_labels=False, reduce_columns=False):
         """return the full SELECT statement represented by
         this :class:`.Query`, embedded within an :class:`.Alias`.
 
         Eager JOIN generation within the query is disabled.
 
-        The statement will not have disambiguating labels
-        applied to the list of selected columns unless the
-        :meth:`.Query.with_labels` method is used to generate a new
-        :class:`.Query` with the option enabled.
-
         :param name: string name to be assigned as the alias;
             this is passed through to :meth:`.FromClause.alias`.
             If ``None``, a name will be deterministically generated
             at compile time.
 
+        :param with_labels: if True, :meth:`.with_labels` will be called
+         on the :class:`.Query` first to apply table-qualified labels
+         to all columns.
+
+        :param reduce_columns: if True, :meth:`.Select.reduce_columns` will
+         be called on the resulting :func:`.select` construct,
+         to remove same-named columns where one also refers to the other
+         via foreign key or WHERE clause equivalence.
+
+         .. versionchanged:: 0.8 the ``with_labels`` and ``reduce_columns``
+            keyword arguments were added.
 
         """
-        return self.enable_eagerloads(False).statement.alias(name=name)
+        q = self.enable_eagerloads(False)
+        if with_labels:
+            q = q.with_labels()
+        q = q.statement
+        if reduce_columns:
+            q = q.reduce_columns()
+        return q.alias(name=name)
 
     def cte(self, name=None, recursive=False):
         """Return the full SELECT statement represented by this
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index e9786cbe1..b403ea88b 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -3687,7 +3687,6 @@ class Join(FromClause):
             self.onclause = onclause
 
         self.isouter = isouter
-        self.__folded_equivalents = None
 
     @property
     def description(self):
@@ -3720,7 +3719,6 @@ class Join(FromClause):
         self.left = clone(self.left, **kw)
         self.right = clone(self.right, **kw)
         self.onclause = clone(self.onclause, **kw)
-        self.__folded_equivalents = None
 
     def get_children(self, **kwargs):
         return self.left, self.right, self.onclause
@@ -3732,7 +3730,7 @@ class Join(FromClause):
             left_right = None
         return sqlutil.join_condition(left, right, a_subset=left_right)
 
-    def select(self, whereclause=None, fold_equivalents=False, **kwargs):
+    def select(self, whereclause=None, **kwargs):
         """Create a :class:`.Select` from this :class:`.Join`.
 
         The equivalent long-hand form, given a :class:`.Join` object
@@ -3746,22 +3744,11 @@ class Join(FromClause):
         :param whereclause: the WHERE criterion that will be sent to
           the :func:`select()` function
 
-        :param fold_equivalents: based on the join criterion of this
-          :class:`.Join`, do not include
-          repeat column names in the column list of the resulting
-          select, for columns that are calculated to be "equivalent"
-          based on the join criterion of this :class:`.Join`. This will
-          recursively apply to any joins directly nested by this one
-          as well.
-
         :param \**kwargs: all other kwargs are sent to the
           underlying :func:`select()` function.
 
         """
-        if fold_equivalents:
-            collist = sqlutil.folded_equivalents(self)
-        else:
-            collist = [self.left, self.right]
+        collist = [self.left, self.right]
 
         return select(collist, whereclause, from_obj=[self], **kwargs)
 
@@ -5149,6 +5136,37 @@ class Select(SelectBase):
         """
         self.append_column(column)
 
+    def reduce_columns(self, only_synonyms=True):
+        """Return a new :func`.select` construct with redundantly
+        named, equivalently-valued columns removed from the columns clause.
+
+        "Redundant" here means two columns where one refers to the
+        other either based on foreign key, or via a simple equality
+        comparison in the WHERE clause of the statement.   The primary purpose
+        of this method is to automatically construct a select statement
+        with all uniquely-named columns, without the need to use table-qualified
+        labels as :meth:`.apply_labels` does.
+
+        When columns are omitted based on foreign key, the referred-to
+        column is the one that's kept.  When columns are omitted based on
+        WHERE eqivalence, the first column in the columns clause is the
+        one that's kept.
+
+        :param only_synonyms: when True, limit the removal of columns
+         to those which have the same name as the equivalent.   Otherwise,
+         all columns that are equivalent to another are removed.
+
+        .. versionadded:: 0.8
+
+        """
+        return self.with_only_columns(
+                sqlutil.reduce_columns(
+                        self.inner_columns,
+                        *(self._whereclause, ) + tuple(self._from_obj),
+                        only_synonyms=only_synonyms
+                )
+            )
+
     @_generative
     def with_only_columns(self, columns):
         """Return a new :func:`.select` construct with its columns
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 0727f0537..6bfaf4b8c 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -585,6 +585,7 @@ def reduce_columns(columns, *clauses, **kw):
 
     """
     ignore_nonexistent_tables = kw.pop('ignore_nonexistent_tables', False)
+    only_synonyms = kw.pop('only_synonyms', False)
 
     columns = util.ordered_column_set(columns)
 
@@ -610,21 +611,27 @@ def reduce_columns(columns, *clauses, **kw):
                         continue
                     else:
                         raise
-                if fk_col.shares_lineage(c):
+                if fk_col.shares_lineage(c) and \
+                    (not only_synonyms or \
+                    c.name == col.name):
                     omit.add(col)
                     break
 
     if clauses:
         def visit_binary(binary):
             if binary.operator == operators.eq:
-                cols = util.column_set(chain(*[c.proxy_set for c in columns.difference(omit)]))
+                cols = util.column_set(chain(*[c.proxy_set
+                            for c in columns.difference(omit)]))
                 if binary.left in cols and binary.right in cols:
-                    for c in columns:
-                        if c.shares_lineage(binary.right):
+                    for c in reversed(columns):
+                        if c.shares_lineage(binary.right) and \
+                            (not only_synonyms or \
+                            c.name == binary.left.name):
                             omit.add(c)
                             break
         for clause in clauses:
-            visitors.traverse(clause, {}, {'binary':visit_binary})
+            if clause is not None:
+                visitors.traverse(clause, {}, {'binary': visit_binary})
 
     return expression.ColumnSet(columns.difference(omit))
 
@@ -677,48 +684,6 @@ def criterion_as_pairs(expression, consider_as_foreign_keys=None,
     visitors.traverse(expression, {}, {'binary':visit_binary})
     return pairs
 
-def folded_equivalents(join, equivs=None):
-    """Return a list of uniquely named columns.
-
-    The column list of the given Join will be narrowed
-    down to a list of all equivalently-named,
-    equated columns folded into one column, where 'equated' means they are
-    equated to each other in the ON clause of this join.
-
-    This function is used by Join.select(fold_equivalents=True).
-
-    Deprecated.   This function is used for a certain kind of
-    "polymorphic_union" which is designed to achieve joined
-    table inheritance where the base table has no "discriminator"
-    column; [ticket:1131] will provide a better way to
-    achieve this.
-
-    """
-    if equivs is None:
-        equivs = set()
-    def visit_binary(binary):
-        if binary.operator == operators.eq and binary.left.name == binary.right.name:
-            equivs.add(binary.right)
-            equivs.add(binary.left)
-    visitors.traverse(join.onclause, {}, {'binary':visit_binary})
-    collist = []
-    if isinstance(join.left, expression.Join):
-        left = folded_equivalents(join.left, equivs)
-    else:
-        left = list(join.left.columns)
-    if isinstance(join.right, expression.Join):
-        right = folded_equivalents(join.right, equivs)
-    else:
-        right = list(join.right.columns)
-    used = set()
-    for c in left + right:
-        if c in equivs:
-            if c.name not in used:
-                collist.append(c)
-                used.add(c.name)
-        else:
-            collist.append(c)
-    return collist
 
 class AliasedRow(object):
     """Wrap a RowProxy with a translation map.
diff --git a/test/orm/inheritance/test_abc_inheritance.py b/test/orm/inheritance/test_abc_inheritance.py
index 19d4f923b..3a61adb18 100644
--- a/test/orm/inheritance/test_abc_inheritance.py
+++ b/test/orm/inheritance/test_abc_inheritance.py
@@ -60,7 +60,6 @@ def produce_test(parent, child, direction):
             super(ABCTest, self).teardown()
 
 
-        @testing.uses_deprecated("fold_equivalents is deprecated.")
         def test_roundtrip(self):
             parent_table = {"a":ta, "b":tb, "c": tc}[parent]
             child_table = {"a":ta, "b":tb, "c": tc}[child]
@@ -84,14 +83,14 @@ def produce_test(parent, child, direction):
 
             abcjoin = polymorphic_union(
                 {"a":ta.select(tb.c.id==None, from_obj=[ta.outerjoin(tb, onclause=atob)]),
-                "b":ta.join(tb, onclause=atob).outerjoin(tc, onclause=btoc).select(tc.c.id==None, fold_equivalents=True),
+                "b":ta.join(tb, onclause=atob).outerjoin(tc, onclause=btoc).select(tc.c.id==None).reduce_columns(),
                 "c":tc.join(tb, onclause=btoc).join(ta, onclause=atob)
                 },"type", "abcjoin"
             )
 
             bcjoin = polymorphic_union(
             {
-            "b":ta.join(tb, onclause=atob).outerjoin(tc, onclause=btoc).select(tc.c.id==None, fold_equivalents=True),
+            "b":ta.join(tb, onclause=atob).outerjoin(tc, onclause=btoc).select(tc.c.id==None).reduce_columns(),
             "c":tc.join(tb, onclause=btoc).join(ta, onclause=atob)
             },"type", "bcjoin"
             )
diff --git a/test/orm/inheritance/test_assorted_poly.py b/test/orm/inheritance/test_assorted_poly.py
index 6ff989e73..fe9fb493b 100644
--- a/test/orm/inheritance/test_assorted_poly.py
+++ b/test/orm/inheritance/test_assorted_poly.py
@@ -639,7 +639,6 @@ class RelationshipTest7(fixtures.MappedTest):
                                     primary_key=True),
                 Column('category', String(70)))
 
-    @testing.uses_deprecated("fold_equivalents is deprecated.")
     def test_manytoone_lazyload(self):
         """test that lazy load clause to a polymorphic child mapper generates
         correctly [ticket:493]"""
@@ -686,8 +685,7 @@ class RelationshipTest7(fixtures.MappedTest):
         car_join = polymorphic_union(
             {
                 'car' : cars.outerjoin(offroad_cars).\
-                        select(offroad_cars.c.car_id == None,
-                                fold_equivalents=True),
+                        select(offroad_cars.c.car_id == None).reduce_columns(),
                 'offroad' : cars.join(offroad_cars)
             }, "type", 'car_join')
 
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index 44e016c86..6c5b98f01 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -900,6 +900,23 @@ class ExpressionTest(QueryTest, AssertsCompiledSQL):
         eq_(a2.name, 'foo2')
         eq_(a3.name, '%%(%d anon)s' % id(a3))
 
+    def test_labeled_subquery(self):
+        User = self.classes.User
+
+        session = create_session()
+        a1 = session.query(User.id).filter(User.id == 7).subquery(with_labels=True)
+        assert a1.c.users_id is not None
+
+    def test_reduced_subquery(self):
+        User = self.classes.User
+        ua = aliased(User)
+
+        session = create_session()
+        a1 = session.query(User.id, ua.id, ua.name).\
+                filter(User.id == ua.id).subquery(reduce_columns=True)
+        self.assert_compile(a1,
+                "SELECT users.id, users_1.name FROM "
+                "users, users AS users_1 WHERE users.id = users_1.id")
 
     def test_label(self):
         User = self.classes.User
diff --git a/test/sql/test_selectable.py b/test/sql/test_selectable.py
index 161eeaff4..ef5f99c40 100644
--- a/test/sql/test_selectable.py
+++ b/test/sql/test_selectable.py
@@ -871,11 +871,11 @@ class ReduceTest(fixtures.TestBase, AssertsExecutionResults):
 
     def test_reduce_selectable(self):
         metadata = MetaData()
-        engineers = Table('engineers', metadata, Column('engineer_id',
-                          Integer, primary_key=True),
+        engineers = Table('engineers', metadata,
+                        Column('engineer_id', Integer, primary_key=True),
                           Column('engineer_name', String(50)))
-        managers = Table('managers', metadata, Column('manager_id',
-                         Integer, primary_key=True),
+        managers = Table('managers', metadata,
+                        Column('manager_id', Integer, primary_key=True),
                          Column('manager_name', String(50)))
         s = select([engineers,
                    managers]).where(engineers.c.engineer_name
@@ -884,6 +884,61 @@ class ReduceTest(fixtures.TestBase, AssertsExecutionResults):
             util.column_set([s.c.engineer_id, s.c.engineer_name,
             s.c.manager_id]))
 
+    def test_reduce_generation(self):
+        m = MetaData()
+        t1 = Table('t1', m, Column('x', Integer, primary_key=True),
+                                        Column('y', Integer))
+        t2 = Table('t2', m, Column('z', Integer, ForeignKey('t1.x')),
+                                        Column('q', Integer))
+        s1 = select([t1, t2])
+        s2 = s1.reduce_columns(only_synonyms=False)
+        eq_(
+            set(s2.inner_columns),
+            set([t1.c.x, t1.c.y, t2.c.q])
+        )
+
+        s2 = s1.reduce_columns()
+        eq_(
+            set(s2.inner_columns),
+            set([t1.c.x, t1.c.y, t2.c.z, t2.c.q])
+        )
+
+
+    def test_reduce_only_synonym_fk(self):
+        m = MetaData()
+        t1 = Table('t1', m, Column('x', Integer, primary_key=True),
+                                        Column('y', Integer))
+        t2 = Table('t2', m, Column('x', Integer, ForeignKey('t1.x')),
+                                        Column('q', Integer, ForeignKey('t1.y')))
+        s1 = select([t1, t2])
+        s1 = s1.reduce_columns(only_synonyms=True)
+        eq_(
+            set(s1.c),
+            set([s1.c.x, s1.c.y, s1.c.q])
+        )
+
+    def test_reduce_only_synonym_lineage(self):
+        m = MetaData()
+        t1 = Table('t1', m, Column('x', Integer, primary_key=True),
+                                        Column('y', Integer),
+                                        Column('z', Integer)
+                            )
+        # test that the first appearance in the columns clause
+        # wins - t1 is first, t1.c.x wins
+        s1 = select([t1])
+        s2 = select([t1, s1]).where(t1.c.x == s1.c.x).where(s1.c.y == t1.c.z)
+        eq_(
+                set(s2.reduce_columns().inner_columns),
+                set([t1.c.x, t1.c.y, t1.c.z, s1.c.y, s1.c.z])
+        )
+
+        # reverse order, s1.c.x wins
+        s1 = select([t1])
+        s2 = select([s1, t1]).where(t1.c.x == s1.c.x).where(s1.c.y == t1.c.z)
+        eq_(
+                set(s2.reduce_columns().inner_columns),
+                set([s1.c.x, t1.c.y, t1.c.z, s1.c.y, s1.c.z])
+        )
 
     def test_reduce_aliased_join(self):
         metadata = MetaData()
@@ -937,7 +992,6 @@ class ReduceTest(fixtures.TestBase, AssertsExecutionResults):
             util.column_set([item_join.c.id, item_join.c.dummy,
             item_join.c.child_name]))
 
-
     def test_reduce_aliased_union_2(self):
         metadata = MetaData()
         page_table = Table('page', metadata, Column('id', Integer,