- A new construct :class:`.Bundle` is added, which allows for specification

of groups of column expressions to a :class:`.Query` construct.
The group of columns are returned as a single tuple by default.  The
behavior of :class:`.Bundle` can be overridden however to provide
any sort of result processing to the returned row.  One example included
is :attr:`.Composite.Comparator.bundle`, which applies a bundled form
of a "composite" mapped attribute.
[ticket:2824]
- The :func:`.composite` construct now maintains the return object
when used in a column-oriented :class:`.Query`, rather than expanding
out into individual columns.  This makes use of the new :class:`.Bundle`
feature internally.  This behavior is backwards incompatible; to
select from a composite column which will expand out, use
``MyClass.some_composite.clauses``.

11 files changed, 634 insertions, 14 deletions

diff --git a/doc/build/changelog/changelog_09.rst b/doc/build/changelog/changelog_09.rst
index b27e282ac..b09ecefae 100644
--- a/doc/build/changelog/changelog_09.rst
+++ b/doc/build/changelog/changelog_09.rst
@@ -13,6 +13,39 @@
     :version: 0.9.0
 
     .. change::
+        :tags: feature, orm
+        :tickets: 2824
+
+        The :func:`.composite` construct now maintains the return object
+        when used in a column-oriented :class:`.Query`, rather than expanding
+        out into individual columns.  This makes use of the new :class:`.Bundle`
+        feature internally.  This behavior is backwards incompatible; to
+        select from a composite column which will expand out, use
+        ``MyClass.some_composite.clauses``.
+
+        .. seealso::
+
+            :ref:`migration_2824`
+
+    .. change::
+        :tags: feature, orm
+        :tickets: 2824
+
+        A new construct :class:`.Bundle` is added, which allows for specification
+        of groups of column expressions to a :class:`.Query` construct.
+        The group of columns are returned as a single tuple by default.  The
+        behavior of :class:`.Bundle` can be overridden however to provide
+        any sort of result processing to the returned row.  The behavior
+        of :class:`.Bundle` is also embedded into composite attributes now
+        when they are used in a column-oriented :class:`.Query`.
+
+        .. seealso::
+
+            :ref:`change_2824`
+
+            :ref:`migration_2824`
+
+    .. change::
         :tags: bug, sql
         :tickets: 2812
 
diff --git a/doc/build/changelog/migration_09.rst b/doc/build/changelog/migration_09.rst
index cf345edbc..2952c67b0 100644
--- a/doc/build/changelog/migration_09.rst
+++ b/doc/build/changelog/migration_09.rst
@@ -52,6 +52,35 @@ in both Python 2 and Python 3 environments.
 Behavioral Changes
 ==================
 
+.. _migration_2824:
+
+Composite attributes are now returned as their object form when queried on a per-attribute basis
+------------------------------------------------------------------------------------------------
+
+Using a :class:`.Query` in conjunction with a composite attribute now returns the object
+type maintained by that composite, rather than being broken out into individual
+columns.   Using the mapping setup at :ref:`mapper_composite`::
+
+    >>> session.query(Vertex.start, Vertex.end).\
+    ...     filter(Vertex.start == Point(3, 4)).all()
+    [(Point(x=3, y=4), Point(x=5, y=6))]
+
+This change is backwards-incompatible with code that expects the indivdual attribute
+to be expanded into individual columns.  To get that behavior, use the ``.clauses``
+accessor::
+
+
+    >>> session.query(Vertex.start.clauses, Vertex.end.clauses).\
+    ...     filter(Vertex.start == Point(3, 4)).all()
+    [(3, 4, 5, 6)]
+
+.. seealso::
+
+    :ref:`change_2824`
+
+:ticket:`2824`
+
+
 .. _migration_2736:
 
 :meth:`.Query.select_from` no longer applies the clause to corresponding entities
@@ -391,6 +420,28 @@ rendering::
 
 :ticket:`722`
 
