Implement support for functions as FROM with columns clause support

WIP

Fixes: #3566
Change-Id: I5b093b72533ef695293e737eb75850b9713e5e03

7 files changed, 514 insertions, 75 deletions

diff --git a/lib/sqlalchemy/sql/coercions.py b/lib/sqlalchemy/sql/coercions.py
index 05e0a4fcf..b5b5082e6 100644
--- a/lib/sqlalchemy/sql/coercions.py
+++ b/lib/sqlalchemy/sql/coercions.py
@@ -431,6 +431,20 @@ class ExpressionElementImpl(_ColumnCoercions, RoleImpl):
             except exc.ArgumentError as err:
                 self._raise_for_expected(element, err=err)
 
+    def _raise_for_expected(self, element, argname=None, resolved=None, **kw):
+        if isinstance(element, roles.AnonymizedFromClauseRole):
+            advice = (
+                "To create a "
+                "column expression from a FROM clause row "
+                "as a whole, use the .record() method."
+            )
+        else:
+            advice = None
+
+        return super(ExpressionElementImpl, self)._raise_for_expected(
+            element, argname=argname, resolved=resolved, advice=advice, **kw
+        )
+
 
 class BinaryElementImpl(ExpressionElementImpl, RoleImpl):
 
@@ -597,6 +611,13 @@ class ColumnArgumentOrKeyImpl(_ReturnsStringKey, RoleImpl):
     __slots__ = ()
 
 
+class StrAsPlainColumnImpl(_CoerceLiterals, RoleImpl):
+    __slots__ = ()
+
+    def _text_coercion(self, element, argname=None):
+        return elements.ColumnClause(element)
+
+
 class ByOfImpl(_CoerceLiterals, _ColumnCoercions, RoleImpl, roles.ByOfRole):
 
     __slots__ = ()
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index ef7db0bbe..db13cff01 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -1718,13 +1718,18 @@ class SQLCompiler(Compiled):
             extract.expr._compiler_dispatch(self, **kwargs),
         )
 
+    def visit_scalar_function_column(self, element, **kw):
+        compiled_fn = self.visit_function(element.fn, **kw)
+        compiled_col = self.visit_column(element, **kw)
+        return "(%s).%s" % (compiled_fn, compiled_col)
+
     def visit_function(self, func, add_to_result_map=None, **kwargs):
         if add_to_result_map is not None:
             add_to_result_map(func.name, func.name, (), func.type)
 
         disp = getattr(self, "visit_%s_func" % func.name.lower(), None)
         if disp:
-            return disp(func, **kwargs)
+            text = disp(func, **kwargs)
         else:
             name = FUNCTIONS.get(func.__class__, None)
             if name:
@@ -1739,7 +1744,7 @@ class SQLCompiler(Compiled):
                     else name
                 )
                 name = name + "%(expr)s"
