Create a real type for Tuple() and handle appropriately in compiler

Improved the :func:`_sql.tuple_` construct such that it behaves predictably
when used in a columns-clause context.  The SQL tuple is not supported as a
"SELECT" columns clause element on most backends; on those that do
(PostgreSQL, not surprisingly), the Python DBAPI does not have a "nested
type" concept so there are still challenges in fetching rows for such an
object. Use of :func:`_sql.tuple_` in a :func:`_sql.select` or
:class:`_orm.Query` will now raise a :class:`_exc.CompileError` at the
point at which the :func:`_sql.tuple_` object is seen as presenting itself
for fetching rows (i.e., if the tuple is in the columns clause of a
subquery, no error is raised).  For ORM use,the :class:`_orm.Bundle` object
is an explicit directive that a series of columns should be returned as a
sub-tuple per row and is suggested by the error message. Additionally ,the
tuple will now render with parenthesis in all contexts. Previously, the
parenthesization would not render in a columns context leading to
non-defined behavior.

As part of this change, Tuple receives a dedicated datatype
which appears to allow us the very desirable change of removing
the bindparam._expanding_in_types attribute as well as
ClauseList._tuple_values (which might already have not been
needed due to #4645).

Fixes: #5127
Change-Id: Iecafa0e0aac2f1f37ec8d0e1631d562611c90200

9 files changed, 117 insertions, 56 deletions

diff --git a/lib/sqlalchemy/dialects/postgresql/array.py b/lib/sqlalchemy/dialects/postgresql/array.py
index 84fbd2e50..7fd271d52 100644
--- a/lib/sqlalchemy/dialects/postgresql/array.py
+++ b/lib/sqlalchemy/dialects/postgresql/array.py
@@ -9,8 +9,10 @@ import re
 
 from ... import types as sqltypes
 from ... import util
+from ...sql import coercions
 from ...sql import expression
 from ...sql import operators
+from ...sql import roles
 
 
 def Any(other, arrexpr, operator=operators.eq):
@@ -41,7 +43,7 @@ def All(other, arrexpr, operator=operators.eq):
     return arrexpr.all(other, operator)
 
 
-class array(expression.Tuple):
+class array(expression.ClauseList, expression.ColumnElement):
 
     """A PostgreSQL ARRAY literal.
 
@@ -97,16 +99,31 @@ class array(expression.Tuple):
     __visit_name__ = "array"
 
     def __init__(self, clauses, **kw):
+        clauses = [
+            coercions.expect(roles.ExpressionElementRole, c) for c in clauses
+        ]
+
         super(array, self).__init__(*clauses, **kw)
-        if isinstance(self.type, ARRAY):
+
+        self._type_tuple = [arg.type for arg in clauses]
+        main_type = kw.pop(
+            "type_",
+            self._type_tuple[0] if self._type_tuple else sqltypes.NULLTYPE,
+        )
+
+        if isinstance(main_type, ARRAY):
             self.type = ARRAY(
-                self.type.item_type,
-                dimensions=self.type.dimensions + 1
-                if self.type.dimensions is not None
+                main_type.item_type,
+                dimensions=main_type.dimensions + 1
+                if main_type.dimensions is not None
                 else 2,
             )
         else:
-            self.type = ARRAY(self.type)
+            self.type = ARRAY(main_type)
+
+    @property
+    def _select_iterable(self):
+        return (self,)
 
     def _bind_param(self, operator, obj, _assume_scalar=False, type_=None):
         if _assume_scalar or operator is operators.getitem:
diff --git a/lib/sqlalchemy/engine/default.py b/lib/sqlalchemy/engine/default.py
index 4fb20a3d5..ec0f2ed9f 100644
--- a/lib/sqlalchemy/engine/default.py
+++ b/lib/sqlalchemy/engine/default.py
@@ -1456,8 +1456,8 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
                     continue
 
                 if key in self._expanded_parameters:
-                    if bindparam._expanding_in_types:
-                        num = len(bindparam._expanding_in_types)
+                    if bindparam.type._is_tuple_type:
+                        num = len(bindparam.type.types)
                         dbtypes = inputsizes[bindparam]
                         positional_inputsizes.extend(
                             [
@@ -1488,8 +1488,8 @@ class DefaultExecutionContext(interfaces.ExecutionContext):
                     continue
 
                 if key in self._expanded_parameters:
-                    if bindparam._expanding_in_types:
-                        num = len(bindparam._expanding_in_types)
+                    if bindparam.type._is_tuple_type:
+                        num = len(bindparam.type.types)
                         dbtypes = inputsizes[bindparam]
                         keyword_inputsizes.update(
                             [
diff --git a/lib/sqlalchemy/orm/evaluator.py b/lib/sqlalchemy/orm/evaluator.py
index caa9ffe10..21d5e72d4 100644
--- a/lib/sqlalchemy/orm/evaluator.py
+++ b/lib/sqlalchemy/orm/evaluator.py
@@ -113,6 +113,9 @@ class EvaluatorCompiler(object):
         get_corresponding_attr = operator.attrgetter(key)
         return lambda obj: get_corresponding_attr(obj)
 
+    def visit_tuple(self, clause):
+        return self.visit_clauselist(clause)
+
     def visit_clauselist(self, clause):
         evaluators = list(map(self.process, clause.clauses))
         if clause.operator is operators.or_:
diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index fa0f9c435..b3a38f802 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -478,7 +478,6 @@ class InElementImpl(RoleImpl):
 
             if non_literal_expressions:
                 return elements.ClauseList(
-                    _tuple_values=isinstance(expr, elements.Tuple),
                     *[
                         non_literal_expressions[o]
                         if o in non_literal_expressions
@@ -509,10 +508,6 @@ class InElementImpl(RoleImpl):
                 # param to IN? check for ARRAY type?
                 element = element._clone(maintain_key=True)
                 element.expanding = True
-            if isinstance(expr, elements.Tuple):
-                element = element._with_expanding_in_types(
-                    [elem.type for elem in expr]
-                )
 
             return element
         else:
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index 542bf58ac..4f4cf7f8b 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -848,10 +848,10 @@ class SQLCompiler(Compiled):
                 (
                     self.bind_names[bindparam],
                     bindparam.type._cached_bind_processor(self.dialect)
-                    if not bindparam._expanding_in_types
+                    if not bindparam.type._is_tuple_type
                     else tuple(
                         elem_type._cached_bind_processor(self.dialect)
-                        for elem_type in bindparam._expanding_in_types
+                        for elem_type in bindparam.type.types
                     ),
                 )
                 for bindparam in self.bind_names
@@ -1018,9 +1018,9 @@ class SQLCompiler(Compiled):
             if bindparam in literal_execute_params:
                 continue
 
-            if bindparam._expanding_in_types:
+            if bindparam.type._is_tuple_type:
                 inputsizes[bindparam] = [
-                    _lookup_type(typ) for typ in bindparam._expanding_in_types
+                    _lookup_type(typ) for typ in bindparam.type.types
                 ]
             else:
                 inputsizes[bindparam] = _lookup_type(bindparam.type)
@@ -1107,7 +1107,7 @@ class SQLCompiler(Compiled):
 
                 if not parameter.literal_execute:
                     parameters.update(to_update)
-                    if parameter._expanding_in_types:
+                    if parameter.type._is_tuple_type:
                         new_processors.update(
                             (
                                 "%s_%s_%s" % (name, i, j),
@@ -1541,6 +1541,9 @@ class SQLCompiler(Compiled):
                 if s
             )
 
+    def visit_tuple(self, clauselist, **kw):
+        return "(%s)" % self.visit_clauselist(clauselist, **kw)
+
     def visit_clauselist(self, clauselist, **kw):
         sep = clauselist.operator
         if sep is None:
@@ -1548,10 +1551,7 @@ class SQLCompiler(Compiled):
         else:
             sep = OPERATORS[clauselist.operator]
 
-        text = self._generate_delimited_list(clauselist.clauses, sep, **kw)
-        if clauselist._tuple_values and self.dialect.tuple_in_values:
-            text = "VALUES " + text
-        return text
+        return self._generate_delimited_list(clauselist.clauses, sep, **kw)
 
     def visit_case(self, clause, **kwargs):
         x = "CASE "
@@ -1824,27 +1824,31 @@ class SQLCompiler(Compiled):
     def _literal_execute_expanding_parameter_literal_binds(
         self, parameter, values
     ):
+
         if not values:
+            assert not parameter.type._is_tuple_type
             replacement_expression = self.visit_empty_set_expr(
-                parameter._expanding_in_types
-                if parameter._expanding_in_types
-                else [parameter.type]
+                [parameter.type]
             )
 
         elif isinstance(values[0], (tuple, list)):
+            assert parameter.type._is_tuple_type
             replacement_expression = (
                 "VALUES " if self.dialect.tuple_in_values else ""
             ) + ", ".join(
                 "(%s)"
                 % (
                     ", ".join(
-                        self.render_literal_value(value, parameter.type)
-                        for value in tuple_element
+                        self.render_literal_value(value, param_type)
+                        for value, param_type in zip(
+                            tuple_element, parameter.type.types
+                        )
                     )
                 )
                 for i, tuple_element in enumerate(values)
             )
         else:
+            assert not parameter.type._is_tuple_type
             replacement_expression = ", ".join(
                 self.render_literal_value(value, parameter.type)
                 for value in values
@@ -1853,6 +1857,7 @@ class SQLCompiler(Compiled):
         return (), replacement_expression
 
     def _literal_execute_expanding_parameter(self, name, parameter, values):
+
         if parameter.literal_execute:
             return self._literal_execute_expanding_parameter_literal_binds(
                 parameter, values
@@ -1860,11 +1865,15 @@ class SQLCompiler(Compiled):
 
         if not values:
             to_update = []
-            replacement_expression = self.visit_empty_set_expr(
-                parameter._expanding_in_types
-                if parameter._expanding_in_types
-                else [parameter.type]
-            )
+            if parameter.type._is_tuple_type:
+
+                replacement_expression = self.visit_empty_set_expr(
+                    parameter.type.types
+                )
+            else:
+                replacement_expression = self.visit_empty_set_expr(
+                    [parameter.type]
+                )
 
         elif isinstance(values[0], (tuple, list)):
             to_update = [
@@ -2560,6 +2569,12 @@ class SQLCompiler(Compiled):
         if keyname is None:
             self._ordered_columns = False
             self._textual_ordered_columns = True
+        if type_._is_tuple_type:
+            raise exc.CompileError(
+                "Most backends don't support SELECTing "
+                "from a tuple() object.  If this is an ORM query, "
+                "consider using the Bundle object."
+            )
         self._result_columns.append((keyname, name, objects, type_))
 
     def _label_select_column(
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 8a506446d..a17612034 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -1059,7 +1059,6 @@ class BindParameter(roles.InElementRole, ColumnElement):
     ]
 
     _is_crud = False
-    _expanding_in_types = ()
     _is_bind_parameter = True
     _key_is_anon = False
 
@@ -1372,15 +1371,6 @@ class BindParameter(roles.InElementRole, ColumnElement):
         else:
             self.type = type_
 
-    def _with_expanding_in_types(self, types):
-        """Return a copy of this :class:`.BindParameter` in
-        the context of an expanding IN against a tuple.
-
-        """
-        cloned = self._clone(maintain_key=True)
-        cloned._expanding_in_types = types
-        return cloned
-
     def _with_value(self, value, maintain_key=False):
         """Return a copy of this :class:`.BindParameter` with the given value
         set.
@@ -2141,7 +2131,6 @@ class ClauseList(
         self.group_contents = kwargs.pop("group_contents", True)
         if kwargs.pop("_flatten_sub_clauses", False):
             clauses = util.flatten_iterator(clauses)
-        self._tuple_values = kwargs.pop("_tuple_values", False)
         self._text_converter_role = text_converter_role = kwargs.pop(
             "_literal_as_text_role", roles.WhereHavingRole
         )
@@ -2168,7 +2157,6 @@ class ClauseList(
         self.group = True
         self.operator = operator
         self.group_contents = True
-        self._tuple_values = False
         self._is_implicitly_boolean = False
         return self
 
@@ -2212,8 +2200,6 @@ class BooleanClauseList(ClauseList, ColumnElement):
     __visit_name__ = "clauselist"
     inherit_cache = True
 
-    _tuple_values = False
-
     def __init__(self, *arg, **kw):
         raise NotImplementedError(
             "BooleanClauseList has a private constructor"
@@ -2471,8 +2457,11 @@ or_ = BooleanClauseList.or_
 class Tuple(ClauseList, ColumnElement):
     """Represent a SQL tuple."""
 
+    __visit_name__ = "tuple"
+
     _traverse_internals = ClauseList._traverse_internals + []
 
+    @util.preload_module("sqlalchemy.sql.sqltypes")
     def __init__(self, *clauses, **kw):
         """Return a :class:`.Tuple`.
 
@@ -2496,15 +2485,12 @@ class Tuple(ClauseList, ColumnElement):
             invoked.
 
         """
+        sqltypes = util.preloaded.sql_sqltypes
 
         clauses = [
             coercions.expect(roles.ExpressionElementRole, c) for c in clauses
         ]
-        self._type_tuple = [arg.type for arg in clauses]
-        self.type = kw.pop(
-            "type_",
-            self._type_tuple[0] if self._type_tuple else type_api.NULLTYPE,
-        )
+        self.type = sqltypes.TupleType(*[arg.type for arg in clauses])
 
         super(Tuple, self).__init__(*clauses, **kw)
 
@@ -2520,7 +2506,8 @@ class Tuple(ClauseList, ColumnElement):
                 _compared_to_operator=operator,
                 unique=True,
                 expanding=True,
-            )._with_expanding_in_types(self._type_tuple)
+                type_=self.type,
+            )
         else:
             return Tuple(
                 *[
@@ -2532,9 +2519,13 @@ class Tuple(ClauseList, ColumnElement):
                         unique=True,
                         type_=type_,
                     )
-                    for o, compared_to_type in zip(obj, self._type_tuple)
+                    for o, compared_to_type in zip(obj, self.type.types)
                 ]
-            ).self_group()
+            )
+
+    def self_group(self, against=None):
+        # Tuple is parenthsized by definition.
+        return self
 
 
 class Case(ColumnElement):
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index 64663a6b0..d29f23361 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -2801,6 +2801,21 @@ class ARRAY(SchemaEventTarget, Indexable, Concatenable, TypeEngine):
             self.item_type._set_parent_with_dispatch(parent)
 
 