+.. _change_2824:
+
+Column Bundles for ORM queries
+------------------------------
+
+The :class:`.Bundle` allows for querying of sets of columns, which are then
+grouped into one name under the tuple returned by the query.  The initial
+purposes of :class:`.Bundle` are 1. to allow "composite" ORM columns to be
+returned as a single value in a column-based result set, rather than expanding
+them out into individual columns and 2. to allow the creation of custom result-set
+constructs within the ORM, using ad-hoc columns and return types, without involving
+the more heavyweight mechanics of mapped classes.
+
+.. seealso::
+
+    :ref:`migration_2824`
+
+    :ref:`bundles`
+
+:ticket:`2824`
+
+
 Server Side Version Counting
 -----------------------------
 
diff --git a/doc/build/orm/mapper_config.rst b/doc/build/orm/mapper_config.rst
index d35910745..420ab3a32 100644
--- a/doc/build/orm/mapper_config.rst
+++ b/doc/build/orm/mapper_config.rst
@@ -772,6 +772,10 @@ class you provide.
     in-place mutation is no longer automatic; see the section below on
     enabling mutability to support tracking of in-place changes.
 
+.. versionchanged:: 0.9
+    Composites will return their object-form, rather than as individual columns,
+    when used in a column-oriented :class:`.Query` construct.  See :ref:`migration_2824`.
+
 A simple example represents pairs of columns as a ``Point`` object.
 ``Point`` represents such a pair as ``.x`` and ``.y``::
 
@@ -911,6 +915,54 @@ the same expression that the base "greater than" does::
         end = composite(Point, x2, y2,
                             comparator_factory=PointComparator)
 
+.. _bundles:
+
+Column Bundles
+===============
+
+The :class:`.Bundle` may be used to query for groups of columns under one
+namespace.
+
+.. versionadded:: 0.9.0
+
+The bundle allows columns to be grouped together::
+
+    from sqlalchemy.orm import Bundle
+
+    bn = Bundle('mybundle', MyClass.data1, MyClass.data2)
+    for row in session.query(bn).filter(bn.c.data1 == 'd1'):
+        print row.mybundle.data1, row.mybundle.data2
+
+The bundle can be subclassed to provide custom behaviors when results
+are fetched.  The method :meth:`.Bundle.create_row_processor` is given
+the :class:`.Query` and a set of "row processor" functions at query execution
+time; these processor functions when given a result row will return the
+individual attribute value, which can then be adapted into any kind of
+return data structure.  Below illustrates replacing the usual :class:`.KeyedTuple`
+return structure with a straight Python dictionary::
+
+    from sqlalchemy.orm import Bundle
+
+    class DictBundle(Bundle):
+        def create_row_processor(self, query, procs, labels):
+            """Override create_row_processor to return values as dictionaries"""
+            def proc(row, result):
+                return dict(
+                            zip(labels, (proc(row, result) for proc in procs))
+                        )
+            return proc
+
+A result from the above bundle will return dictionary values::
+
+    bn = DictBundle('mybundle', MyClass.data1, MyClass.data2)
+    for row in session.query(bn).filter(bn.c.data1 == 'd1'):
+        print row.mybundle['data1'], row.mybundle['data2']
+
+The :class:`.Bundle` construct is also integrated into the behavior
+of :func:`.composite`, where it is used to return composite attributes as objects
+when queried as individual attributes.
+
+
 .. _maptojoin:
 
 Mapping a Class against Multiple Tables
diff --git a/doc/build/orm/query.rst b/doc/build/orm/query.rst
index 73aa5c555..344c4e013 100644
--- a/doc/build/orm/query.rst
+++ b/doc/build/orm/query.rst
@@ -31,6 +31,9 @@ ORM-Specific Query Constructs
 
 .. autoclass:: sqlalchemy.orm.util.AliasedInsp
 
+.. autoclass:: sqlalchemy.orm.query.Bundle
+	:members:
+
 .. autoclass:: sqlalchemy.util.KeyedTuple
 	:members: keys, _fields, _asdict
 
diff --git a/lib/sqlalchemy/orm/__init__.py b/lib/sqlalchemy/orm/__init__.py
index e70cc1c55..5cd0f2854 100644
--- a/lib/sqlalchemy/orm/__init__.py
+++ b/lib/sqlalchemy/orm/__init__.py
@@ -62,7 +62,7 @@ from .scoping import (
     scoped_session
 )
 from . import mapper as mapperlib
