5 files changed, 296 insertions, 24 deletions
diff --git a/lib/sqlalchemy/sql/compiler.py b/lib/sqlalchemy/sql/compiler.py
index a5a3975b1..52116a231 100644
--- a/lib/sqlalchemy/sql/compiler.py
+++ b/lib/sqlalchemy/sql/compiler.py
@@ -772,7 +772,7 @@ class SQLCompiler(Compiled):
 
     def visit_over(self, over, **kwargs):
         return "%s OVER (%s)" % (
-            over.func._compiler_dispatch(self, **kwargs),
+            over.element._compiler_dispatch(self, **kwargs),
             ' '.join(
                 '%s BY %s' % (word, clause._compiler_dispatch(self, **kwargs))
                 for word, clause in (
@@ -783,6 +783,12 @@ class SQLCompiler(Compiled):
             )
         )
 
+    def visit_withingroup(self, withingroup, **kwargs):
+        return "%s WITHIN GROUP (ORDER BY %s)" % (
+            withingroup.element._compiler_dispatch(self, **kwargs),
+            withingroup.order_by._compiler_dispatch(self, **kwargs)
+        )
+
     def visit_funcfilter(self, funcfilter, **kwargs):
         return "%s FILTER (WHERE %s)" % (
             funcfilter.func._compiler_dispatch(self, **kwargs),
diff --git a/lib/sqlalchemy/sql/elements.py b/lib/sqlalchemy/sql/elements.py
index d5d364c77..618b987e1 100644
--- a/lib/sqlalchemy/sql/elements.py
+++ b/lib/sqlalchemy/sql/elements.py
@@ -2970,21 +2970,21 @@ class Over(ColumnElement):
     order_by = None
     partition_by = None
 
-    def __init__(self, func, partition_by=None, order_by=None):
+    def __init__(self, element, partition_by=None, order_by=None):
         """Produce an :class:`.Over` object against a function.
 
         Used against aggregate or so-called "window" functions,
         for database backends that support window functions.
 
-        E.g.::
+        :func:`~.expression.over` is usually called using
+        the :meth:`.FunctionElement.over` method, e.g.::
 
-            from sqlalchemy import over
-            over(func.row_number(), order_by='x')
+            func.row_number().over(order_by='x')
 
-        Would produce "ROW_NUMBER() OVER(ORDER BY x)".
+        Would produce ``ROW_NUMBER() OVER(ORDER BY x)``.
 
-        :param func: a :class:`.FunctionElement` construct, typically
-         generated by :data:`~.expression.func`.
+        :param element: a :class:`.FunctionElement`, :class:`.WithinGroup`,
+         or other compatible construct.
         :param partition_by: a column element or string, or a list
          of such, that will be used as the PARTITION BY clause
          of the OVER construct.
@@ -2997,8 +2997,14 @@ class Over(ColumnElement):
 
         .. versionadded:: 0.7
 
+        .. seealso::
+
+            :data:`.expression.func`
+
+            :func:`.expression.within_group`
+
         """
-        self.func = func
+        self.element = element
         if order_by is not None:
             self.order_by = ClauseList(
                 *util.to_list(order_by),
@@ -3008,17 +3014,29 @@ class Over(ColumnElement):
                 *util.to_list(partition_by),
                 _literal_as_text=_literal_as_label_reference)
 
+    @property
+    def func(self):
+        """the element referred to by this :class:`.Over`
+        clause.
+
+        .. deprecated:: 1.1 the ``func`` element has been renamed to
+           ``.element``.  The two attributes are synonymous though
+           ``.func`` is read-only.
+
+        """
+        return self.element
+
     @util.memoized_property
     def type(self):
-        return self.func.type
+        return self.element.type
 
     def get_children(self, **kwargs):
         return [c for c in
-                (self.func, self.partition_by, self.order_by)
+                (self.element, self.partition_by, self.order_by)
                 if c is not None]
 
     def _copy_internals(self, clone=_clone, **kw):
-        self.func = clone(self.func, **kw)
+        self.element = clone(self.element, **kw)
         if self.partition_by is not None:
             self.partition_by = clone(self.partition_by, **kw)
         if self.order_by is not None:
@@ -3028,7 +3046,106 @@ class Over(ColumnElement):
     def _from_objects(self):
         return list(itertools.chain(
             *[c._from_objects for c in
-                (self.func, self.partition_by, self.order_by)
+                (self.element, self.partition_by, self.order_by)
+              if c is not None]
+        ))
+
+
+class WithinGroup(ColumnElement):
+    """Represent a WITHIN GROUP (ORDER BY) clause.
+
+    This is a special operator against so-called
+    so-called "ordered set aggregate" and "hypothetical
+    set aggregate" functions, including ``percentile_cont()``,
+    ``rank()``, ``dense_rank()``, etc.
+
+    It's supported only by certain database backends, such as PostgreSQL,
+    Oracle and MS SQL Server.
+
+    The :class:`.WithinGroup` consturct extracts its type from the
+    method :meth:`.FunctionElement.within_group_type`.  If this returns
+    ``None``, the function's ``.type`` is used.
+
+    """
+    __visit_name__ = 'withingroup'
+
+    order_by = None
+
+    def __init__(self, element, *order_by):
+        """Produce a :class:`.WithinGroup` object against a function.
+
+        Used against so-called "ordered set aggregate" and "hypothetical
+        set aggregate" functions, including :class:`.percentile_cont`,
+        :class:`.rank`, :class:`.dense_rank`, etc.
+
+        :func:`~.expression.within_group` is usually called using
+        the :meth:`.FunctionElement.within_group` method, e.g.::
+
+            from sqlalchemy import within_group
+            stmt = select([
+                department.c.id,
+                func.percentile_cont(0.5).within_group(
+                    department.c.salary.desc()
+                )
+            ])
+
+        The above statement would produce SQL similar to
+        ``SELECT department.id, percentile_cont(0.5)
+        WITHIN GROUP (ORDER BY department.salary DESC)``.
+
+        :param element: a :class:`.FunctionElement` construct, typically
+         generated by :data:`~.expression.func`.
+        :param \*order_by: one or more column elements that will be used
+         as the ORDER BY clause of the WITHIN GROUP construct.
+
+        .. versionadded:: 1.1
+
+        .. seealso::
+
+            :data:`.expression.func`
+
+            :func:`.expression.over`
+
+        """
+        self.element = element
+        if order_by is not None:
+            self.order_by = ClauseList(
+                *util.to_list(order_by),
+                _literal_as_text=_literal_as_label_reference)
+
+    def over(self, partition_by=None, order_by=None):
+        """Produce an OVER clause against this :class:`.WithinGroup`
+        construct.
+
+        This function has the same signature as that of
+        :meth:`.FunctionElement.over`.
+
+        """
+        return Over(self, partition_by=partition_by, order_by=order_by)
+
+    @util.memoized_property
+    def type(self):
+        wgt = self.element.within_group_type(self)
+        if wgt is not None:
+            return wgt
+        else:
+            return self.element.type
+
+    def get_children(self, **kwargs):
+        return [c for c in
+                (self.func, self.order_by)
+                if c is not None]
+
+    def _copy_internals(self, clone=_clone, **kw):
+        self.element = clone(self.element, **kw)
+        if self.order_by is not None:
+            self.order_by = clone(self.order_by, **kw)
+
+    @property
+    def _from_objects(self):
+        return list(itertools.chain(
+            *[c._from_objects for c in
+                (self.element, self.order_by)
               if c is not None]
         ))
 
diff --git a/lib/sqlalchemy/sql/expression.py b/lib/sqlalchemy/sql/expression.py
index 9bd424e21..79d25a39e 100644
--- a/lib/sqlalchemy/sql/expression.py
+++ b/lib/sqlalchemy/sql/expression.py
@@ -24,7 +24,7 @@ __all__ = [
     'literal', 'literal_column', 'not_', 'null', 'nullsfirst', 'nullslast',
     'or_', 'outparam', 'outerjoin', 'over', 'select', 'subquery',
     'table', 'text',
-    'tuple_', 'type_coerce', 'union', 'union_all', 'update']
+    'tuple_', 'type_coerce', 'union', 'union_all', 'update', 'within_group']
 
 
 from .visitors import Visitable
@@ -34,7 +34,7 @@ from .elements import ClauseElement, ColumnElement,\
     BindParameter, CollectionAggregate, UnaryExpression, BooleanClauseList, \
     Label, Cast, Case, ColumnClause, TextClause, Over, Null, \
     True_, False_, BinaryExpression, Tuple, TypeClause, Extract, \
-    Grouping, not_, \
+    Grouping, WithinGroup, not_, \
     collate, literal_column, between,\
     literal, outparam, type_coerce, ClauseList, FunctionFilter
 
@@ -67,6 +67,7 @@ text = public_factory(TextClause._create_text, ".expression.text")
 table = public_factory(TableClause, ".expression.table")
 column = public_factory(ColumnClause, ".expression.column")
 over = public_factory(Over, ".expression.over")
+within_group = public_factory(WithinGroup, ".expression.within_group")
 label = public_factory(Label, ".expression.label")
 case = public_factory(Case, ".expression.case")
 cast = public_factory(Cast, ".expression.cast")
diff --git a/lib/sqlalchemy/sql/functions.py b/lib/sqlalchemy/sql/functions.py
index d536c3008..d5d0eb7f2 100644
--- a/lib/sqlalchemy/sql/functions.py
+++ b/lib/sqlalchemy/sql/functions.py
@@ -12,9 +12,9 @@ from . import sqltypes, schema
 from .base import Executable, ColumnCollection
 from .elements import ClauseList, Cast, Extract, _literal_as_binds, \
     literal_column, _type_from_args, ColumnElement, _clone,\
-    Over, BindParameter, FunctionFilter, Grouping
+    Over, BindParameter, FunctionFilter, Grouping, WithinGroup
 from .selectable import FromClause, Select, Alias
-
+from . import util as sqlutil
 from . import operators
 from .visitors import VisitableType
 from .. import util
@@ -116,6 +116,21 @@ class FunctionElement(Executable, ColumnElement, FromClause):
         """
         return Over(self, partition_by=partition_by, order_by=order_by)
 
+    def within_group(self, *order_by):
+        """Produce a WITHIN GROUP (ORDER BY expr) clause against this function.
+
+        Used against so-called "ordered set aggregate" and "hypothetical
+        set aggregate" functions, including :class:`.percentile_cont`,
+        :class:`.rank`, :class:`.dense_rank`, etc.
+
+        See :func:`~.expression.within_group` for a full description.
+
+        .. versionadded:: 1.1
+
+
+        """
+        return WithinGroup(self, *order_by)
+
     def filter(self, *criterion):
         """Produce a FILTER clause against this function.
 
@@ -157,6 +172,18 @@ class FunctionElement(Executable, ColumnElement, FromClause):
         self._reset_exported()
         FunctionElement.clauses._reset(self)
 
+    def within_group_type(self, within_group):
+        """For types that define their return type as based on the criteria
+        within a WITHIN GROUP (ORDER BY) expression, called by the
+        :class:`.WithinGroup` construct.
+
+        Returns None by default, in which case the function's normal ``.type``
+        is used.
+
+        """
+
+        return None
+
     def alias(self, name=None, flat=False):
         """Produce a :class:`.Alias` construct against this
         :class:`.FunctionElement`.
@@ -493,7 +520,7 @@ class GenericFunction(util.with_metaclass(_GenericMeta, Function)):
     def __init__(self, *args, **kwargs):
         parsed_args = kwargs.pop('_parsed_args', None)
         if parsed_args is None:
-            parsed_args = [_literal_as_binds(c) for c in args]
+            parsed_args = [_literal_as_binds(c, self.name) for c in args]
         self.packagenames = []
         self._bind = kwargs.get('bind', None)
         self.clause_expr = ClauseList(
@@ -538,10 +565,10 @@ class ReturnTypeFromArgs(GenericFunction):
     """Define a function whose return type is the same as its arguments."""
 
     def __init__(self, *args, **kwargs):
-        args = [_literal_as_binds(c) for c in args]
+        args = [_literal_as_binds(c, self.name) for c in args]
         kwargs.setdefault('type_', _type_from_args(args))
         kwargs['_parsed_args'] = args
-        GenericFunction.__init__(self, *args, **kwargs)
+        super(ReturnTypeFromArgs, self).__init__(*args, **kwargs)
 
 
 class coalesce(ReturnTypeFromArgs):
@@ -589,7 +616,7 @@ class count(GenericFunction):
     def __init__(self, expression=None, **kwargs):
         if expression is None:
             expression = literal_column('*')
-        GenericFunction.__init__(self, expression, **kwargs)
+        super(count, self).__init__(expression, **kwargs)
 
 
 class current_date(AnsiFunction):
@@ -646,5 +673,123 @@ class array_agg(GenericFunction):
         args = [_literal_as_binds(c) for c in args]
         kwargs.setdefault('type_', sqltypes.Array(_type_from_args(args)))
         kwargs['_parsed_args'] = args
-        GenericFunction.__init__(self, *args, **kwargs)
+        super(array_agg, self).__init__(*args, **kwargs)
+
+
+class OrderedSetAgg(GenericFunction):
+    """Define a function where the return type is based on the sort
+    expression type as defined by the expression passed to the
+    :meth:`.FunctionElement.within_group` method."""
+
+    array_for_multi_clause = False
+
+    def within_group_type(self, within_group):
+        func_clauses = self.clause_expr.element
+        order_by = sqlutil.unwrap_order_by(within_group.order_by)
+        if self.array_for_multi_clause and len(func_clauses.clauses) > 1:
+            return sqltypes.Array(order_by[0].type)
+        else:
+            return order_by[0].type
+
+
+class mode(OrderedSetAgg):
+    """implement the ``mode`` ordered-set aggregate function.
+
+    This function must be used with the :meth:`.FunctionElement.within_group`
+    modifier to supply a sort expression to operate upon.
+
+    The return type of this function is the same as the sort expression.
+
+    .. versionadded:: 1.1
+
+    """
+
+
+class percentile_cont(OrderedSetAgg):
+    """implement the ``percentile_cont`` ordered-set aggregate function.
+
+    This function must be used with the :meth:`.FunctionElement.within_group`
+    modifier to supply a sort expression to operate upon.
+
+    The return type of this function is the same as the sort expression,
+    or if the arguments are an array, an :class:`.Array` of the sort
+    expression's type.
+
+    .. versionadded:: 1.1
+
+    """
+
+    array_for_multi_clause = True
+
+
+class percentile_disc(OrderedSetAgg):
+    """implement the ``percentile_disc`` ordered-set aggregate function.
+
+    This function must be used with the :meth:`.FunctionElement.within_group`
+    modifier to supply a sort expression to operate upon.
 
+    The return type of this function is the same as the sort expression,
+    or if the arguments are an array, an :class:`.Array` of the sort
+    expression's type.
+
+    .. versionadded:: 1.1
+
+    """
+
+    array_for_multi_clause = True
+
+
+class rank(GenericFunction):
+    """Implement the ``rank`` hypothetical-set aggregate function.
+
+    This function must be used with the :meth:`.FunctionElement.within_group`
+    modifier to supply a sort expression to operate upon.
+
+    The return type of this function is :class:`.Integer`.
+
+    .. versionadded:: 1.1
+
+    """
+    type = sqltypes.Integer()
+
+
+class dense_rank(GenericFunction):
+    """Implement the ``dense_rank`` hypothetical-set aggregate function.
+
+    This function must be used with the :meth:`.FunctionElement.within_group`
+    modifier to supply a sort expression to operate upon.
+
+    The return type of this function is :class:`.Integer`.
+
+    .. versionadded:: 1.1
+
+    """
+    type = sqltypes.Integer()
+
+
+class percent_rank(GenericFunction):
+    """Implement the ``percent_rank`` hypothetical-set aggregate function.
+
+    This function must be used with the :meth:`.FunctionElement.within_group`
+    modifier to supply a sort expression to operate upon.
+
+    The return type of this function is :class:`.Numeric`.
+
+    .. versionadded:: 1.1
+
+    """
+    type = sqltypes.Numeric()
+
+
+class cume_dist(GenericFunction):
+    """Implement the ``cume_dist`` hypothetical-set aggregate function.
+
+    This function must be used with the :meth:`.FunctionElement.within_group`
+    modifier to supply a sort expression to operate upon.
+
+    The return type of this function is :class:`.Numeric`.
+
+    .. versionadded:: 1.1
+
+    """
+    type = sqltypes.Numeric()
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index 8f502fc86..cbd74faac 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -154,6 +154,7 @@ def unwrap_order_by(clause):
     without DESC/ASC/NULLS FIRST/NULLS LAST"""
 
     cols = util.column_set()
+    result = []
     stack = deque([clause])
     while stack:
         t = stack.popleft()
@@ -166,11 +167,13 @@ def unwrap_order_by(clause):
                 t = t.element
             if isinstance(t, (_textual_label_reference)):
                 continue
-            cols.add(t)
+            if t not in cols:
+                cols.add(t)
+                result.append(t)
         else:
             for c in t.get_children():
                 stack.append(c)
-    return cols
+    return result
 
 
 def clause_is_present(clause, search):