- The :func:`.type_coerce` construct is now a fully fledged Core

expression element which is late-evaluated at compile time.  Previously,
the function was only a conversion function which would handle different
expression inputs by returning either a :class:`.Label` of a column-oriented
expression or a copy of a given :class:`.BindParameter` object,
which in particular prevented the operation from being logically
maintained when an ORM-level expression transformation would convert
a column to a bound parameter (e.g. for lazy loading).
fixes #3531

10 files changed, 373 insertions, 65 deletions

diff --git a/doc/build/changelog/changelog_11.rst b/doc/build/changelog/changelog_11.rst
index c910136fe..824623421 100644
--- a/doc/build/changelog/changelog_11.rst
+++ b/doc/build/changelog/changelog_11.rst
@@ -22,6 +22,23 @@
     :version: 1.1.0b1
 
     .. change::
+        :tags: bug, sql
+        :tickets: 3531
+
+        The :func:`.type_coerce` construct is now a fully fledged Core
+        expression element which is late-evaluated at compile time.  Previously,
+        the function was only a conversion function which would handle different
+        expression inputs by returning either a :class:`.Label` of a column-oriented
+        expression or a copy of a given :class:`.BindParameter` object,
+        which in particular prevented the operation from being logically
+        maintained when an ORM-level expression transformation would convert
+        a column to a bound parameter (e.g. for lazy loading).
+
+        .. seealso::
+
+            :ref:`change_3531`
+
+    .. change::
         :tags: bug, orm
         :tickets: 3526
 
diff --git a/doc/build/changelog/migration_11.rst b/doc/build/changelog/migration_11.rst
index ff26a00bd..367f20e62 100644
--- a/doc/build/changelog/migration_11.rst
+++ b/doc/build/changelog/migration_11.rst
@@ -435,6 +435,110 @@ can be done like any other type::
 
 :ticket:`2919`
 
+.. _change_3531:
+
+The type_coerce function is now a persistent SQL element
+--------------------------------------------------------
+
+The :func:`.expression.type_coerce` function previously would return
+an object either of type :class:`.BindParameter` or :class:`.Label`, depending
+on the input.  An effect this would have was that in the case where expression
+transformations were used, such as the conversion of an element from a
+:class:`.Column` to a :class:`.BindParameter` that's critical to ORM-level
+lazy loading, the type coercion information would not be used since it would
+have been lost already.
+
+To improve this behavior, the function now returns a persistent
+:class:`.TypeCoerce` container around the given expression, which itself
+remains unaffected; this construct is evaluated explicitly by the
+SQL compiler.  This allows for the coercion of the inner expression
+to be maintained no matter how the statement is modified, including if
+the contained element is replaced with a different one, as is common
+within the ORM's lazy loading feature.
+
+The test case illustrating the effect makes use of a heterogeneous
+primaryjoin condition in conjunction with custom types and lazy loading.
+Given a custom type that applies a CAST as a "bind expression"::
+
+    class StringAsInt(TypeDecorator):
+        impl = String
+
+        def column_expression(self, col):
+            return cast(col, Integer)
+
+        def bind_expression(self, value):
+            return cast(value, String)
+
+Then, a mapping where we are equating a string "id" column on one
+table to an integer "id" column on the other::
+
+    class Person(Base):
+        __tablename__ = 'person'
+        id = Column(StringAsInt, primary_key=True)
+
+        pets = relationship(
+            'Pets',
+            primaryjoin=(
+                'foreign(Pets.person_id)'
+                '==cast(type_coerce(Person.id, Integer), Integer)'
+            )
+        )
+
+    class Pets(Base):
+        __tablename__ = 'pets'
+        id = Column('id', Integer, primary_key=True)
+        person_id = Column('person_id', Integer)
+
+Above, in the :paramref:`.relationship.primaryjoin` expression, we are
+using :func:`.type_coerce` to handle bound parameters passed via
+lazyloading as integers, since we already know these will come from
+our ``StringAsInt`` type which maintains the value as an integer in
+Python. We are then using :func:`.cast` so that as a SQL expression,
+the VARCHAR "id"  column will be CAST to an integer for a regular non-
+converted join as with :meth:`.Query.join` or :func:`.orm.joinedload`.
+That is, a joinedload of ``.pets`` looks like::
+
+    SELECT person.id AS person_id, pets_1.id AS pets_1_id,
+           pets_1.person_id AS pets_1_person_id
+    FROM person
+    LEFT OUTER JOIN pets AS pets_1
+    ON pets_1.person_id = CAST(person.id AS INTEGER)
+
+Without the CAST in the ON clause of the join, strongly-typed databases
+such as Postgresql will refuse to implicitly compare the integer and fail.
+
+The lazyload case of ``.pets`` relies upon replacing
+the ``Person.id`` column at load time with a bound parameter, which receives
+a Python-loaded value.  This replacement is specifically where the intent
+of our :func:`.type_coerce` function would be lost.  Prior to the change,
+this lazy load comes out as::
+
+    SELECT pets.id AS pets_id, pets.person_id AS pets_person_id
+    FROM pets
+    WHERE pets.person_id = CAST(CAST(%(param_1)s AS VARCHAR) AS INTEGER)
+    {'param_1': 5}
+
+Where above, we see that our in-Python value of ``5`` is CAST first
+to a VARCHAR, then back to an INTEGER in SQL; a double CAST which works,
+but is nevertheless not what we asked for.
+
+With the change, the :func:`.type_coerce` function maintains a wrapper
+even after the column is swapped out for a bound parameter, and the query now
+looks like::
+
+    SELECT pets.id AS pets_id, pets.person_id AS pets_person_id
+    FROM pets
+    WHERE pets.person_id = CAST(%(param_1)s AS INTEGER)
+    {'param_1': 5}
+
+Where our outer CAST that's in our primaryjoin still takes effect, but the
+needless CAST that's in part of the ``StringAsInt`` custom type is removed
+as intended by the :func:`.type_coerce` function.
+
+
+:ticket:`3531`
+
+
 Key Behavioral Changes - ORM
 ============================
 