-from .query import AliasOption, Query
+from .query import AliasOption, Query, Bundle
 from ..util.langhelpers import public_factory
 from .. import util as _sa_util
 from . import strategies as _strategies
diff --git a/lib/sqlalchemy/orm/attributes.py b/lib/sqlalchemy/orm/attributes.py
index 949eafca4..da9d62d13 100644
--- a/lib/sqlalchemy/orm/attributes.py
+++ b/lib/sqlalchemy/orm/attributes.py
@@ -143,6 +143,12 @@ class QueryableAttribute(interfaces._MappedAttribute,
     def __clause_element__(self):
         return self.comparator.__clause_element__()
 
+    def _query_clause_element(self):
+        """like __clause_element__(), but called specifically
+        by :class:`.Query` to allow special behavior."""
+
+        return self.comparator._query_clause_element()
+
     def of_type(self, cls):
         return QueryableAttribute(
                     self.class_,
@@ -153,7 +159,7 @@ class QueryableAttribute(interfaces._MappedAttribute,
                     of_type=cls)
 
     def label(self, name):
-        return self.__clause_element__().label(name)
+        return self._query_clause_element().label(name)
 
     def operate(self, op, *other, **kwargs):
         return op(self.comparator, *other, **kwargs)
diff --git a/lib/sqlalchemy/orm/descriptor_props.py b/lib/sqlalchemy/orm/descriptor_props.py
index 457b26523..bbfe602d0 100644
--- a/lib/sqlalchemy/orm/descriptor_props.py
+++ b/lib/sqlalchemy/orm/descriptor_props.py
@@ -16,6 +16,7 @@ from . import attributes
 from .. import util, sql, exc as sa_exc, event, schema
 from ..sql import expression
 from . import properties
+from . import query
 
 
 class DescriptorProperty(MapperProperty):
@@ -83,9 +84,9 @@ class CompositeProperty(DescriptorProperty):
     :class:`.CompositeProperty` is constructed using the :func:`.composite`
     function.
 
-    See also:
+    .. seealso::
 
-    :ref:`mapper_composite`
+        :ref:`mapper_composite`
 
     """
     def __init__(self, class_, *attrs, **kwargs):
@@ -154,6 +155,7 @@ class CompositeProperty(DescriptorProperty):
         util.set_creation_order(self)
         self._create_descriptor()
 
+
     def instrument_class(self, mapper):
         super(CompositeProperty, self).instrument_class(mapper)
         self._setup_event_handlers()
@@ -354,6 +356,18 @@ class CompositeProperty(DescriptorProperty):
     def _comparator_factory(self, mapper):
         return self.comparator_factory(self, mapper)
 
+    class CompositeBundle(query.Bundle):
+        def __init__(self, property, expr):
+            self.property = property
+            super(CompositeProperty.CompositeBundle, self).__init__(
+                        property.key, *expr)
+
+        def create_row_processor(self, query, procs, labels):
+            def proc(row, result):
+                return self.property.composite_class(*[proc(row, result) for proc in procs])
+            return proc
+
+
     class Comparator(PropComparator):
         """Produce boolean, comparison, and other operators for
         :class:`.CompositeProperty` attributes.
@@ -373,10 +387,18 @@ class CompositeProperty(DescriptorProperty):
 
         """
 
+
+        __hash__ = None
+
+        @property
+        def clauses(self):
+            return self.__clause_element__()
+
         def __clause_element__(self):
             return expression.ClauseList(group=False, *self._comparable_elements)
 
-        __hash__ = None
+        def _query_clause_element(self):
+            return CompositeProperty.CompositeBundle(self.prop, self.__clause_element__())
 
         @util.memoized_property
         def _comparable_elements(self):
diff --git a/lib/sqlalchemy/orm/interfaces.py b/lib/sqlalchemy/orm/interfaces.py
index 4a1a1823d..2f4aa5208 100644
--- a/lib/sqlalchemy/orm/interfaces.py
+++ b/lib/sqlalchemy/orm/interfaces.py
@@ -320,6 +320,9 @@ class PropComparator(operators.ColumnOperators):
     def __clause_element__(self):
         raise NotImplementedError("%r" % self)
 
+    def _query_clause_element(self):
+        return self.__clause_element__()
+
     def adapt_to_entity(self, adapt_to_entity):
         """Return a copy of this PropComparator which will use the given
         :class:`.AliasedInsp` to produce corresponding expressions.
diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index d64575aec..c3e5aa10d 100644
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -35,6 +35,8 @@ from ..sql import (
         util as sql_util,
         expression, visitors
     )
+from ..sql.base import ColumnCollection
+from ..sql import operators
 from . import properties
 
 __all__ = ['Query', 'QueryContext', 'aliased']
@@ -2890,6 +2892,8 @@ class _QueryEntity(object):
             if not isinstance(entity, util.string_types) and \
                         _is_mapped_class(entity):
                 cls = _MapperEntity
+            elif isinstance(entity, Bundle):
+                cls = _BundleEntity
             else:
                 cls = _ColumnEntity
         return object.__new__(cls)
@@ -3089,6 +3093,163 @@ class _MapperEntity(_QueryEntity):
     def __str__(self):
         return str(self.mapper)
 
+@inspection._self_inspects
+class Bundle(object):
+    """A grouping of SQL expressions that are returned by a :class:`.Query`
+    under one namespace.
+
+    The :class:`.Bundle` essentially allows nesting of the tuple-based
+    results returned by a column-oriented :class:`.Query` object.  It also
+    is extensible via simple subclassing, where the primary capability
+    to override is that of how the set of expressions should be returned,
+    allowing post-processing as well as custom return types, without
+    involving ORM identity-mapped classes.
+
+    .. versionadded:: 0.9.0
+
+    .. seealso::
+
+        :ref:`bundles`
+
+    """
+
+    def __init__(self, name, *exprs):
+        """Construct a new :class:`.Bundle`.
+
+        e.g.::
+
+            bn = Bundle("mybundle", MyClass.x, MyClass.y)
+
+            for row in session.query(bn).filter(bn.c.x == 5).filter(bn.c.y == 4):
+                print(row.mybundle.x, row.mybundle.y)
+
+        """
+        self.name = self._label = name
+        self.exprs = exprs
+        self.c = self.columns = ColumnCollection()
+        self.columns.update((getattr(col, "key", col._label), col)
+                    for col in exprs)
+
+    columns = None
+    """A namespace of SQL expressions referred to by this :class:`.Bundle`.
+
+        e.g.::
+
+            bn = Bundle("mybundle", MyClass.x, MyClass.y)
+
+            q = sess.query(bn).filter(bn.c.x == 5)
+
+        Nesting of bundles is also supported::
+
+            b1 = Bundle("b1",
+                    Bundle('b2', MyClass.a, MyClass.b),
+                    Bundle('b3', MyClass.x, MyClass.y)
+                )
+
+            q = sess.query(b1).filter(b1.c.b2.c.a == 5).filter(b1.c.b3.c.y == 9)
+
+    .. seealso::
+
+        :attr:`.Bundle.c`
+
+    """
+
+    c = None
+    """An alias for :attr:`.Bundle.columns`."""
+
+    def _clone(self):
+        cloned = self.__class__.__new__(self.__class__)
+        cloned.__dict__.update(self.__dict__)
+        return cloned
+
+    def __clause_element__(self):
+        return expression.ClauseList(group=False, *self.c)
+
+    @property
+    def clauses(self):
+        return self.__clause_element__().clauses
+
+    def label(self, name):
+        """Provide a copy of this :class:`.Bundle` passing a new label."""
+
+        cloned = self._clone()
+        cloned.name = name
+        return cloned
+
+    def create_row_processor(self, query, procs, labels):
+        """Produce the "row processing" function for this :class:`.Bundle`.
+
+        May be overridden by subclasses.
+
+        .. seealso::
+
+            :ref:`bundles` - includes an example of subclassing.
+
+        """
+        def proc(row, result):
+            return util.KeyedTuple([proc(row, None) for proc in procs], labels)
+        return proc
+
+
+class _BundleEntity(_QueryEntity):
+    def __init__(self, query, bundle, setup_entities=True):
+        query._entities.append(self)
+        self.bundle = self.entity_zero = bundle
+        self.type = type(bundle)
+        self._label_name = bundle.name
+        self._entities = []
+
+        if setup_entities:
+            for expr in bundle.exprs:
+                if isinstance(expr, Bundle):
+                    _BundleEntity(self, expr)
+                else:
+                    _ColumnEntity(self, expr, namespace=self)
+
+        self.entities = ()
+
+        self.filter_fn = lambda item: item
+
+    def corresponds_to(self, entity):
+        # TODO: this seems to have no effect for
+        # _ColumnEntity either
+        return False
+
+    @property
+    def entity_zero_or_selectable(self):
+        for ent in self._entities:
+            ezero = ent.entity_zero_or_selectable
+            if ezero is not None:
+                return ezero
+        else:
+            return None
+
+    def adapt_to_selectable(self, query, sel):
+        c = _BundleEntity(query, self.bundle, setup_entities=False)
+        #c._label_name = self._label_name
+        #c.entity_zero = self.entity_zero
+        #c.entities = self.entities
+
+        for ent in self._entities:
+            ent.adapt_to_selectable(c, sel)
+
+    def setup_entity(self, ext_info, aliased_adapter):
+        for ent in self._entities:
+            ent.setup_entity(ext_info, aliased_adapter)
+
+    def setup_context(self, query, context):
+        for ent in self._entities:
+            ent.setup_context(query, context)
+
+    def row_processor(self, query, context, custom_rows):
+        procs, labels = zip(
+                *[ent.row_processor(query, context, custom_rows)
+                for ent in self._entities]
+            )
+
+        proc = self.bundle.create_row_processor(query, procs, labels)
+
+        return proc, self._label_name
 
 class _ColumnEntity(_QueryEntity):
     """Column/expression based entity."""
@@ -3105,7 +3266,7 @@ class _ColumnEntity(_QueryEntity):
                                     interfaces.PropComparator
                                 )):
             self._label_name = column.key
