Raise for unexpected polymorphic identity

A query that is against an mapped inheritance subclass which also uses
:meth:`.Query.select_entity_from` or a similar technique in order  to
provide an existing subquery to SELECT from, will now raise an error if the
given subquery returns entities that do not correspond to the given
subclass, that is, they are sibling or superclasses in the same hierarchy.
Previously, these would be returned without error.  Additionally, if the
inheritance mapping is a single-inheritance mapping, the given subquery
must apply the appropriate filtering against the polymorphic discriminator
column in order to avoid this error; previously, the :class:`.Query` would
add this criteria to the outside query however this interferes with some
kinds of query that return other kinds of entities as well.

Fixes: #5122
Change-Id: I60cf8c1300d5bb279ad99f0f01fefe7e008a159b

1 files changed, 33 insertions, 7 deletions

diff --git a/lib/sqlalchemy/orm/loading.py b/lib/sqlalchemy/orm/loading.py
index b823de4ab..218449cdd 100644
--- a/lib/sqlalchemy/orm/loading.py
+++ b/lib/sqlalchemy/orm/loading.py
@@ -643,11 +643,9 @@ def _instance_processor(
                 identity_token,
             )
             if not is_not_primary_key(identitykey[1]):
-                raise sa_exc.InvalidRequestError(
-                    "Row with identity key %s can't be loaded into an "
-                    "object; the polymorphic discriminator column '%s' is "
-                    "NULL" % (identitykey, polymorphic_discriminator)
-                )
+                return identitykey
+            else:
+                return None
 
         _instance = _decorate_polymorphic_switch(
             _instance,
@@ -843,6 +841,8 @@ def _decorate_polymorphic_switch(
         else:
             if sub_mapper is mapper:
                 return None
+            elif not sub_mapper.isa(mapper):
+                return False
 
             return _instance_processor(
                 sub_mapper,
@@ -863,11 +863,37 @@ def _decorate_polymorphic_switch(
             _instance = polymorphic_instances[discriminator]
             if _instance:
                 return _instance(row)
+            elif _instance is False:
+                identitykey = ensure_no_pk(row)
+
+                if identitykey:
+                    raise sa_exc.InvalidRequestError(
+                        "Row with identity key %s can't be loaded into an "
+                        "object; the polymorphic discriminator column '%s' "
+                        "refers to %s, which is not a sub-mapper of "
+                        "the requested %s"
+                        % (
+                            identitykey,
+                            polymorphic_on,
+                            mapper.polymorphic_map[discriminator],
+                            mapper,
+                        )
+                    )
+                else:
+                    return None
             else:
                 return instance_fn(row)
         else:
-            ensure_no_pk(row)
-            return None
+            identitykey = ensure_no_pk(row)
+
+            if identitykey:
+                raise sa_exc.InvalidRequestError(
+                    "Row with identity key %s can't be loaded into an "
+                    "object; the polymorphic discriminator column '%s' is "
+                    "NULL" % (identitykey, polymorphic_on)
+                )
+            else:
+                return None
 
     return polymorphic_instance