Add new "expanding" feature to bindparam()

Added a new kind of :func:`.bindparam` called "expanding".  This is
for use in ``IN`` expressions where the list of elements is rendered
into individual bound parameters at statement execution time, rather
than at statement compilation time.  This allows both a single bound
parameter name to be linked to an IN expression of multiple elements,
as well as allows query caching to be used with IN expressions.  The
new feature allows the related features of "select in" loading and
"polymorphic in" loading to make use of the baked query extension
to reduce call overhead.   This feature should be considered to be
**experimental** for 1.2.

Fixes: #3953
Change-Id: Ie708414a3ab9c0af29998a2c7f239ff7633b1f6e

3 files changed, 62 insertions, 12 deletions

diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index cc4248009..6da064797 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -350,6 +350,14 @@ class SQLCompiler(Compiled):
     columns with the table name (i.e. MySQL only)
     """
 
+    contains_expanding_parameters = False
+    """True if we've encountered bindparam(..., expanding=True).
+
+    These need to be converted before execution time against the
+    string statement.
+
+    """
+
     ansi_bind_rules = False
     """SQL 92 doesn't allow bind parameters to be used
     in the columns clause of a SELECT, nor does it allow
@@ -370,8 +378,14 @@ class SQLCompiler(Compiled):
     True unless using an unordered TextAsFrom.
     """
 
-    insert_prefetch = update_prefetch = ()
+    _numeric_binds = False
+    """
+    True if paramstyle is "numeric".  This paramstyle is trickier than
+    all the others.
 
+    """
+
+    insert_prefetch = update_prefetch = ()
 
     def __init__(self, dialect, statement, column_keys=None,
                  inline=False, **kwargs):
@@ -418,6 +432,7 @@ class SQLCompiler(Compiled):
         self.positional = dialect.positional
         if self.positional:
             self.positiontup = []
+            self._numeric_binds = dialect.paramstyle == "numeric"
         self.bindtemplate = BIND_TEMPLATES[dialect.paramstyle]
 
         self.ctes = None
@@ -439,7 +454,7 @@ class SQLCompiler(Compiled):
         ) and statement._returning:
             self.returning = statement._returning
 
-        if self.positional and dialect.paramstyle == 'numeric':
+        if self.positional and self._numeric_binds:
             self._apply_numbered_params()
 
     @property
@@ -492,7 +507,8 @@ class SQLCompiler(Compiled):
         return dict(
             (key, value) for key, value in
             ((self.bind_names[bindparam],
-              bindparam.type._cached_bind_processor(self.dialect))
+              bindparam.type._cached_bind_processor(self.dialect)
+              )
              for bindparam in self.bind_names)
             if value is not None
         )
@@ -1238,7 +1254,8 @@ class SQLCompiler(Compiled):
 
         self.binds[bindparam.key] = self.binds[name] = bindparam
 
-        return self.bindparam_string(name, **kwargs)
+        return self.bindparam_string(
+            name, expanding=bindparam.expanding, **kwargs)
 
     def render_literal_bindparam(self, bindparam, **kw):
         value = bindparam.effective_value
@@ -1300,13 +1317,18 @@ class SQLCompiler(Compiled):
         self.anon_map[derived] = anonymous_counter + 1
         return derived + "_" + str(anonymous_counter)
 
-    def bindparam_string(self, name, positional_names=None, **kw):
+    def bindparam_string(
+            self, name, positional_names=None, expanding=False, **kw):
         if self.positional:
             if positional_names is not None:
                 positional_names.append(name)
             else:
                 self.positiontup.append(name)
-        return self.bindtemplate % {'name': name}
+        if expanding:
+            self.contains_expanding_parameters = True
+            return "([EXPANDING_%s])" % name
+        else:
+            return self.bindtemplate % {'name': name}
 
     def visit_cte(self, cte, asfrom=False, ashint=False,
                   fromhints=None,
diff --git a/lib/sqlalchemy/sql/default_comparator.py b/lib/sqlalchemy/sql/default_comparator.py
index d409ebacc..4ba53ef75 100644
--- a/lib/sqlalchemy/sql/default_comparator.py
+++ b/lib/sqlalchemy/sql/default_comparator.py
@@ -127,10 +127,18 @@ def _in_impl(expr, op, seq_or_selectable, negate_op, **kw):
         return _boolean_compare(expr, op, seq_or_selectable,
                                 negate=negate_op, **kw)
     elif isinstance(seq_or_selectable, ClauseElement):
-        raise exc.InvalidRequestError(
-            'in_() accepts'
-            ' either a list of expressions '
-            'or a selectable: %r' % seq_or_selectable)
+        if isinstance(seq_or_selectable, BindParameter) and \
+                seq_or_selectable.expanding:
+            return _boolean_compare(
+                expr, op,
+                seq_or_selectable,
+                negate=negate_op)
+        else:
+            raise exc.InvalidRequestError(
+                'in_() accepts'
+                ' either a list of expressions, '
+                'a selectable, or an "expanding" bound parameter: %r'
+                % seq_or_selectable)
 
     # Handle non selectable arguments as sequences
     args = []
@@ -139,8 +147,8 @@ def _in_impl(expr, op, seq_or_selectable, negate_op, **kw):
             if not isinstance(o, operators.ColumnOperators):
                 raise exc.InvalidRequestError(
                     'in_() accepts'
-                    ' either a list of expressions '
-                    'or a selectable: %r' % o)
+                    ' either a list of expressions, '
+                    'a selectable, or an "expanding" bound parameter: %r' % o)
         elif o is None:
             o = Null()
         else:
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 001c3d042..414e3f477 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -867,6 +867,7 @@ class BindParameter(ColumnElement):
     def __init__(self, key, value=NO_ARG, type_=None,
                  unique=False, required=NO_ARG,
                  quote=None, callable_=None,
+                 expanding=False,
                  isoutparam=False,
                  _compared_to_operator=None,
                  _compared_to_type=None):
@@ -1052,6 +1053,23 @@ class BindParameter(ColumnElement):
           "OUT" parameter.  This applies to backends such as Oracle which
           support OUT parameters.
 
+        :param expanding:
+          if True, this parameter will be treated as an "expanding" parameter
+          at execution time; the parameter value is expected to be a sequence,
+          rather than a scalar value, and the string SQL statement will
+          be transformed on a per-execution basis to accomodate the sequence
+          with a variable number of parameter slots passed to the DBAPI.
+          This is to allow statement caching to be used in conjunction with
+          an IN clause.
+
+          .. note:: The "expanding" feature does not support "executemany"-
+             style parameter sets, nor does it support empty IN expressions.
+
+          .. note:: The "expanding" feature should be considered as
+             **experimental** within the 1.2 series.
+
+          .. versionadded:: 1.2
+
         .. seealso::
 
             :ref:`coretutorial_bind_param`
@@ -1093,6 +1111,8 @@ class BindParameter(ColumnElement):
         self.callable = callable_
         self.isoutparam = isoutparam
         self.required = required
+        self.expanding = expanding
+
         if type_ is None:
             if _compared_to_type is not None:
                 self.type = \