- The "wildcard" loader options, in particular the one set up by

the :func:`.orm.load_only` option to cover all attributes not
explicitly mentioned, now takes into account the superclasses
of a given entity, if that entity is mapped with inheritance mapping,
so that attribute names within the superclasses are also omitted
from the load.  Additionally, the polymorphic discriminator column
is unconditionally included in the list, just in the same way that
primary key columns are, so that even with load_only() set up,
polymorphic loading of subtypes continues to function correctly.
fixes #3287

doc/build/changelog/changelog_09.rst

@@ -14,6 +14,21 @@
 .. changelog::
     :version: 0.9.9
 
+    .. change::
+        :tags: bug, orm
+        :versions: 1.0.0
+        :tickets: 3287
+
+        The "wildcard" loader options, in particular the one set up by
+        the :func:`.orm.load_only` option to cover all attributes not
+        explicitly mentioned, now takes into account the superclasses
+        of a given entity, if that entity is mapped with inheritance mapping,
+        so that attribute names within the superclasses are also omitted
+        from the load.  Additionally, the polymorphic discriminator column
+        is unconditionally included in the list, just in the same way that
+        primary key columns are, so that even with load_only() set up,
+        polymorphic loading of subtypes continues to function correctly.
+
     .. change::
         :tags: bug, sql
         :versions: 1.0.0

lib/sqlalchemy/orm/mapper.py

@@ -2387,6 +2387,13 @@ class Mapper(InspectionAttr):
             collection.update(self._pks_by_table[table])
         return collection
 
+    @_memoized_configured_property
+    def _should_undefer_in_wildcard(self):
+        cols = set(self.primary_key)
+        if self.polymorphic_on is not None:
+            cols.add(self.polymorphic_on)
+        return cols
+
     @_memoized_configured_property
     def _primary_key_propkeys(self):
         return set([prop.key for prop in self._all_pk_props])

lib/sqlalchemy/orm/path_registry.py

@@ -52,6 +52,9 @@ class PathRegistry(object):
 
     """
 
+    is_token = False
+    is_root = False
+
     def __eq__(self, other):
         return other is not None and \
             self.path == other.path
@@ -153,6 +156,8 @@ class RootRegistry(PathRegistry):
     """
     path = ()
     has_entity = False
+    is_aliased_class = False
+    is_root = True
 
     def __getitem__(self, entity):
         return entity._path_registry
@@ -168,6 +173,15 @@ class TokenRegistry(PathRegistry):
 
     has_entity = False
 
+    is_token = True
+
+    def generate_for_superclasses(self):
+        if not self.parent.is_aliased_class and not self.parent.is_root:
+            for ent in self.parent.mapper.iterate_to_root():
+                yield TokenRegistry(self.parent.parent[ent], self.token)
+        else:
+            yield self
+
     def __getitem__(self, entity):
         raise NotImplementedError()
 

lib/sqlalchemy/orm/strategies.py

@@ -231,7 +231,8 @@ class DeferredColumnLoader(LoaderStrategy):
             (
                 loadopt and
                 'undefer_pks' in loadopt.local_opts and
-                set(self.columns).intersection(self.parent.primary_key)
+                set(self.columns).intersection(
+                    self.parent._should_undefer_in_wildcard)
             )
             or
             (

lib/sqlalchemy/orm/strategy_options.py

@@ -364,6 +364,7 @@ class _UnboundLoad(Load):
             return None
 
         token = start_path[0]
+
         if isinstance(token, util.string_types):
             entity = self._find_entity_basestring(query, token, raiseerr)
         elif isinstance(token, PropComparator):
@@ -407,10 +408,18 @@ class _UnboundLoad(Load):
         # prioritize "first class" options over those
         # that were "links in the chain", e.g. "x" and "y" in
         # someload("x.y.z") versus someload("x") / someload("x.y")
-        if self._is_chain_link:
-            effective_path.setdefault(context, "loader", loader)
+
+        if effective_path.is_token:
+            for path in effective_path.generate_for_superclasses():
+                if self._is_chain_link:
+                    path.setdefault(context, "loader", loader)
+                else:
+                    path.set(context, "loader", loader)
         else:
-            effective_path.set(context, "loader", loader)
+            if self._is_chain_link:
+                effective_path.setdefault(context, "loader", loader)
+            else:
+                effective_path.set(context, "loader", loader)
 
     def _find_entity_prop_comparator(self, query, token, mapper, raiseerr):
         if _is_aliased_class(mapper):

test/orm/test_deferred.py

@@ -2,10 +2,14 @@ import sqlalchemy as sa
 from sqlalchemy import testing, util
 from sqlalchemy.orm import mapper, deferred, defer, undefer, Load, \
     load_only, undefer_group, create_session, synonym, relationship, Session,\
-    joinedload, defaultload
+    joinedload, defaultload, aliased, contains_eager, with_polymorphic
 from sqlalchemy.testing import eq_, AssertsCompiledSQL, assert_raises_message
 from test.orm import _fixtures
-from sqlalchemy.orm import strategies
+
+
+from .inheritance._poly_fixtures import Company, Person, Engineer, Manager, \
+    Boss, Machine, Paperwork, _Polymorphic
+
 
 class DeferredTest(AssertsCompiledSQL, _fixtures.FixtureTest):
 
@@ -595,3 +599,128 @@ class DeferredOptionsTest(AssertsCompiledSQL, _fixtures.FixtureTest):
         )
 
 
