Merge "Implement SQL VALUES in core."

8 files changed, 521 insertions, 0 deletions

diff --git a/doc/build/changelog/unreleased_14/4868.rst b/doc/build/changelog/unreleased_14/4868.rst
new file mode 100644
index 000000000..49a79b7bf
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/4868.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: change, sql
+    :tickets: 4868
+
+    Added a core :class:`Values` object that enables a VALUES construct
+    to be used in the FROM clause of an SQL statement for databases that
+    support it (mainly PostgreSQL and SQL Server).
diff --git a/lib/sqlalchemy/__init__.py b/lib/sqlalchemy/__init__.py
index 0f18aba33..f7b9214aa 100644
--- a/lib/sqlalchemy/__init__.py
+++ b/lib/sqlalchemy/__init__.py
@@ -76,6 +76,7 @@ from .sql import type_coerce  # noqa
 from .sql import union  # noqa
 from .sql import union_all  # noqa
 from .sql import update  # noqa
+from .sql import values  # noqa
 from .sql import within_group  # noqa
 from .types import ARRAY  # noqa
 from .types import BIGINT  # noqa
diff --git a/lib/sqlalchemy/sql/__init__.py b/lib/sqlalchemy/sql/__init__.py
index 281b7d0f2..78de80734 100644
--- a/lib/sqlalchemy/sql/__init__.py
+++ b/lib/sqlalchemy/sql/__init__.py
@@ -77,6 +77,8 @@ from .expression import union  # noqa
 from .expression import union_all  # noqa
 from .expression import Update  # noqa
 from .expression import update  # noqa
+from .expression import Values  # noqa
+from .expression import values  # noqa
 from .expression import within_group  # noqa
 from .visitors import ClauseVisitor  # noqa
 
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index ae9c3c73a..14f4bda8c 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -2292,6 +2292,46 @@ class SQLCompiler(Compiled):
 
         return text
 