+class TupleType(TypeEngine):
+    """represent the composite type of a Tuple."""
+
+    _is_tuple_type = True
+
+    def __init__(self, *types):
+        self.types = types
+
+    def result_processor(self, dialect, coltype):
+        raise NotImplementedError(
+            "The tuple type does not support being fetched "
+            "as a column in a result row."
+        )
+
+
 class REAL(Float):
 
     """The SQL REAL type."""
diff --git a/lib/sqlalchemy/sql/type_api.py b/lib/sqlalchemy/sql/type_api.py
index 1284ef515..0da88dc54 100644
--- a/lib/sqlalchemy/sql/type_api.py
+++ b/lib/sqlalchemy/sql/type_api.py
@@ -45,6 +45,7 @@ class TypeEngine(Traversible):
 
     _sqla_type = True
     _isnull = False
+    _is_tuple_type = False
 
     class Comparator(operators.ColumnOperators):
         """Base class for custom comparison operations defined at the
diff --git a/lib/sqlalchemy/testing/suite/test_select.py b/lib/sqlalchemy/testing/suite/test_select.py
index 9fb481676..cff1f2cfc 100644
--- a/lib/sqlalchemy/testing/suite/test_select.py
+++ b/lib/sqlalchemy/testing/suite/test_select.py
@@ -665,6 +665,30 @@ class PostCompileParamsTest(
             )
         )
 
+    @testing.requires.tuple_in
+    def test_execute_tuple_expanding_plus_literal_heterogeneous_execute(self):
+        table = self.tables.some_table
+
+        stmt = select([table.c.id]).where(
+            tuple_(table.c.x, table.c.z).in_(
+                bindparam("q", expanding=True, literal_execute=True)
+            )
+        )
+
+        with self.sql_execution_asserter() as asserter:
+            with config.db.connect() as conn:
+                conn.execute(stmt, q=[(5, "z1"), (12, "z3")])
+
+        asserter.assert_(
+            CursorSQL(
+                "SELECT some_table.id \nFROM some_table "
+                "\nWHERE (some_table.x, some_table.z) "
+                "IN (%s(5, 'z1'), (12, 'z3'))"
+                % ("VALUES " if config.db.dialect.tuple_in_values else ""),
+                () if config.db.dialect.positional else {},
+            )
+        )
+
 
 class ExpandingBoundInTest(fixtures.TablesTest):
     __backend__ = True