-            column = column.__clause_element__()
+            column = column._query_clause_element()
         else:
             self._label_name = getattr(column, 'key', None)
 
@@ -3118,6 +3279,9 @@ class _ColumnEntity(_QueryEntity):
 
             if c is not column:
                 return
+        elif isinstance(column, Bundle):
+            _BundleEntity(query, column)
+            return
 
         if not isinstance(column, sql.ColumnElement):
             raise sa_exc.InvalidRequestError(
@@ -3125,7 +3289,7 @@ class _ColumnEntity(_QueryEntity):
                 "expected - got '%r'" % (column, )
             )
 
-        type_ = column.type
+        self.type = type_ = column.type
         if type_.hashable:
             self.filter_fn = lambda item: item
         else:
@@ -3177,10 +3341,6 @@ class _ColumnEntity(_QueryEntity):
         else:
             return None
 
-    @property
-    def type(self):
-        return self.column.type
-
     def adapt_to_selectable(self, query, sel):
         c = _ColumnEntity(query, sel.corresponding_column(self.column))
         c._label_name = self._label_name
@@ -3193,6 +3353,8 @@ class _ColumnEntity(_QueryEntity):
         self.froms.add(ext_info.selectable)
 
     def corresponds_to(self, entity):
+        # TODO: just returning False here,
+        # no tests fail
         if self.entity_zero is None:
             return False
         elif _is_aliased_class(entity):