diff --git a/doc/build/core/sqlelement.rst b/doc/build/core/sqlelement.rst
index 30a6ed568..cf52a0166 100644
--- a/doc/build/core/sqlelement.rst
+++ b/doc/build/core/sqlelement.rst
@@ -141,6 +141,9 @@ used to construct any kind of typed SQL expression.
 .. autoclass:: sqlalchemy.sql.elements.True_
    :members:
 
+.. autoclass:: TypeCoerce
+   :members:
+
 .. autoclass:: sqlalchemy.sql.operators.custom_op
    :members:
 
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 52116a231..691195772 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -765,6 +765,9 @@ class SQLCompiler(Compiled):
         x += "END"
         return x
 
+    def visit_type_coerce(self, type_coerce, **kw):
+        return type_coerce.typed_expression._compiler_dispatch(self, **kw)
+
     def visit_cast(self, cast, **kwargs):
         return "CAST(%s AS %s)" % \
             (cast.clause._compiler_dispatch(self, **kwargs),
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 618b987e1..70046c66b 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -124,67 +124,6 @@ def literal(value, type_=None):
     return BindParameter(None, value, type_=type_, unique=True)
 
 
-def type_coerce(expression, type_):
-    """Associate a SQL expression with a particular type, without rendering
-    ``CAST``.
-
-    E.g.::
-
-        from sqlalchemy import type_coerce
-
-        stmt = select([type_coerce(log_table.date_string, StringDateTime())])
-
-    The above construct will produce SQL that is usually otherwise unaffected
-    by the :func:`.type_coerce` call::
-
-        SELECT date_string FROM log
-
-    However, when result rows are fetched, the ``StringDateTime`` type
-    will be applied to result rows on behalf of the ``date_string`` column.
-
-    A type that features bound-value handling will also have that behavior
-    take effect when literal values or :func:`.bindparam` constructs are
-    passed to :func:`.type_coerce` as targets.
-    For example, if a type implements the :meth:`.TypeEngine.bind_expression`
-    method or :meth:`.TypeEngine.bind_processor` method or equivalent,
-    these functions will take effect at statement compilation/execution time
-    when a literal value is passed, as in::
-
-        # bound-value handling of MyStringType will be applied to the
-        # literal value "some string"
-        stmt = select([type_coerce("some string", MyStringType)])
-
-    :func:`.type_coerce` is similar to the :func:`.cast` function,
-    except that it does not render the ``CAST`` expression in the resulting
-    statement.
-
-    :param expression: A SQL expression, such as a :class:`.ColumnElement`
-     expression or a Python string which will be coerced into a bound literal
-     value.
-
-    :param type_: A :class:`.TypeEngine` class or instance indicating
-     the type to which the expression is coerced.
-
-    .. seealso::
-
-        :func:`.cast`
-
-    """
-    type_ = type_api.to_instance(type_)
-
-    if hasattr(expression, '__clause_element__'):
-        return type_coerce(expression.__clause_element__(), type_)
-    elif isinstance(expression, BindParameter):
-        bp = expression._clone()
-        bp.type = type_
-        return bp
-    elif not isinstance(expression, Visitable):
-        if expression is None:
-            return Null()
-        else:
-            return literal(expression, type_=type_)
-    else:
-        return Label(None, expression, type_=type_)
 
 
 def outparam(key, type_=None):
@@ -2347,6 +2286,109 @@ class Cast(ColumnElement):
         return self.clause._from_objects
 
 
+class TypeCoerce(ColumnElement):
+    """Represent a Python-side type-coercion wrapper.
+
+    :class:`.TypeCoerce` supplies the :func:`.expression.type_coerce`
+    function; see that function for usage details.
+
+    .. versionchanged:: 1.1 The :func:`.type_coerce` function now produces
+       a persistent :class:`.TypeCoerce` wrapper object rather than
+       translating the given object in place.
+
+    .. seealso::
+
+        :func:`.expression.type_coerce`
+
+    """
+
+    __visit_name__ = 'type_coerce'
+
+    def __init__(self, expression, type_):
+        """Associate a SQL expression with a particular type, without rendering
+        ``CAST``.
+
+        E.g.::
+
+            from sqlalchemy import type_coerce
+
+            stmt = select([
+                type_coerce(log_table.date_string, StringDateTime())
+            ])
+
+        The above construct will produce a :class:`.TypeCoerce` object, which
+        renders SQL that labels the expression, but otherwise does not
+        modify its value on the SQL side::
+
+            SELECT date_string AS anon_1 FROM log
+
+        When result rows are fetched, the ``StringDateTime`` type
+        will be applied to result rows on behalf of the ``date_string`` column.
+        The rationale for the "anon_1" label is so that the type-coerced
+        column remains separate in the list of result columns vs. other
+        type-coerced or direct values of the target column.  In order to
+        provide a named label for the expression, use
+        :meth:`.ColumnElement.label`::
+
+            stmt = select([
+                type_coerce(
+                    log_table.date_string, StringDateTime()).label('date')
+            ])
+
+
+        A type that features bound-value handling will also have that behavior
+        take effect when literal values or :func:`.bindparam` constructs are
+        passed to :func:`.type_coerce` as targets.
+        For example, if a type implements the
+        :meth:`.TypeEngine.bind_expression`
+        method or :meth:`.TypeEngine.bind_processor` method or equivalent,
+        these functions will take effect at statement compilation/execution
+        time when a literal value is passed, as in::
+
+            # bound-value handling of MyStringType will be applied to the
+            # literal value "some string"
+            stmt = select([type_coerce("some string", MyStringType)])
+
+        :func:`.type_coerce` is similar to the :func:`.cast` function,
+        except that it does not render the ``CAST`` expression in the resulting
+        statement.
+
+        :param expression: A SQL expression, such as a :class:`.ColumnElement`
+         expression or a Python string which will be coerced into a bound
+         literal value.
+
+        :param type_: A :class:`.TypeEngine` class or instance indicating
+         the type to which the expression is coerced.
+
+        .. seealso::
+
+            :func:`.cast`
+
+        """
+        self.type = type_api.to_instance(type_)
+        self.clause = _literal_as_binds(expression, type_=self.type)
+
+    def _copy_internals(self, clone=_clone, **kw):
+        self.clause = clone(self.clause, **kw)
+        self.__dict__.pop('typed_expression', None)
+
+    def get_children(self, **kwargs):
+        return self.clause,
+
+    @property
+    def _from_objects(self):
+        return self.clause._from_objects
+
+    @util.memoized_property
+    def typed_expression(self):
+        if isinstance(self.clause, BindParameter):
+            bp = self.clause._clone()
+            bp.type = self.type
+            return bp
+        else:
+            return self.clause
+
+
 class Extract(ColumnElement):
     """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
 
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index 79d25a39e..27fae8ca4 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -36,7 +36,7 @@ from .elements import ClauseElement, ColumnElement,\
     True_, False_, BinaryExpression, Tuple, TypeClause, Extract, \
     Grouping, WithinGroup, not_, \
     collate, literal_column, between,\
-    literal, outparam, type_coerce, ClauseList, FunctionFilter
+    literal, outparam, TypeCoerce, ClauseList, FunctionFilter
 
 from .elements import SavepointClause, RollbackToSavepointClause, \
     ReleaseSavepointClause
@@ -92,6 +92,7 @@ asc = public_factory(UnaryExpression._create_asc, ".expression.asc")
 desc = public_factory(UnaryExpression._create_desc, ".expression.desc")
 distinct = public_factory(
     UnaryExpression._create_distinct, ".expression.distinct")
+type_coerce = public_factory(TypeCoerce, ".expression.type_coerce")
 true = public_factory(True_._instance, ".expression.true")
 false = public_factory(False_._instance, ".expression.false")
 null = public_factory(Null._instance, ".expression.null")
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 0c48ea8c2..4abb9b15a 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -13,7 +13,7 @@ import datetime as dt
 import codecs
 
 from .type_api import TypeEngine, TypeDecorator, to_instance
-from .elements import quoted_name, type_coerce, _defer_name
+from .elements import quoted_name, TypeCoerce as type_coerce, _defer_name
 from .. import exc, util, processors
 from .base import _bind_or_error, SchemaEventTarget
 from . import operators
diff --git a/test/orm/test_lazy_relations.py b/test/orm/test_lazy_relations.py
index ea39753b4..706f49d6f 100644
--- a/test/orm/test_lazy_relations.py
+++ b/test/orm/test_lazy_relations.py
@@ -1073,3 +1073,75 @@ class RefersToSelfLazyLoadInterferenceTest(fixtures.MappedTest):
         session.query(B).options(
             sa.orm.joinedload('parent').joinedload('zc')).all()
 
+
+class TypeCoerceTest(fixtures.MappedTest, testing.AssertsExecutionResults,):
+    """ORM-level test for [ticket:3531]"""
+
+    class StringAsInt(TypeDecorator):
+        impl = String
+
+        def column_expression(self, col):
+            return sa.cast(col, Integer)
+
+        def bind_expression(self, col):
+            return sa.cast(col, String)
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            'person', metadata,
+            Column("id", cls.StringAsInt, primary_key=True),
+        )
+        Table(
+            "pets", metadata,
+            Column("id", Integer, primary_key=True),
+            Column("person_id", cls.StringAsInt()),
+        )
+
+    @classmethod
+    def setup_classes(cls):
+        class Person(cls.Basic):
+            pass
+
+        class Pet(cls.Basic):
+            pass
+
+    @classmethod
+    def setup_mappers(cls):
+        mapper(cls.classes.Person, cls.tables.person, properties=dict(
+            pets=relationship(
+                cls.classes.Pet, primaryjoin=(
+                    orm.foreign(cls.tables.pets.c.person_id) ==
+                    sa.cast(
+                        sa.type_coerce(cls.tables.person.c.id, Integer),
+                        Integer
+                    )
+                )
+            )
+        ))
+
+        mapper(cls.classes.Pet, cls.tables.pets)
+
+    def test_lazyload_singlecast(self):
+        Person = self.classes.Person
+        Pet = self.classes.Pet
+
+        s = Session()
+        s.add_all([
+            Person(id=5), Pet(id=1, person_id=5)
+        ])
+        s.commit()
+
+        p1 = s.query(Person).first()
+
+        with self.sql_execution_asserter() as asserter:
+            p1.pets
+
+        asserter.assert_(
+            CompiledSQL(
+                "SELECT pets.id AS pets_id, CAST(pets.person_id AS INTEGER) "
+                "AS pets_person_id FROM pets "
+                "WHERE pets.person_id = CAST(:param_1 AS INTEGER)",
+                [{'param_1': 5}]
+            )
+        )
diff --git a/test/sql/test_compiler.py b/test/sql/test_compiler.py
index 7ff7d68af..c957b2f8a 100644
--- a/test/sql/test_compiler.py
+++ b/test/sql/test_compiler.py
@@ -3484,13 +3484,15 @@ class ResultMapTest(fixtures.TestBase):
         tc = type_coerce(t.c.a, String)
         stmt = select([t.c.a, l1, tc])
         comp = stmt.compile()
-        tc_anon_label = comp._create_result_map()['a_1'][1][0]
+        tc_anon_label = comp._create_result_map()['anon_1'][1][0]
         eq_(
             comp._create_result_map(),
             {
                 'a': ('a', (t.c.a, 'a', 'a'), t.c.a.type),
                 'bar': ('bar', (l1, 'bar'), l1.type),
-                'a_1': ('%%(%d a)s' % id(tc), (tc_anon_label, 'a_1'), tc.type),
+                'anon_1': (
+                    '%%(%d anon)s' % id(tc),
+                    (tc_anon_label, 'anon_1', tc), tc.type),
             },
         )
 
diff --git a/test/sql/test_types.py b/test/sql/test_types.py
index 288482392..f1fb611fb 100644
--- a/test/sql/test_types.py
+++ b/test/sql/test_types.py
@@ -12,6 +12,7 @@ from sqlalchemy import (
     BLOB, NCHAR, NVARCHAR, CLOB, TIME, DATE, DATETIME, TIMESTAMP, SMALLINT,
     INTEGER, DECIMAL, NUMERIC, FLOAT, REAL, Array)
 from sqlalchemy.sql import ddl
+from sqlalchemy.sql import visitors
 from sqlalchemy import inspection
 from sqlalchemy import exc, types, util, dialects
 for name in dialects.__all__:
@@ -789,6 +790,68 @@ class TypeCoerceCastTest(fixtures.TablesTest):
             [('BIND_INd1', 'BIND_INd1BIND_OUT')]
         )
 
+    def test_cast_replace_col_w_bind(self):
+        self._test_replace_col_w_bind(cast)
+
+    def test_type_coerce_replace_col_w_bind(self):
+        self._test_replace_col_w_bind(type_coerce)
+
+    def _test_replace_col_w_bind(self, coerce_fn):
+        MyType = self.MyType
+
+        t = self.tables.t
+        t.insert().values(data=coerce_fn('d1', MyType)).execute()
+
+        stmt = select([t.c.data, coerce_fn(t.c.data, MyType)])
+
+        def col_to_bind(col):
+            if col is t.c.data:
+                return bindparam(None, "x", type_=col.type, unique=True)
+            return None
+
+        # ensure we evaulate the expression so that we can see
+        # the clone resets this info
+        stmt.compile()
+
+        new_stmt = visitors.replacement_traverse(stmt, {}, col_to_bind)
+
+        # original statement
+        eq_(
+            testing.db.execute(stmt).fetchall(),
+            [('BIND_INd1', 'BIND_INd1BIND_OUT')]
+        )
+
+        # replaced with binds; CAST can't affect the bound parameter
+        # on the way in here
+        eq_(
+            testing.db.execute(new_stmt).fetchall(),
+            [('x', 'BIND_INxBIND_OUT')] if coerce_fn is type_coerce
+            else [('x', 'xBIND_OUT')]
+        )
+
+    def test_cast_bind(self):
+        self._test_bind(cast)
+
+    def test_type_bind(self):
+        self._test_bind(type_coerce)
+
+    def _test_bind(self, coerce_fn):
+        MyType = self.MyType
+
+        t = self.tables.t
+        t.insert().values(data=coerce_fn('d1', MyType)).execute()
+
+        stmt = select([
+            bindparam(None, "x", String(50), unique=True),
+            coerce_fn(bindparam(None, "x", String(50), unique=True), MyType)
+        ])
+
+        eq_(
+            testing.db.execute(stmt).fetchall(),
+            [('x', 'BIND_INxBIND_OUT')] if coerce_fn is type_coerce
+            else [('x', 'xBIND_OUT')]
+        )
+
     @testing.fails_on(
         "oracle", "ORA-00906: missing left parenthesis - "
         "seems to be CAST(:param AS type)")
@@ -822,6 +885,7 @@ class TypeCoerceCastTest(fixtures.TablesTest):
             [('BIND_INd1BIND_OUT', )])
 
 
+
 class VariantTest(fixtures.TestBase, AssertsCompiledSQL):
 
     def setup(self):