+    def visit_values(self, element, asfrom=False, from_linter=None, **kw):
+        v = "VALUES %s" % ", ".join(
+            self.process(elem, literal_binds=element.literal_binds)
+            for elem in element._data
+        )
+
+        if isinstance(element.name, elements._truncated_label):
+            name = self._truncated_identifier("values", element.name)
+        else:
+            name = element.name
+
+        if element._is_lateral:
+            lateral = "LATERAL "
+        else:
+            lateral = ""
+
+        if asfrom:
+            if from_linter:
+                from_linter.froms[element] = (
+                    name if name is not None else "(unnamed VALUES element)"
+                )
+
+            if name:
+                v = "%s(%s)%s (%s)" % (
+                    lateral,
+                    v,
+                    self.get_render_as_alias_suffix(self.preparer.quote(name)),
+                    (
+                        ", ".join(
+                            c._compiler_dispatch(
+                                self, include_table=False, **kw
+                            )
+                            for c in element.columns
+                        )
+                    ),
+                )
+            else:
+                v = "%s(%s)" % (lateral, v)
+        return v
+
     def get_render_as_alias_suffix(self, alias_name_text):
         return " AS " + alias_name_text
 
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index 780648df0..4dc2b8bbf 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -32,6 +32,7 @@ __all__ = [
     "Selectable",
     "TableClause",
     "Update",
+    "Values",
     "alias",
     "and_",
     "asc",
@@ -80,6 +81,7 @@ __all__ = [
     "Subquery",
     "TableSample",
     "tablesample",
+    "values",
 ]
 
 
@@ -156,6 +158,7 @@ from .selectable import TableClause  # noqa
 from .selectable import TableSample  # noqa
 from .selectable import TextAsFrom  # noqa
 from .selectable import TextualSelect  # noqa
+from .selectable import Values  # noqa
 from .visitors import Visitable  # noqa
 from ..util.langhelpers import public_factory  # noqa
 
@@ -184,6 +187,7 @@ label = public_factory(Label, ".sql.expression.label")
 case = public_factory(Case, ".sql.expression.case")
 cast = public_factory(Cast, ".sql.expression.cast")
 cte = public_factory(CTE._factory, ".sql.expression.cte")
+values = public_factory(Values, ".sql.expression.values")
 extract = public_factory(Extract, ".sql.expression.extract")
 tuple_ = public_factory(Tuple, ".sql.expression.tuple_")
 except_ = public_factory(
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 3c8990a84..a44b079c4 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -46,6 +46,7 @@ from .elements import ColumnClause
 from .elements import GroupedElement
 from .elements import Grouping
 from .elements import literal_column
+from .elements import Tuple
 from .elements import UnaryExpression
 from .visitors import InternalTraversal
 from .. import exc
@@ -2010,6 +2011,127 @@ class ForUpdateArg(ClauseElement):
             self.of = None
 
 
+class Values(Generative, FromClause):
+    """represent a ``VALUES`` construct that can be used as a FROM element
+    in a statement.
+
+    The :class:`.Values` object is created from the
+    :func:`~.sql.expression.values` function.
+
+    .. versionadded:: 1.4
+
+    """
+
+    named_with_column = True
+    __visit_name__ = "values"
+
+    _data = ()
+
+    _traverse_internals = [
+        ("_column_args", InternalTraversal.dp_clauseelement_list,),
+        ("_data", InternalTraversal.dp_clauseelement_list),
+        ("name", InternalTraversal.dp_string),
+        ("literal_binds", InternalTraversal.dp_boolean),
+    ]
+
+    def __init__(self, *columns, **kw):
+        r"""Construct a :class:`.Values` construct.
+
+        The column expressions and the actual data for
+        :class:`.Values` are given in two separate steps.  The
+        constructor receives the column expressions typically as
+        :func:`.column` constructs, and the data is then passed via the
+        :meth:`.Values.data` method as a list, which can be called multiple
+        times to add more data, e.g.::
+
+            from sqlalchemy import column
+            from sqlalchemy import values
+
+            value_expr = values(
+                column('id', Integer),
+                column('name', Integer),
+                name="my_values"
+            ).data(
+                [(1, 'name1'), (2, 'name2'), (3, 'name3')]
+            )
+
+        :param \*columns: column expressions, typically composed using
+         :func:`.column` objects.
+
+        :param name: the name for this VALUES construct.  If omitted, the
+         VALUES construct will be unnamed in a SQL expression.   Different
+         backends may have different requirements here.
+
+        :param literal_binds: Defaults to False.  Whether or not to render
+         the data values inline in the SQL output, rather than using bound
+         parameters.
+
+        """
+
+        super(Values, self).__init__()
+        self._column_args = columns
+        self.name = kw.pop("name", None)
+        self.literal_binds = kw.pop("literal_binds", False)
+        self.named_with_column = self.name is not None
+
+    @_generative
+    def alias(self, name, **kw):
+        """Return a new :class:`.Values` construct that is a copy of this
+        one with the given name.
+
+        This method is a VALUES-specific specialization of the
+        :class:`.FromClause.alias` method.
+
+        .. seealso::
+
+            :ref:`core_tutorial_aliases`
+
+            :func:`~.expression.alias`
+
+        """
+        self.name = name
+        self.named_with_column = self.name is not None
+
+    @_generative
+    def lateral(self, name=None):
+        """Return a new :class:`.Values` with the lateral flag set, so that
+        it renders as LATERAL.
+
+        .. seealso::
+
+            :func:`~.expression.lateral`
+
+        """
+        self._is_lateral = True
+        if name is not None:
+            self.name = name
+
+    @_generative
+    def data(self, values):
+        """Return a new :class:`.Values` construct, adding the given data
+        to the data list.
+
+        E.g.::
+
+            my_values = my_values.data([(1, 'value 1'), (2, 'value2')])
+
+        :param values: a sequence (i.e. list) of tuples that map to the
+         column expressions given in the :class:`.Values` constructor.
+
+        """
+
+        self._data += tuple(Tuple(*row).self_group() for row in values)
+
+    def _populate_column_collection(self):
+        for c in self._column_args:
+            self._columns.add(c)
+            c.table = self
+
+    @property
+    def _from_objects(self):
+        return [self]
+
+
 class SelectBase(
     roles.SelectStatementRole,
     roles.DMLSelectRole,
diff --git a/test/sql/test_compare.py b/test/sql/test_compare.py
index 185244094..b5fad54dc 100644
--- a/test/sql/test_compare.py
+++ b/test/sql/test_compare.py
@@ -25,6 +25,7 @@ from sqlalchemy import tuple_
 from sqlalchemy import union
 from sqlalchemy import union_all
 from sqlalchemy import util
+from sqlalchemy import values
 from sqlalchemy.dialects import mysql
 from sqlalchemy.dialects import postgresql
 from sqlalchemy.schema import Sequence
@@ -471,6 +472,43 @@ class CoreFixtures(object):
             table("a", column("q"), column("y", Integer)),
         ),
         lambda: (table_a, table_b),
+        lambda: (
+            values(
+                column("mykey", Integer),
+                column("mytext", String),
+                column("myint", Integer),
+                name="myvalues",
+            ).data([(1, "textA", 99), (2, "textB", 88)]),
+            values(
+                column("mykey", Integer),
+                column("mytext", String),
+                column("myint", Integer),
+                name="myothervalues",
+            ).data([(1, "textA", 99), (2, "textB", 88)]),
+            values(
+                column("mykey", Integer),
+                column("mytext", String),
+                column("myint", Integer),
+                name="myvalues",
+            ).data([(1, "textA", 89), (2, "textG", 88)]),
+            values(
+                column("mykey", Integer),
+                column("mynottext", String),
+                column("myint", Integer),
+                name="myvalues",
+            ).data([(1, "textA", 99), (2, "textB", 88)]),
+            # TODO: difference in type
+            # values(
+            #    [
+            #        column("mykey", Integer),
+            #        column("mytext", Text),
+            #        column("myint", Integer),
+            #    ],
+            #    (1, "textA", 99),
+            #    (2, "textB", 88),
+            #    alias_name="myvalues",
+            # ),
+        ),
     ]
 
     dont_compare_values_fixtures = [
diff --git a/test/sql/test_values.py b/test/sql/test_values.py
new file mode 100644
index 000000000..154701ea0
--- /dev/null
+++ b/test/sql/test_values.py
@@ -0,0 +1,307 @@
+from sqlalchemy import alias
+from sqlalchemy import Column
+from sqlalchemy import column
+from sqlalchemy import ForeignKey
+from sqlalchemy import Integer
+from sqlalchemy import String
+from sqlalchemy import Table
+from sqlalchemy import testing
+from sqlalchemy import true
+from sqlalchemy.engine import default
+from sqlalchemy.sql import select
+from sqlalchemy.sql import Values
+from sqlalchemy.sql.compiler import FROM_LINTING
+from sqlalchemy.testing import AssertsCompiledSQL
+from sqlalchemy.testing import fixtures
+
+
+class ValuesTest(fixtures.TablesTest, AssertsCompiledSQL):
+    __dialect__ = default.DefaultDialect(supports_native_boolean=True)
+
+    run_setup_bind = None
+
+    run_create_tables = None
+
+    @classmethod
+    def define_tables(cls, metadata):
+        Table(
+            "people",
+            metadata,
+            Column("people_id", Integer, primary_key=True),
+            Column("age", Integer),
+            Column("name", String(30)),
+        )
+        Table(
+            "bookcases",
+            metadata,
+            Column("bookcase_id", Integer, primary_key=True),
+            Column(
+                "bookcase_owner_id", Integer, ForeignKey("people.people_id")
+            ),
+            Column("bookcase_shelves", Integer),
+            Column("bookcase_width", Integer),
+        )
+        Table(
+            "books",
+            metadata,
+            Column("book_id", Integer, primary_key=True),
+            Column(
+                "bookcase_id", Integer, ForeignKey("bookcases.bookcase_id")
+            ),
+            Column("book_owner_id", Integer, ForeignKey("people.people_id")),
+            Column("book_weight", Integer),
+        )
+
+    def test_column_quoting(self):
+        v1 = Values(
+            column("CaseSensitive", Integer),
+            column("has spaces", String),
+            name="Spaces and Cases",
+        ).data([(1, "textA", 99), (2, "textB", 88)])
+        self.assert_compile(
+            select([v1]),
+            'SELECT "Spaces and Cases"."CaseSensitive", '
+            '"Spaces and Cases"."has spaces" FROM '
+            "(VALUES (:param_1, :param_2, :param_3), "
+            "(:param_4, :param_5, :param_6)) "
+            'AS "Spaces and Cases" ("CaseSensitive", "has spaces")',
+        )
+
+    @testing.fixture
+    def literal_parameter_fixture(self):
+        def go(literal_binds):
+            return Values(
+                column("mykey", Integer),
+                column("mytext", String),
+                column("myint", Integer),
+                name="myvalues",
+                literal_binds=literal_binds,
+            ).data([(1, "textA", 99), (2, "textB", 88)])
+
+        return go
+
+    def test_bound_parameters(self, literal_parameter_fixture):
+        literal_parameter_fixture = literal_parameter_fixture(False)
+
+        stmt = select([literal_parameter_fixture])
+
+        self.assert_compile(
+            stmt,
+            "SELECT myvalues.mykey, myvalues.mytext, myvalues.myint FROM "
+            "(VALUES (:param_1, :param_2, :param_3), "
+            "(:param_4, :param_5, :param_6)"
+            ") AS myvalues (mykey, mytext, myint)",
+            checkparams={
+                "param_1": 1,
+                "param_2": "textA",
+                "param_3": 99,
+                "param_4": 2,
+                "param_5": "textB",
+                "param_6": 88,
+            },
+        )
+
+    def test_literal_parameters(self, literal_parameter_fixture):
+        literal_parameter_fixture = literal_parameter_fixture(True)
+
+        stmt = select([literal_parameter_fixture])
+
+        self.assert_compile(
+            stmt,
+            "SELECT myvalues.mykey, myvalues.mytext, myvalues.myint FROM "
+            "(VALUES (1, 'textA', 99), (2, 'textB', 88)"
+            ") AS myvalues (mykey, mytext, myint)",
+            checkparams={},
+        )
+
+    def test_with_join_unnamed(self):
+        people = self.tables.people
+        values = Values(
+            column("column1", Integer), column("column2", Integer),
+        ).data([(1, 1), (2, 1), (3, 2), (3, 3)])
+        stmt = select([people, values]).select_from(
+            people.join(values, values.c.column2 == people.c.people_id)
+        )
+        self.assert_compile(
+            stmt,
+            "SELECT people.people_id, people.age, people.name, column1, "
+            "column2 FROM people JOIN (VALUES (:param_1, :param_2), "
+            "(:param_3, :param_4), (:param_5, :param_6), "
+            "(:param_7, :param_8)) "
+            "ON people.people_id = column2",
+            checkparams={
+                "param_1": 1,
+                "param_2": 1,
+                "param_3": 2,
+                "param_4": 1,
+                "param_5": 3,
+                "param_6": 2,
+                "param_7": 3,
+                "param_8": 3,
+            },
+        )
+
+    def test_with_join_named(self):
+        people = self.tables.people
+        values = Values(
+            column("bookcase_id", Integer),
+            column("bookcase_owner_id", Integer),
+            name="bookcases",
+        ).data([(1, 1), (2, 1), (3, 2), (3, 3)])
+        stmt = select([people, values]).select_from(
+            people.join(
+                values, values.c.bookcase_owner_id == people.c.people_id
+            )
+        )
+        self.assert_compile(
+            stmt,
+            "SELECT people.people_id, people.age, people.name, "
+            "bookcases.bookcase_id, bookcases.bookcase_owner_id FROM people "
+            "JOIN (VALUES (:param_1, :param_2), (:param_3, :param_4), "
+            "(:param_5, :param_6), (:param_7, :param_8)) AS bookcases "
+            "(bookcase_id, bookcase_owner_id) "
+            "ON people.people_id = bookcases.bookcase_owner_id",
+            checkparams={
+                "param_1": 1,
+                "param_2": 1,
+                "param_3": 2,
+                "param_4": 1,
+                "param_5": 3,
+                "param_6": 2,
+                "param_7": 3,
+                "param_8": 3,
+            },
+        )
+
+    def test_with_aliased_join(self):
+        people = self.tables.people
+        values = (
+            Values(
+                column("bookcase_id", Integer),
+                column("bookcase_owner_id", Integer),
+            )
+            .data([(1, 1), (2, 1), (3, 2), (3, 3)])
+            .alias("bookcases")
+        )
+        stmt = select([people, values]).select_from(
+            people.join(
+                values, values.c.bookcase_owner_id == people.c.people_id
+            )
+        )
+        self.assert_compile(
+            stmt,
+            "SELECT people.people_id, people.age, people.name, "
+            "bookcases.bookcase_id, bookcases.bookcase_owner_id FROM people "
+            "JOIN (VALUES (:param_1, :param_2), (:param_3, :param_4), "
+            "(:param_5, :param_6), (:param_7, :param_8)) AS bookcases "
+            "(bookcase_id, bookcase_owner_id) "
+            "ON people.people_id = bookcases.bookcase_owner_id",
+            checkparams={
+                "param_1": 1,
+                "param_2": 1,
+                "param_3": 2,
+                "param_4": 1,
+                "param_5": 3,
+                "param_6": 2,
+                "param_7": 3,
+                "param_8": 3,
+            },
+        )
+
+    def test_with_standalone_aliased_join(self):
+        people = self.tables.people
+        values = Values(
+            column("bookcase_id", Integer),
+            column("bookcase_owner_id", Integer),
+        ).data([(1, 1), (2, 1), (3, 2), (3, 3)])
+        values = alias(values, "bookcases")
+
+        stmt = select([people, values]).select_from(
+            people.join(
+                values, values.c.bookcase_owner_id == people.c.people_id
+            )
+        )
+        self.assert_compile(
+            stmt,
+            "SELECT people.people_id, people.age, people.name, "
+            "bookcases.bookcase_id, bookcases.bookcase_owner_id FROM people "
+            "JOIN (VALUES (:param_1, :param_2), (:param_3, :param_4), "
+            "(:param_5, :param_6), (:param_7, :param_8)) AS bookcases "
+            "(bookcase_id, bookcase_owner_id) "
+            "ON people.people_id = bookcases.bookcase_owner_id",
+            checkparams={
+                "param_1": 1,
+                "param_2": 1,
+                "param_3": 2,
+                "param_4": 1,
+                "param_5": 3,
+                "param_6": 2,
+                "param_7": 3,
+                "param_8": 3,
+            },
+        )
+
+    def test_lateral(self):
+        people = self.tables.people
+        values = (
+            Values(
+                column("bookcase_id", Integer),
+                column("bookcase_owner_id", Integer),
+                name="bookcases",
+            )
+            .data([(1, 1), (2, 1), (3, 2), (3, 3)])
+            .lateral()
+        )
+        stmt = select([people, values]).select_from(
+            people.join(values, true())
+        )
+        self.assert_compile(
+            stmt,
+            "SELECT people.people_id, people.age, people.name, "
+            "bookcases.bookcase_id, bookcases.bookcase_owner_id FROM people "
+            "JOIN LATERAL (VALUES (:param_1, :param_2), (:param_3, :param_4), "
+            "(:param_5, :param_6), (:param_7, :param_8)) AS bookcases "
+            "(bookcase_id, bookcase_owner_id) "
+            "ON true",
+            checkparams={
+                "param_1": 1,
+                "param_2": 1,
+                "param_3": 2,
+                "param_4": 1,
+                "param_5": 3,
+                "param_6": 2,
+                "param_7": 3,
+                "param_8": 3,
+            },
+        )
+
+    def test_from_linting_named(self):
+        people = self.tables.people
+        values = Values(
+            column("bookcase_id", Integer),
+            column("bookcase_owner_id", Integer),
+            name="bookcases",
+        ).data([(1, 1), (2, 1), (3, 2), (3, 3)])
+        stmt = select([people, values])
+
+        with testing.expect_warnings(
+            r"SELECT statement has a cartesian product between FROM "
+            r'element\(s\) "(?:bookcases|people)" and '
+            r'FROM element "(?:people|bookcases)"'
+        ):
+            stmt.compile(linting=FROM_LINTING)
+
+    def test_from_linting_unnamed(self):
+        people = self.tables.people
+        values = Values(
+            column("bookcase_id", Integer),
+            column("bookcase_owner_id", Integer),
+        ).data([(1, 1), (2, 1), (3, 2), (3, 3)])
+        stmt = select([people, values])
+
+        with testing.expect_warnings(
+            r"SELECT statement has a cartesian product between FROM "
+            r'element\(s\) "(?:\(unnamed VALUES element\)|people)" and '
+            r'FROM element "(?:people|\(unnamed VALUES element\))"'
+        ):
+            stmt.compile(linting=FROM_LINTING)