diff --git a/test/orm/test_bundle.py b/test/orm/test_bundle.py
new file mode 100644
index 000000000..305f8d3c6
--- /dev/null
+++ b/test/orm/test_bundle.py
@@ -0,0 +1,245 @@
+from sqlalchemy.testing import fixtures, eq_
+from sqlalchemy.testing.schema import Table, Column
+from sqlalchemy.orm import Bundle, Session
+from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy import Integer, select, ForeignKey, String, func
+from sqlalchemy.orm import mapper, relationship, aliased
+
+class BundleTest(fixtures.MappedTest, AssertsCompiledSQL):
+    __dialect__ = 'default'
+
+    run_inserts = 'once'
+    run_setup_mappers = 'once'
+    run_deletes = None
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table('data', metadata,
+                Column('id', Integer, primary_key=True,
+                            test_needs_autoincrement=True),
+                Column('d1', String(10)),
+                Column('d2', String(10)),
+                Column('d3', String(10))
+            )
+
+        Table('other', metadata,
+                Column('id', Integer, primary_key=True, test_needs_autoincrement=True),
+                Column('data_id', ForeignKey('data.id')),
+                Column('o1', String(10))
+            )
+
+    @classmethod
+    def setup_classes(cls):
+        class Data(cls.Basic):
+            pass
+        class Other(cls.Basic):
+            pass
+
+    @classmethod
+    def setup_mappers(cls):
+        mapper(cls.classes.Data, cls.tables.data, properties={
+                'others': relationship(cls.classes.Other)
+            })
+        mapper(cls.classes.Other, cls.tables.other)
+
+    @classmethod
+    def insert_data(cls):
+        sess = Session()
+        sess.add_all([
+            cls.classes.Data(d1='d%dd1' % i, d2='d%dd2' % i, d3='d%dd3' % i,
+                    others=[cls.classes.Other(o1="d%do%d" % (i, j)) for j in range(5)])
+            for i in range(10)
+        ])
+        sess.commit()
+
+    def test_c_attr(self):
+        Data = self.classes.Data
+
+        b1 = Bundle('b1', Data.d1, Data.d2)
+
+        self.assert_compile(
+            select([b1.c.d1, b1.c.d2]),
+            "SELECT data.d1, data.d2 FROM data"
+        )
+
+    def test_result(self):
+        Data = self.classes.Data
+        sess = Session()
+
+        b1 = Bundle('b1', Data.d1, Data.d2)
+
+        eq_(
+            sess.query(b1).filter(b1.c.d1.between('d3d1', 'd5d1')).all(),
+            [(('d3d1', 'd3d2'),), (('d4d1', 'd4d2'),), (('d5d1', 'd5d2'),)]
+        )
+
+    def test_subclass(self):
+        Data = self.classes.Data
+        sess = Session()
+
+        class MyBundle(Bundle):
+            def create_row_processor(self, query, procs, labels):
+                def proc(row, result):
+                    return dict(
+                                zip(labels, (proc(row, result) for proc in procs))
+                            )
+                return proc
+
+        b1 = MyBundle('b1', Data.d1, Data.d2)
+
+        eq_(
+            sess.query(b1).filter(b1.c.d1.between('d3d1', 'd5d1')).all(),
+            [({'d2': 'd3d2', 'd1': 'd3d1'},),
+                ({'d2': 'd4d2', 'd1': 'd4d1'},),
+                ({'d2': 'd5d2', 'd1': 'd5d1'},)]
+        )
+
+    def test_multi_bundle(self):
+        Data = self.classes.Data
+        Other = self.classes.Other
+
+        d1 = aliased(Data)
+
+        b1 = Bundle('b1', d1.d1, d1.d2)
+        b2 = Bundle('b2', Data.d1, Other.o1)
+
+        sess = Session()
+
+        q = sess.query(b1, b2).join(Data.others).join(d1, d1.id == Data.id).\
+            filter(b1.c.d1 == 'd3d1')
+        eq_(
+            q.all(),
+            [
+                (('d3d1', 'd3d2'), ('d3d1', 'd3o0')),
+                (('d3d1', 'd3d2'), ('d3d1', 'd3o1')),
+                (('d3d1', 'd3d2'), ('d3d1', 'd3o2')),
+                (('d3d1', 'd3d2'), ('d3d1', 'd3o3')),
+                (('d3d1', 'd3d2'), ('d3d1', 'd3o4'))]
+        )
+
+    def test_bundle_nesting(self):
+        Data = self.classes.Data
+        sess = Session()
+
+        b1 = Bundle('b1', Data.d1, Bundle('b2', Data.d2, Data.d3))
+
+        eq_(
+            sess.query(b1).
+                filter(b1.c.d1.between('d3d1', 'd7d1')).
+                filter(b1.c.b2.c.d2.between('d4d2', 'd6d2')).
+                all(),
+            [(('d4d1', ('d4d2', 'd4d3')),), (('d5d1', ('d5d2', 'd5d3')),),
+                (('d6d1', ('d6d2', 'd6d3')),)]
+        )
+
+    def test_bundle_nesting_unions(self):
+        Data = self.classes.Data
+        sess = Session()
+
+        b1 = Bundle('b1', Data.d1, Bundle('b2', Data.d2, Data.d3))
+
+        q1 = sess.query(b1).\
+                filter(b1.c.d1.between('d3d1', 'd7d1')).\
+                filter(b1.c.b2.c.d2.between('d4d2', 'd5d2'))
+
+        q2 = sess.query(b1).\
+                filter(b1.c.d1.between('d3d1', 'd7d1')).\
+                filter(b1.c.b2.c.d2.between('d5d2', 'd6d2'))
+
+        eq_(
+            q1.union(q2).all(),
+            [(('d4d1', ('d4d2', 'd4d3')),), (('d5d1', ('d5d2', 'd5d3')),),
+                (('d6d1', ('d6d2', 'd6d3')),)]
+        )
+
+        # naming structure is preserved
+        row = q1.union(q2).first()
+        eq_(row.b1.d1, 'd4d1')
+        eq_(row.b1.b2.d2, 'd4d2')
+
+
+    def test_query_count(self):
+        Data = self.classes.Data
+        b1 = Bundle('b1', Data.d1, Data.d2)
+        eq_(Session().query(b1).count(), 10)
+
+    def test_join_relationship(self):
+        Data = self.classes.Data
+
+        sess = Session()
+        b1 = Bundle('b1', Data.d1, Data.d2)
+        q = sess.query(b1).join(Data.others)
+        self.assert_compile(q,
+            "SELECT data.d1 AS data_d1, data.d2 AS data_d2 FROM data "
+            "JOIN other ON data.id = other.data_id"
+        )
+
+    def test_join_selectable(self):
+        Data = self.classes.Data
+        Other = self.classes.Other
+
+        sess = Session()
+        b1 = Bundle('b1', Data.d1, Data.d2)
+        q = sess.query(b1).join(Other)
+        self.assert_compile(q,
+            "SELECT data.d1 AS data_d1, data.d2 AS data_d2 FROM data "
+            "JOIN other ON data.id = other.data_id"
+        )
+
+
+    def test_joins_from_adapted_entities(self):
+        Data = self.classes.Data
+
+        # test for #1853 in terms of bundles
+        # specifically this exercises adapt_to_selectable()
+
+        b1 = Bundle('b1', Data.id, Data.d1, Data.d2)
+
+        session = Session()
+        first = session.query(b1)
+        second = session.query(b1)
+        unioned = first.union(second)
+        subquery = session.query(Data.id).subquery()
+        joined = unioned.outerjoin(subquery, subquery.c.id == Data.id)
+        joined = joined.order_by(Data.id, Data.d1, Data.d2)
+
+        self.assert_compile(
+            joined,
+            "SELECT anon_1.data_id AS anon_1_data_id, anon_1.data_d1 AS anon_1_data_d1, "
+            "anon_1.data_d2 AS anon_1_data_d2 FROM "
+            "(SELECT data.id AS data_id, data.d1 AS data_d1, data.d2 AS data_d2 FROM "
+            "data UNION SELECT data.id AS data_id, data.d1 AS data_d1, "
+            "data.d2 AS data_d2 FROM data) AS anon_1 "
+            "LEFT OUTER JOIN (SELECT data.id AS id FROM data) AS anon_2 "
+            "ON anon_2.id = anon_1.data_id "
+            "ORDER BY anon_1.data_id, anon_1.data_d1, anon_1.data_d2")
+
+        # tuple nesting still occurs
+        eq_(
+            joined.all(),
+            [((1, 'd0d1', 'd0d2'),), ((2, 'd1d1', 'd1d2'),),
+            ((3, 'd2d1', 'd2d2'),), ((4, 'd3d1', 'd3d2'),),
+            ((5, 'd4d1', 'd4d2'),), ((6, 'd5d1', 'd5d2'),),
+            ((7, 'd6d1', 'd6d2'),), ((8, 'd7d1', 'd7d2'),),
+            ((9, 'd8d1', 'd8d2'),), ((10, 'd9d1', 'd9d2'),)]
+        )
+
+    def test_clause_expansion(self):
+        Data = self.classes.Data
+
+        b1 = Bundle('b1', Data.id, Data.d1, Data.d2)
+
+        sess = Session()
+        self.assert_compile(
+            sess.query(Data).order_by(b1),
+            "SELECT data.id AS data_id, data.d1 AS data_d1, "
+            "data.d2 AS data_d2, data.d3 AS data_d3 FROM data "
+            "ORDER BY data.id, data.d1, data.d2"
+        )
+
+        self.assert_compile(
+            sess.query(func.row_number().over(order_by=b1)),
+            "SELECT row_number() OVER (ORDER BY data.id, data.d1, data.d2) "
+            "AS anon_1 FROM data"
+        )
+
diff --git a/test/orm/test_composites.py b/test/orm/test_composites.py
index 5e7b91f3e..eabc9ca7b 100644
--- a/test/orm/test_composites.py
+++ b/test/orm/test_composites.py
@@ -214,17 +214,45 @@ class PointTest(fixtures.MappedTest):
             ((), [Point(x=None, y=None)], ())
         )
 