-            return ".".join(
+            text = ".".join(
                 [
                     (
                         self.preparer.quote(tok)
@@ -1752,6 +1757,10 @@ class SQLCompiler(Compiled):
                 + [name]
             ) % {"expr": self.function_argspec(func, **kwargs)}
 
+        if func._with_ordinality:
+            text += " WITH ORDINALITY"
+        return text
+
     def visit_next_value_func(self, next_value, **kw):
         return self.visit_sequence(next_value.sequence)
 
@@ -2527,6 +2536,27 @@ class SQLCompiler(Compiled):
             else:
                 return self.preparer.format_alias(cte, cte_name)
 
+    def visit_table_valued_alias(self, element, **kw):
+        text = self.visit_alias(element, **kw)
+        if kw.get("asfrom") and element.named:
+            text += "(%s)" % (
+                ", ".join(
+                    "%s%s"
+                    % (
+                        col.name,
+                        " %s"
+                        % self.dialect.type_compiler.process(col.type, **kw)
+                        if not col.type._isnull
+                        else "",
+                    )
+                    for col in element.element.c
+                )
+            )
+        return text
+
+    def visit_table_valued_column(self, element, **kw):
+        return self.visit_column(element, **kw)
+
     def visit_alias(
         self,
         alias,
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index 5ea3526ea..7dc60ce9d 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -4317,11 +4317,82 @@ class Label(roles.LabeledColumnExprRole, ColumnElement):
         return self.key, e
 
 
+class NamedColumn(ColumnElement):
+    is_literal = False
+    table = None
+
+    def _compare_name_for_result(self, other):
+        return (hasattr(other, "name") and self.name == other.name) or (
+            hasattr(other, "_label") and self._label == other._label
+        )
+
+    @util.memoized_property
+    def description(self):
+        if util.py3k:
+            return self.name
+        else:
+            return self.name.encode("ascii", "backslashreplace")
+
+    @HasMemoized.memoized_attribute
+    def _key_label(self):
+        if self.key != self.name:
+            return self._gen_label(self.key)
+        else:
+            return self._label
+
+    @HasMemoized.memoized_attribute
+    def _label(self):
+        return self._gen_label(self.name)
+
+    @HasMemoized.memoized_attribute
+    def _render_label_in_columns_clause(self):
+        return True
+
+    def _gen_label(self, name, dedupe_on_key=True):
+        return name
+
+    def _bind_param(self, operator, obj, type_=None, expanding=False):
+        return BindParameter(
+            self.key,
+            obj,
+            _compared_to_operator=operator,
+            _compared_to_type=self.type,
+            type_=type_,
+            unique=True,
+            expanding=expanding,
+        )
+
+    def _make_proxy(
+        self,
+        selectable,
+        name=None,
+        name_is_truncatable=False,
+        disallow_is_literal=False,
+        **kw
+    ):
+        c = ColumnClause(
+            coercions.expect(roles.TruncatedLabelRole, name or self.name)
+            if name_is_truncatable
+            else (name or self.name),
+            type_=self.type,
+            _selectable=selectable,
+            is_literal=False,
+        )
+        c._propagate_attrs = selectable._propagate_attrs
+        if name is None:
+            c.key = self.key
+        c._proxies = [self]
+        if selectable._is_clone_of is not None:
+            c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
+        return c.key, c
+
+
 class ColumnClause(
     roles.DDLReferredColumnRole,
     roles.LabeledColumnExprRole,
+    roles.StrAsPlainColumnRole,
     Immutable,
-    ColumnElement,
+    NamedColumn,
 ):
     """Represents a column expression from any textual string.
 
@@ -4360,6 +4431,9 @@ class ColumnClause(
 
     """
 
+    table = None
+    is_literal = False
+
     __visit_name__ = "column"
 
     _traverse_internals = [
@@ -4470,27 +4544,6 @@ class ColumnClause(
         self.type = type_api.to_instance(type_)
         self.is_literal = is_literal
 
-    def _compare_name_for_result(self, other):
-        if (
-            self.is_literal
-            or self.table is None
-            or self.table._is_textual
-            or not hasattr(other, "proxy_set")
-            or (
-                isinstance(other, ColumnClause)
-                and (
-                    other.is_literal
-                    or other.table is None
-                    or other.table._is_textual
-                )
-            )
-        ):
-            return (hasattr(other, "name") and self.name == other.name) or (
-                hasattr(other, "_label") and self._label == other._label
-            )
-        else:
-            return other.proxy_set.intersection(self.proxy_set)
-
     def get_children(self, column_tables=False, **kw):
         # override base get_children() to not return the Table
         # or selectable that is parent to this column.  Traversals
@@ -4505,24 +4558,6 @@ class ColumnClause(
         else:
             return []
 
-    @util.memoized_property
-    def description(self):
-        if util.py3k:
-            return self.name
-        else:
-            return self.name.encode("ascii", "backslashreplace")
-
-    @HasMemoized.memoized_attribute
-    def _key_label(self):
-        if self.key != self.name:
-            return self._gen_label(self.key)
-        else:
-            return self._label
-
-    @HasMemoized.memoized_attribute
-    def _label(self):
-        return self._gen_label(self.name)
-
     @HasMemoized.memoized_attribute
     def _render_label_in_columns_clause(self):
         return self.table is not None
@@ -4531,6 +4566,27 @@ class ColumnClause(
     def _ddl_label(self):
         return self._gen_label(self.name, dedupe_on_key=False)
 
+    def _compare_name_for_result(self, other):
+        if (
+            self.is_literal
+            or self.table is None
+            or self.table._is_textual
+            or not hasattr(other, "proxy_set")
+            or (
+                isinstance(other, ColumnClause)
+                and (
+                    other.is_literal
+                    or other.table is None
+                    or other.table._is_textual
+                )
+            )
+        ):
+            return (hasattr(other, "name") and self.name == other.name) or (
+                hasattr(other, "_label") and self._label == other._label
+            )
+        else:
+            return other.proxy_set.intersection(self.proxy_set)
+
     def _gen_label(self, name, dedupe_on_key=True):
         t = self.table
         if self.is_literal:
@@ -4575,17 +4631,6 @@ class ColumnClause(
         else:
             return name
 
-    def _bind_param(self, operator, obj, type_=None, expanding=False):
-        return BindParameter(
-            self.key,
-            obj,
-            _compared_to_operator=operator,
-            _compared_to_type=self.type,
-            type_=type_,
-            unique=True,
-            expanding=expanding,
-        )
-
     def _make_proxy(
         self,
         selectable,
@@ -4627,6 +4672,46 @@ class ColumnClause(
         return c.key, c
 
 
+class Record(NamedColumn):
+    _traverse_internals = [
+        ("name", InternalTraversal.dp_anon_name),
+        ("type", InternalTraversal.dp_type),
+        ("fromclause", InternalTraversal.dp_clauseelement),
+    ]
+
+    __visit_name__ = "column"
+
+    @util.preload_module("sqlalchemy.sql.sqltypes")
+    def __init__(self, fromclause):
+        sqltypes = util.preloaded.sql_sqltypes
+        self.name = fromclause.name
+        self.fromclause = fromclause
+        self.type = sqltypes.RecordType()
+
+    @property
+    def _from_objects(self):
+        return [self.fromclause]
+
+
+class TableValuedColumn(NamedColumn):
+    __visit_name__ = "table_valued_column"
+
+    _traverse_internals = [
+        ("name", InternalTraversal.dp_anon_name),
+        ("type", InternalTraversal.dp_type),
+        ("scalar_alias", InternalTraversal.dp_clauseelement),
+    ]
+
+    def __init__(self, scalar_alias):
+        self.scalar_alias = scalar_alias
+        self.key = self.name = scalar_alias.name
+        self.type = scalar_alias.element.type
+
+    @property
+    def _from_objects(self):
+        return [self.scalar_alias]
+
+
 class CollationClause(ColumnElement):
     __visit_name__ = "collation"
 
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index a9ea98d04..f493b08db 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -17,22 +17,25 @@ from . import sqltypes
 from . import util as sqlutil
 from .base import ColumnCollection
 from .base import Executable
+from .base import Generative
 from .base import HasMemoized
 from .elements import _type_from_args
 from .elements import BinaryExpression
 from .elements import BindParameter
 from .elements import Cast
 from .elements import ClauseList
+from .elements import ColumnClause
 from .elements import ColumnElement
 from .elements import Extract
 from .elements import FunctionFilter
 from .elements import Grouping
 from .elements import literal_column
+from .elements import NamedColumn
 from .elements import Over
 from .elements import WithinGroup
-from .selectable import Alias
 from .selectable import FromClause
 from .selectable import Select
+from .selectable import TableValuedAlias
 from .visitors import InternalTraversal
 from .visitors import TraversibleType
 from .. import util
@@ -63,7 +66,7 @@ def register_function(identifier, fn, package="_default"):
     reg[identifier] = fn
 
 
-class FunctionElement(Executable, ColumnElement, FromClause):
+class FunctionElement(Executable, ColumnElement, FromClause, Generative):
     """Base for SQL function-oriented constructs.
 
     .. seealso::
@@ -80,11 +83,19 @@ class FunctionElement(Executable, ColumnElement, FromClause):
 
     """
 
-    _traverse_internals = [("clause_expr", InternalTraversal.dp_clauseelement)]
+    _traverse_internals = [
+        ("clause_expr", InternalTraversal.dp_clauseelement),
+        ("_table_valued", InternalTraversal.dp_clauseelement_tuple),
+        ("_table_values_named", InternalTraversal.dp_boolean),
+        ("_with_ordinality", InternalTraversal.dp_boolean),
+    ]
 
     packagenames = ()
 
     _has_args = False
+    _table_valued = None
+    _table_values_named = False
+    _with_ordinality = False
 
     def __init__(self, *clauses, **kwargs):
         r"""Construct a :class:`.FunctionElement`.
@@ -123,6 +134,171 @@ class FunctionElement(Executable, ColumnElement, FromClause):
             self, multiparams, params, execution_options
         )
 
+    def scalar_table_valued(self, name, type_=None):
+        """Return a column expression that's against this
+        :class:`_functions.FunctionElement` as a scalar
+        table-valued expression.
+
+        The returned expression is similar to that returned by a single column
+        accessed off of a :meth:`_functions.FunctionElement.table_valued`
+        construct, except no FROM clause is generated; the function is rendered
+        in the similar way as a scalar subquery.
+
+        E.g.::
+
+            >>> from sqlalchemy import func, select
+            >>> fn = func.jsonb_each("{'k', 'v'}").scalar_table_valued("key")
+            >>> print(select(fn))
+            SELECT (jsonb_each(:jsonb_each_1)).key
+
+        .. versionadded:: 1.4.0b2
+
+        .. seealso::
+
+            :meth:`_functions.FunctionElement.table_valued`
+
+            :meth:`_functions.FunctionElement.named_table_valued`
+
+            :meth:`_functions.FunctionElement.alias`
+
+            :meth:`_functions.FunctionElement.column_valued`
+
+        """  # noqa E501
+
+        return ScalarFunctionColumn(self, name, type_)
+
+    def table_valued(self, *expr, **kw):
+        """Return a :class:`_sql.TableValuedAlias` representation of this
+        :class:`_functions.FunctionElement` with table-valued expressions added.
+
+        e.g.::
+
+            >>> fn = (
+            ...     func.generate_series(1, 5).
+            ...     table_valued("value", "start", "stop", "step")
+            ... )
+
+            >>> print(select(fn))
+            SELECT anon_1.value, anon_1.start, anon_1.stop, anon_1.step
+            FROM generate_series(:generate_series_1, :generate_series_2) AS anon_1
+
+            >>> print(select(fn.c.value, fn.c.stop).where(fn.c.value > 2))
+            SELECT anon_1.value, anon_1.stop
+            FROM generate_series(:generate_series_1, :generate_series_2) AS anon_1
+            WHERE anon_1.value > :value_1
+
+        A WITH ORDINALITY expression may be generated by passing the keyword
+        argument "with_ordinality"::
+
+            >>> fn = func.generate_series(4, 1, -1).table_valued("gen", with_ordinality="ordinality")
+            >>> print(select(fn))
+
+        .. versionadded:: 1.4.0b2
+
+        .. seealso::
+
+            :meth:`_functions.FunctionElement.table_valued`
+
+            :meth:`_functions.FunctionElement.named_table_valued`
+
+            :meth:`_functions.FunctionElement.alias`
+
+
+        """  # noqa 501
+
+        new_func = self._generate()
+        new_func._table_valued = [
+            coercions.expect(roles.StrAsPlainColumnRole, elem) for elem in expr
+        ]
+
+        with_ordinality = kw.pop("with_ordinality", None)
+        if with_ordinality:
+            new_func._table_valued += (
+                coercions.expect(roles.StrAsPlainColumnRole, with_ordinality),
+            )
+            new_func._with_ordinality = True
+
+        return new_func.alias()
+
+    def named_table_valued(self, *expr, **kw):
+        """Return a :class:`_sql.TableValuedAlias` representation of this
+        :class:`_functions.FunctionElement` with named table-valued
+        expressions added.
+
+        E.g.::
+
+
+            >>> fn = (
+            ...     func.json_to_recordset(
+            ...             '[{"a":1,"b":"foo"},{"a":"2","c":"bar"}]'
+            ...     )
+            ...     .named_table_valued(column("a", Integer), column("b", String))
+            ... )
+            >>> print(select(fn.c.a, fn.c.b))
+            SELECT anon_1.a, anon_1.b
+            FROM json_to_recordset(:json_to_recordset_1) AS anon_1(a INTEGER, b VARCHAR)
+
+        A WITH ORDINALITY expression may be generated by passing the keyword
+        argument "with_ordinality"::
+
+            >>> fn = (
+            ...     func.json_object_keys('{"a1":"1","a2":"2","a3":"3"}')
+            ...     .named_table_valued("keys", with_ordinality="n")
+            ... )
+
+            >>> print(select(fn))
+
+        .. seealso::
+
+            :meth:`_functions.FunctionElement.table_valued`
+
+            :meth:`_functions.FunctionElement.named_table_valued`
+
+            :meth:`_functions.FunctionElement.alias`
+
+
+        """  # noqa E501
+
+        new_func = self._generate()
+        new_func._table_valued = [
+            coercions.expect(roles.StrAsPlainColumnRole, elem) for elem in expr
+        ]
+        with_ordinality = kw.pop("with_ordinality", None)
+
+        if with_ordinality:
+            new_func._table_valued += (
+                coercions.expect(roles.StrAsPlainColumnRole, with_ordinality),
+            )
+            new_func._with_ordinality = True
+
+        new_func._table_values_named = True
+        return new_func.alias()
+
+    def column_valued(self, name=None):
+        """Return this :class:`_function.FunctionElement` as a column expression that
+        selects from itself as a FROM clause.
+
+        E.g.::
+
+            >>> from sqlalchemy import select, func
+            >>> gs = func.generate_series(1, 5, -1).column_valued()
+            >>> print(select(gs))
+            SELECT anon_1
+            FROM generate_series(:generate_series_1, :generate_series_2, :generate_series_3) AS anon_1
+
+        This is shorthand for::
+
+            gs = func.generate_series(1, 5, -1).alias().column
+
+
+        .. seealso::
+
+            :meth:`_functions.FunctionElement.alias`
+
+        """  # noqa 501
+
+        return self.alias(name=name).column
+
     @property
     def columns(self):
         r"""The set of columns exported by this :class:`.FunctionElement`.
@@ -142,8 +318,15 @@ class FunctionElement(Executable, ColumnElement, FromClause):
 
 
         """
-        col = self.label(None)
-        return ColumnCollection(columns=[(col.key, col)])
+        if self._table_valued:
+            cols = [
+                ColumnClause(elem) if isinstance(elem, str) else elem
+                for elem in self._table_valued
+            ]
+        else:
+            cols = [self.label(None)]
+
+        return ColumnCollection(columns=[(col.key, col) for col in cols])
 
     @HasMemoized.memoized_attribute
     def clauses(self):
@@ -305,37 +488,65 @@ class FunctionElement(Executable, ColumnElement, FromClause):
 
         return None
 
-    def alias(self, name=None, flat=False):
+    def alias(self, name=None):
         r"""Produce a :class:`_expression.Alias` construct against this
         :class:`.FunctionElement`.
 
+        .. note::
+
+            The :meth:`_functions.FunctionElement.alias` method is part of the
+            mechanism by which "table valued" SQL functions are created.
+            However, most use cases are covered by higher level methods on
+            :class:`_functions.FunctionElement` including
+            :meth:`_functions.FunctionElement.table_valued`,
+            :meth:`_functions.FunctionElement.named_table_valued`, and
+            :meth:`_functions.FunctionElement.column_valued`.
+
         This construct wraps the function in a named alias which
         is suitable for the FROM clause, in the style accepted for example
-        by PostgreSQL.
+        by PostgreSQL.  A column expression is also provided using the
+        special ``.column`` attribute, which may
+        be used to refer to the output of the function as a scalar value
+        in the columns or where clause, for a backend such as PostgreSQL.
+
+        For a full table-valued expression, use the
+        :meth:`_function.FunctionElement.table_valued` method first to
+        establish named columns.
 
         e.g.::
 
-            from sqlalchemy.sql import column
+            >>> from sqlalchemy import func, select, column
+            >>> data_view = func.unnest([1, 2, 3]).alias("data_view")
+            >>> print(select(data_view.column))
+            SELECT data_view
+            FROM unnest(:unnest_1) AS data_view
 
-            stmt = select(column('data_view')).\
-                select_from(SomeTable).\
-                select_from(func.unnest(SomeTable.data).alias('data_view')
-            )
+        The :meth:`_functions.FunctionElement.column_valued` method provides
+        a shortcut for the above pattern::
+
+            >>> data_view = func.unnest([1, 2, 3]).column_valued("data_view")
+            >>> print(select(data_view))
+            SELECT data_view
+            FROM unnest(:unnest_1) AS data_view
 
-        Would produce:
+        .. versionadded:: 1.4.0b2  Added the ``.column`` accessor
 
-        .. sourcecode:: sql
+        .. seealso::
 
-            SELECT data_view
-            FROM sometable, unnest(sometable.data) AS data_view
+            :meth:`_functions.FunctionElement.table_valued`
+
+            :meth:`_functions.FunctionElement.named_table_valued`
+
+            :meth:`_functions.FunctionElement.scalar_table_valued`
+
+            :meth:`_functions.FunctionElement.column_valued`
 
-        .. versionadded:: 0.9.8 The :meth:`.FunctionElement.alias` method
-           is now supported.  Previously, this method's behavior was
-           undefined and did not behave consistently across versions.
 
         """
 
-        return Alias._construct(self, name)
+        return TableValuedAlias._construct(
+            self, name, named=self._table_values_named
+        )
 
     def select(self):
         """Produce a :func:`_expression.select` construct
@@ -441,6 +652,24 @@ class FunctionAsBinary(BinaryExpression):
         self.sql_function.clauses.clauses[self.right_index - 1] = value
 
 
+class ScalarFunctionColumn(NamedColumn):
+    __visit_name__ = "scalar_function_column"
+
+    _traverse_internals = [
+        ("name", InternalTraversal.dp_anon_name),
+        ("type", InternalTraversal.dp_type),
+        ("fn", InternalTraversal.dp_clauseelement),
+    ]
+
+    is_literal = False
+    table = None
+
+    def __init__(self, fn, name, type_=None):
+        self.fn = fn
+        self.name = name
+        self.type = sqltypes.to_instance(type_)
+
+
 class _FunctionGenerator(object):
     """Generate SQL function expressions.
 
@@ -586,10 +815,9 @@ class Function(FunctionElement):
 
         func.mypackage.some_function(col1, col2)
 
-
     .. seealso::
 
-        :ref:`coretutorial_functions`
+        :ref:`tutorial_functions` - in the :ref:`unified_tutorial`
 
         :data:`.func` - namespace which produces registered or ad-hoc
         :class:`.Function` instances.
diff --git a/lib/sqlalchemy/sql/roles.py b/lib/sqlalchemy/sql/roles.py
index 2c4ff75c4..52743bd50 100644
--- a/lib/sqlalchemy/sql/roles.py
+++ b/lib/sqlalchemy/sql/roles.py
@@ -44,6 +44,10 @@ class ColumnArgumentOrKeyRole(ColumnArgumentRole):
     _role_name = "Column expression or string key"
 
 
+class StrAsPlainColumnRole(ColumnArgumentRole):
+    _role_name = "Column expression or string key"
+
+
 class ColumnListRole(SQLRole):
     """Elements suitable for forming comma separated lists of expressions."""
 
diff --git a/lib/sqlalchemy/sql/selectable.py b/lib/sqlalchemy/sql/selectable.py
index 8e478583f..c299065ae 100644
--- a/lib/sqlalchemy/sql/selectable.py
+++ b/lib/sqlalchemy/sql/selectable.py
@@ -51,6 +51,8 @@ from .elements import ColumnClause
 from .elements import GroupedElement
 from .elements import Grouping
 from .elements import literal_column
+from .elements import Record
+from .elements import TableValuedColumn
 from .elements import UnaryExpression
 from .visitors import InternalTraversal
 from .. import exc
@@ -623,6 +625,33 @@ class FromClause(roles.AnonymizedFromClauseRole, Selectable):
 
         return Alias._construct(self, name)
 
+    def record(self):
+        """Return a :class:`_sql.Record` object for this
+        :class:`_expression.FromClause`.
+
+        A "record" is a :class:`_sql.ColumnElement` that represents a complete
+        row in a table.   Support for this construct is backend dependent,
+        however databases such as PostgreSQL and Oracle have support
+        for "record" datatypes.
+
+        E.g.::
+
+            >>> from sqlalchemy import select, column, func, table
+            >>> a = table("a", column("id"), column("x"), column("y"))
+            >>> stmt = select(func.row_to_json(a.record()))
+            >>> print(stmt)
+            SELECT row_to_json(a) AS row_to_json_1
+            FROM a
+
+        .. versionadded:: 1.4.0b2
+
+        .. seealso::
+
+            :ref:`tutorial_functions` - in the :ref:`unified_tutorial`
+
+        """
+        return Record(self)
+
     def tablesample(self, sampling, name=None, seed=None):
         """Return a TABLESAMPLE alias of this :class:`_expression.FromClause`.
 
@@ -1583,6 +1612,44 @@ class Alias(roles.DMLTableRole, AliasedReturnsRows):
         ).alias(name=name, flat=flat)
 
 
+class TableValuedAlias(Alias):
+    """An alias that includes the ability to be used in a columns context.
+
+    Provides the :attr:`_sql.ScalarAlias.column` accessor that returns
+    a :class:`_sql.ColumnElement` representing this object.
+
+    The main use case for this construct is that of PostgreSQL functions
+    that may be used in the FROM clause of a SELECT.
+
+    .. versionadded:: 1.4.0b2
+
+    """
+
+    __visit_name__ = "table_valued_alias"
+
+    named = False
+
+    _traverse_internals = [
+        ("element", InternalTraversal.dp_clauseelement),
+        ("name", InternalTraversal.dp_anon_name),
+        ("named", InternalTraversal.dp_boolean),
+    ]
+
+    def _init(self, selectable, name=None, named=False):
+        self.named = named
+        super(TableValuedAlias, self)._init(selectable, name=name)
+
+    @HasMemoized.memoized_attribute
+    def column(self):
+        """Return a column expression representing this
+        :class:`_sql.ScalarAlias`."""
+
+        return TableValuedColumn(self)
+
+    def alias(self, name=None):
+        return self.element.alias(name=name)
+
+
 class Lateral(AliasedReturnsRows):
     """Represent a LATERAL subquery.
 
diff --git a/lib/sqlalchemy/sql/sqltypes.py b/lib/sqlalchemy/sql/sqltypes.py
index d20c8168d..cac7d74a4 100644
--- a/lib/sqlalchemy/sql/sqltypes.py
+++ b/lib/sqlalchemy/sql/sqltypes.py
@@ -3110,6 +3110,10 @@ class NullType(TypeEngine):
     comparator_factory = Comparator
 
 
+class RecordType(TypeEngine):
+    """Refers to a table record type."""
+
+
 class MatchType(Boolean):
     """Refers to the return type of the MATCH operator.