+class InheritanceTest(_Polymorphic):
+    __dialect__ = 'default'
+
+    def test_load_only_subclass(self):
+        s = Session()
+        q = s.query(Manager).options(load_only("status", "manager_name"))
+        self.assert_compile(
+            q,
+            "SELECT managers.person_id AS managers_person_id, "
+            "people.person_id AS people_person_id, "
+            "people.type AS people_type, "
+            "managers.status AS managers_status, "
+            "managers.manager_name AS managers_manager_name "
+            "FROM people JOIN managers "
+            "ON people.person_id = managers.person_id "
+            "ORDER BY people.person_id"
+        )
+
+    def test_load_only_subclass_and_superclass(self):
+        s = Session()
+        q = s.query(Boss).options(load_only("status", "manager_name"))
+        self.assert_compile(
+            q,
+            "SELECT managers.person_id AS managers_person_id, "
+            "people.person_id AS people_person_id, "
+            "people.type AS people_type, "
+            "managers.status AS managers_status, "
+            "managers.manager_name AS managers_manager_name "
+            "FROM people JOIN managers "
+            "ON people.person_id = managers.person_id JOIN boss "
+            "ON managers.person_id = boss.boss_id ORDER BY people.person_id"
+        )
+
+    def test_load_only_alias_subclass(self):
+        s = Session()
+        m1 = aliased(Manager, flat=True)
+        q = s.query(m1).options(load_only("status", "manager_name"))
+        self.assert_compile(
+            q,
+            "SELECT managers_1.person_id AS managers_1_person_id, "
+            "people_1.person_id AS people_1_person_id, "
+            "people_1.type AS people_1_type, "
+            "managers_1.status AS managers_1_status, "
+            "managers_1.manager_name AS managers_1_manager_name "
+            "FROM people AS people_1 JOIN managers AS "
+            "managers_1 ON people_1.person_id = managers_1.person_id "
+            "ORDER BY people_1.person_id"
+        )
+
+    def test_load_only_subclass_from_relationship_polymorphic(self):
+        s = Session()
+        wp = with_polymorphic(Person, [Manager], flat=True)
+        q = s.query(Company).join(Company.employees.of_type(wp)).options(
+            contains_eager(Company.employees.of_type(wp)).
+            load_only(wp.Manager.status, wp.Manager.manager_name)
+        )
+        self.assert_compile(
+            q,
+            "SELECT people_1.person_id AS people_1_person_id, "
+            "people_1.type AS people_1_type, "
+            "managers_1.person_id AS managers_1_person_id, "
+            "managers_1.status AS managers_1_status, "
+            "managers_1.manager_name AS managers_1_manager_name, "
+            "companies.company_id AS companies_company_id, "
+            "companies.name AS companies_name "
+            "FROM companies JOIN (people AS people_1 LEFT OUTER JOIN "
+            "managers AS managers_1 ON people_1.person_id = "
+            "managers_1.person_id) ON companies.company_id = "
+            "people_1.company_id"
+        )
+
+    def test_load_only_subclass_from_relationship(self):
+        s = Session()
+        from sqlalchemy import inspect
+        inspect(Company).add_property("managers", relationship(Manager))
+        q = s.query(Company).join(Company.managers).options(
+            contains_eager(Company.managers).
+            load_only("status", "manager_name")
+        )
+        self.assert_compile(
+            q,
+            "SELECT companies.company_id AS companies_company_id, "
+            "companies.name AS companies_name, "
+            "managers.person_id AS managers_person_id, "
+            "people.person_id AS people_person_id, "
+            "people.type AS people_type, "
+            "managers.status AS managers_status, "
+            "managers.manager_name AS managers_manager_name "
+            "FROM companies JOIN (people JOIN managers ON people.person_id = "
+            "managers.person_id) ON companies.company_id = people.company_id"
+        )
+
+
+    def test_defer_on_wildcard_subclass(self):
+        # pretty much the same as load_only except doesn't
+        # exclude the primary key
+
+        s = Session()
+        q = s.query(Manager).options(
+            defer(".*"), undefer("status"))
+        self.assert_compile(
+            q,
+            "SELECT managers.status AS managers_status "
+            "FROM people JOIN managers ON "
+            "people.person_id = managers.person_id ORDER BY people.person_id"
+        )
+
+    def test_defer_super_name_on_subclass(self):
+        s = Session()
+        q = s.query(Manager).options(defer("name"))
+        self.assert_compile(
+            q,
+            "SELECT managers.person_id AS managers_person_id, "
+            "people.person_id AS people_person_id, "
+            "people.company_id AS people_company_id, "
+            "people.type AS people_type, managers.status AS managers_status, "
+            "managers.manager_name AS managers_manager_name "
+            "FROM people JOIN managers "
+            "ON people.person_id = managers.person_id "
+            "ORDER BY people.person_id"
+        )
+
+
+
+