-    def test_query_cols(self):
+    def test_query_cols_legacy(self):
         Edge = self.classes.Edge
 
         sess = self._fixture()
 
         eq_(
-            sess.query(Edge.start, Edge.end).all(),
+            sess.query(Edge.start.clauses, Edge.end.clauses).all(),
             [(3, 4, 5, 6), (14, 5, 2, 7)]
         )
 
+    def test_query_cols(self):
+        Edge = self.classes.Edge
+        Point = self.classes.Point
+
+        sess = self._fixture()
+
+        start, end = Edge.start, Edge.end
+
+        eq_(
+            sess.query(start, end).filter(start == Point(3, 4)).all(),
+            [(Point(3, 4), Point(5, 6))]
+        )
+
+    def test_query_cols_labeled(self):
+        Edge = self.classes.Edge
+        Point = self.classes.Point
+
+        sess = self._fixture()
+
+        start, end = Edge.start, Edge.end
+
+        row = sess.query(start.label('s1'), end).filter(start == Point(3, 4)).first()
+        eq_(row.s1.x, 3)
+        eq_(row.s1.y, 4)
+        eq_(row.end.x, 5)
+        eq_(row.end.y, 6)
+
     def test_delete(self):
+        Point = self.classes.Point
         Graph, Edge = self.classes.Graph, self.classes.Edge
 
         sess = self._fixture()
@@ -235,7 +263,10 @@ class PointTest(fixtures.MappedTest):
         sess.flush()
         eq_(
             sess.query(Edge.start, Edge.end).all(),
-            [(3, 4, 5, 6), (14, 5, None, None)]
+            [
+                (Point(x=3, y=4), Point(x=5, y=6)),
+                (Point(x=14, y=5), Point(x=None, y=None))
+            ]
         )
 
     def test_save_null(self):
@@ -863,3 +894,15 @@ class ComparatorTest(fixtures.MappedTest, testing.AssertsCompiledSQL):
             "edge_1.x2, edge_1.y2"
         )
 
+    def test_clause_expansion(self):
+        self._fixture(False)
+        Edge = self.classes.Edge
+        from sqlalchemy.orm import configure_mappers
+        configure_mappers()
+
+        self.assert_compile(
+            select([Edge]).order_by(Edge.start),
+            "SELECT edge.id, edge.x1, edge.y1, edge.x2, edge.y2 FROM edge "
+            "ORDER BY edge.x1, edge.y1"
+        )
+