DELETE and UPDATE with LIMIT and ORDER BY

diff --git a/doc/src/sgml/ref/delete.sgml b/doc/src/sgml/ref/delete.sgml
index 20417a1..1b4195c 100644
--- a/doc/src/sgml/ref/delete.sgml
+++ b/doc/src/sgml/ref/delete.sgml
@@ -25,6 +25,9 @@ PostgreSQL documentation
 DELETE FROM [ ONLY ] <replaceable class="PARAMETER">table_name</replaceable> [ * ] [ [ AS ] <replaceable class="parameter">alias</replaceable> ]
     [ USING <replaceable class="PARAMETER">using_list</replaceable> ]
     [ WHERE <replaceable class="PARAMETER">condition</replaceable> | WHERE CURRENT OF <replaceable class="PARAMETER">cursor_name</replaceable> ]
+    [ ORDER BY <replaceable class="parameter">expression</replaceable> [ ASC | DESC | USING <replaceable class="parameter">operator</replaceable> ] [ NULLS { FIRST | LAST } ] [, ...] 
+    [ LIMIT { <replaceable class="parameter">count</replaceable> | ALL } ]
+    [ OFFSET <replaceable class="parameter">start</replaceable> [ ROW | ROWS ] ] ]
     [ RETURNING * | <replaceable class="parameter">output_expression</replaceable> [ [ AS ] <replaceable class="parameter">output_name</replaceable> ] [, ...] ]
 </synopsis>
  </refsynopsisdiv>
@@ -143,6 +146,76 @@ DELETE FROM [ ONLY ] <replaceable class="PARAMETER">table_name</replaceable> [ *
     </listitem>
    </varlistentry>
 
+     <para>
+      If the <literal>ORDER BY</literal> clause is specified, the
+      deleted rows are sorted in the specified order.
+     </para>
+
+    <para>
+      Optionally one can add the key word <literal>ASC</> (ascending) or
+      <literal>DESC</> (descending) after any expression in the
+      <literal>ORDER BY</> clause.  If not specified, <literal>ASC</> is
+      assumed by default.  Alternatively, a specific ordering operator
+      name can be specified in the <literal>USING</> clause.
+      An ordering operator must be a less-than or greater-than
+      member of some B-tree operator family.
+      <literal>ASC</> is usually equivalent to <literal>USING &lt;</> and
+      <literal>DESC</> is usually equivalent to <literal>USING &gt;</>.
+      (But the creator of a user-defined data type can define exactly what the
+      default sort ordering is, and it might correspond to operators with other
+      names.)
+     </para>
+
+    <para>
+      If <literal>NULLS LAST</> is specified, null values sort after all
+      non-null values; if <literal>NULLS FIRST</> is specified, null values
+      sort before all non-null values.  If neither is specified, the default
+      behavior is <literal>NULLS LAST</> when <literal>ASC</> is specified
+      or implied, and <literal>NULLS FIRST</> when <literal>DESC</> is specified
+      (thus, the default is to act as though nulls are larger than non-nulls).
+      When <literal>USING</> is specified, the default nulls ordering depends
+      on whether the operator is a less-than or greater-than operator.
+    </para>
+
+    <para>
+      Note that ordering options apply only to the expression they follow;
+      for example <literal>ORDER BY x, y DESC</> does not mean
+      the same thing as <literal>ORDER BY x DESC, y DESC</>.
+    </para>
+
+    <para>
+      Character-string data is sorted according to the collation that applies
+      to the column being sorted.  That can be overridden at need by including
+      a <literal>COLLATE</> clause in the
+      <replaceable class="parameter">expression</replaceable>, for example
+      <literal>ORDER BY mycolumn COLLATE "en_US"</>.
+      For more information see <xref linkend="sql-syntax-collate-exprs"> and
+      <xref linkend="collation">.
+    </para>
+
+   <varlistentry>
+    <term><replaceable class="parameter">count</replaceable></term>
+    <term><replaceable class="parameter">start</replaceable></term>
+    <listitem>
+     <para>
+     <replaceable class="parameter">count</replaceable> specifies the
+     maximum number of rows to delete, while <replaceable
+     class="parameter">start</replaceable> specifies the number of rows
+     to skip before starting to delete rows.  When both are specified,
+     <replaceable class="parameter">start</replaceable> rows are skipped
+     before starting to count the <replaceable
+     class="parameter">count</replaceable> rows to be deleted.
+     </para>
+
+     <para>
+     If the <replaceable class="parameter">count</replaceable> expression
+     evaluates to NULL, it is treated as <literal>LIMIT ALL</>, i.e., no
+     limit.  If <replaceable class="parameter">start</replaceable> evaluates
+     to NULL, it is treated the same as <literal>OFFSET 0</>.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><replaceable class="PARAMETER">cursor_name</replaceable></term>
     <listitem>
diff --git a/doc/src/sgml/ref/update.sgml b/doc/src/sgml/ref/update.sgml
index 8a1619f..8d6f8d7 100644
--- a/doc/src/sgml/ref/update.sgml
+++ b/doc/src/sgml/ref/update.sgml
@@ -29,6 +29,9 @@ UPDATE [ ONLY ] <replaceable class="PARAMETER">table_name</replaceable> [ * ] [
         } [, ...]
     [ FROM <replaceable class="PARAMETER">from_list</replaceable> ]
     [ WHERE <replaceable class="PARAMETER">condition</replaceable> | WHERE CURRENT OF <replaceable class="PARAMETER">cursor_name</replaceable> ]
+    [ ORDER BY <replaceable class="parameter">expression</replaceable> [ ASC | DESC | USING <replaceable class="parameter">operator</replaceable> ] [ NULLS { FIRST | LAST } ] [, ...] 
+    [ LIMIT { <replaceable class="parameter">count</replaceable> | ALL } ]
+    [ OFFSET <replaceable class="parameter">start</replaceable> [ ROW | ROWS ] ] ]
     [ RETURNING * | <replaceable class="parameter">output_expression</replaceable> [ [ AS ] <replaceable class="parameter">output_name</replaceable> ] [, ...] ]
 </synopsis>
  </refsynopsisdiv>
@@ -190,6 +193,76 @@ UPDATE [ ONLY ] <replaceable class="PARAMETER">table_name</replaceable> [ * ] [
     </listitem>
    </varlistentry>
 
+     <para>
+      If the <literal>ORDER BY</literal> clause is specified, the
+      updated rows are sorted in the specified order.
+     </para>
+
+    <para>
+      Optionally one can add the key word <literal>ASC</> (ascending) or
+      <literal>DESC</> (descending) after any expression in the
+      <literal>ORDER BY</> clause.  If not specified, <literal>ASC</> is
+      assumed by default.  Alternatively, a specific ordering operator
+      name can be specified in the <literal>USING</> clause.
+      An ordering operator must be a less-than or greater-than
+      member of some B-tree operator family.
+      <literal>ASC</> is usually equivalent to <literal>USING &lt;</> and
+      <literal>DESC</> is usually equivalent to <literal>USING &gt;</>.
+      (But the creator of a user-defined data type can define exactly what the
+      default sort ordering is, and it might correspond to operators with other
+      names.)
+     </para>
+
+    <para>
+      If <literal>NULLS LAST</> is specified, null values sort after all
+      non-null values; if <literal>NULLS FIRST</> is specified, null values
+      sort before all non-null values.  If neither is specified, the default
+      behavior is <literal>NULLS LAST</> when <literal>ASC</> is specified
+      or implied, and <literal>NULLS FIRST</> when <literal>DESC</> is specified
+      (thus, the default is to act as though nulls are larger than non-nulls).
+      When <literal>USING</> is specified, the default nulls ordering depends
+      on whether the operator is a less-than or greater-than operator.
+    </para>
+
+    <para>
+      Note that ordering options apply only to the expression they follow;
+      for example <literal>ORDER BY x, y DESC</> does not mean
+      the same thing as <literal>ORDER BY x DESC, y DESC</>.
+    </para>
+
+    <para>
+      Character-string data is sorted according to the collation that applies
+      to the column being sorted.  That can be overridden at need by including
+      a <literal>COLLATE</> clause in the
+      <replaceable class="parameter">expression</replaceable>, for example
+      <literal>ORDER BY mycolumn COLLATE "en_US"</>.
+      For more information see <xref linkend="sql-syntax-collate-exprs"> and
+      <xref linkend="collation">.
+    </para>
+
+   <varlistentry>
+    <term><replaceable class="parameter">count</replaceable></term>
+    <term><replaceable class="parameter">start</replaceable></term>
+    <listitem>
+     <para>
+     <replaceable class="parameter">count</replaceable> specifies the
+     maximum number of rows to update, while <replaceable
+     class="parameter">start</replaceable> specifies the number of rows
+     to skip before starting to update rows.  When both are specified,
+     <replaceable class="parameter">start</replaceable> rows are skipped
+     before starting to count the <replaceable
+     class="parameter">count</replaceable> rows to be updated.
+     </para>
+
+     <para>
+     If the <replaceable class="parameter">count</replaceable> expression
+     evaluates to NULL, it is treated as <literal>LIMIT ALL</>, i.e., no
+     limit.  If <replaceable class="parameter">start</replaceable> evaluates
+     to NULL, it is treated the same as <literal>OFFSET 0</>.
+     </para>
+    </listitem>
+   </varlistentry>
+
    <varlistentry>
     <term><replaceable class="PARAMETER">cursor_name</replaceable></term>
     <listitem>
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index 00a0fed..1a8dfc6 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -2996,6 +2996,9 @@ _copyDeleteStmt(const DeleteStmt *from)
 	COPY_NODE_FIELD(relation);
 	COPY_NODE_FIELD(usingClause);
 	COPY_NODE_FIELD(whereClause);
+	COPY_NODE_FIELD(sortClause);
+	COPY_NODE_FIELD(limitOffset);
+	COPY_NODE_FIELD(limitCount);
 	COPY_NODE_FIELD(returningList);
 	COPY_NODE_FIELD(withClause);
 
@@ -3011,6 +3014,9 @@ _copyUpdateStmt(const UpdateStmt *from)
 	COPY_NODE_FIELD(targetList);
 	COPY_NODE_FIELD(whereClause);
 	COPY_NODE_FIELD(fromClause);
+	COPY_NODE_FIELD(sortClause);
+	COPY_NODE_FIELD(limitOffset);
+	COPY_NODE_FIELD(limitCount);
 	COPY_NODE_FIELD(returningList);
 	COPY_NODE_FIELD(withClause);
 
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 46573ae..2136be3 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -1023,6 +1023,9 @@ _equalDeleteStmt(const DeleteStmt *a, const DeleteStmt *b)
 	COMPARE_NODE_FIELD(relation);
 	COMPARE_NODE_FIELD(usingClause);
 	COMPARE_NODE_FIELD(whereClause);
+	COMPARE_NODE_FIELD(sortClause);
+	COMPARE_NODE_FIELD(limitOffset);
+	COMPARE_NODE_FIELD(limitCount);
 	COMPARE_NODE_FIELD(returningList);
 	COMPARE_NODE_FIELD(withClause);
 
@@ -1036,6 +1039,9 @@ _equalUpdateStmt(const UpdateStmt *a, const UpdateStmt *b)
 	COMPARE_NODE_FIELD(targetList);
 	COMPARE_NODE_FIELD(whereClause);
 	COMPARE_NODE_FIELD(fromClause);
+	COMPARE_NODE_FIELD(sortClause);
+	COMPARE_NODE_FIELD(limitOffset);
+	COMPARE_NODE_FIELD(limitCount);
 	COMPARE_NODE_FIELD(returningList);
 	COMPARE_NODE_FIELD(withClause);
 
diff --git a/src/backend/nodes/nodeFuncs.c b/src/backend/nodes/nodeFuncs.c
index 3e8189c..3c73295 100644
--- a/src/backend/nodes/nodeFuncs.c
+++ b/src/backend/nodes/nodeFuncs.c
@@ -3403,6 +3403,12 @@ raw_expression_tree_walker(Node *node,
 					return true;
 				if (walker(stmt->whereClause, context))
 					return true;
+				if (walker(stmt->sortClause, context))
+					return true;
+				if (walker(stmt->limitOffset, context))
+					return true;
+				if (walker(stmt->limitCount, context))
+					return true;
 				if (walker(stmt->returningList, context))
 					return true;
 				if (walker(stmt->withClause, context))
@@ -3421,6 +3427,12 @@ raw_expression_tree_walker(Node *node,
 					return true;
 				if (walker(stmt->fromClause, context))
 					return true;
+				if (walker(stmt->sortClause, context))
+					return true;
+				if (walker(stmt->limitOffset, context))
+					return true;
+				if (walker(stmt->limitCount, context))
+					return true;
 				if (walker(stmt->returningList, context))
 					return true;
 				if (walker(stmt->withClause, context))
diff --git a/src/backend/parser/analyze.c b/src/backend/parser/analyze.c
index 567dd54..fca98b3 100644
--- a/src/backend/parser/analyze.c
+++ b/src/backend/parser/analyze.c
@@ -432,6 +432,19 @@ transformDeleteStmt(ParseState *pstate, DeleteStmt *stmt)
 	qual = transformWhereClause(pstate, stmt->whereClause,
 								EXPR_KIND_WHERE, "WHERE");
 
+	qry->sortClause = transformSortClause(pstate,
+										  stmt->sortClause,
+										  &qry->targetList,
+										  EXPR_KIND_ORDER_BY,
+										  false /* allow SQL92 rules */ );
+
+	/* transform LIMIT */
+	qry->limitOffset = transformLimitClause(pstate, stmt->limitOffset,
+											EXPR_KIND_OFFSET, "OFFSET");
+	qry->limitCount = transformLimitClause(pstate, stmt->limitCount,
+										   EXPR_KIND_LIMIT, "LIMIT");
+
+
 	qry->returningList = transformReturningList(pstate, stmt->returningList);
 
 	/* done building the range table and jointree */
@@ -2246,6 +2259,17 @@ transformUpdateStmt(ParseState *pstate, UpdateStmt *stmt)
 	 */
 	qry->targetList = transformUpdateTargetList(pstate, stmt->targetList);
 
+	qry->sortClause = transformSortClause(pstate,
+										  stmt->sortClause,
+										  &qry->targetList,
+										  EXPR_KIND_ORDER_BY,
+										  false /* allow SQL92 rules */ );
+	/* transform LIMIT */
+	qry->limitOffset = transformLimitClause(pstate, stmt->limitOffset,
+											EXPR_KIND_OFFSET, "OFFSET");
+	qry->limitCount = transformLimitClause(pstate, stmt->limitCount,
+										   EXPR_KIND_LIMIT, "LIMIT");
+
 	qry->rtable = pstate->p_rtable;
 	qry->jointree = makeFromExpr(pstate->p_joinlist, qual);
 
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index 89d2836..de446ff 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -396,7 +396,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 				relation_expr_list dostmt_opt_list
 				transform_element_list transform_type_list
 				TriggerTransitions TriggerReferencing
-				publication_name_list
+				publication_name_list opt_addtional_criteria
 
 %type <list>	group_by_list
 %type <node>	group_by_item empty_grouping_set rollup_clause cube_clause
@@ -10632,13 +10632,16 @@ returning_clause:
  *****************************************************************************/
 
 DeleteStmt: opt_with_clause DELETE_P FROM relation_expr_opt_alias
-			using_clause where_or_current_clause returning_clause
+			using_clause where_or_current_clause opt_addtional_criteria returning_clause
 				{
 					DeleteStmt *n = makeNode(DeleteStmt);
 					n->relation = $4;
 					n->usingClause = $5;
 					n->whereClause = $6;
-					n->returningList = $7;
+					n->limitCount = list_nth($7, 1);
+					n->limitOffset = list_nth($7, 2);
+					n->sortClause =list_nth($7, 0);
+					n->returningList = $8;
 					n->withClause = $1;
 					$$ = (Node *)n;
 				}
@@ -10704,6 +10707,7 @@ UpdateStmt: opt_with_clause UPDATE relation_expr_opt_alias
 			SET set_clause_list
 			from_clause
 			where_or_current_clause
+			opt_addtional_criteria
 			returning_clause
 				{
 					UpdateStmt *n = makeNode(UpdateStmt);
@@ -10711,7 +10715,10 @@ UpdateStmt: opt_with_clause UPDATE relation_expr_opt_alias
 					n->targetList = $5;
 					n->fromClause = $6;
 					n->whereClause = $7;
-					n->returningList = $8;
+					n->limitCount = list_nth($8, 1);
+					n->limitOffset = list_nth($8, 2);
+					n->sortClause =list_nth($8, 0);
+					n->returningList = $9;
 					n->withClause = $1;
 					$$ = (Node *)n;
 				}
@@ -11253,6 +11260,13 @@ select_offset_value:
 			a_expr									{ $$ = $1; }
 		;
 
+opt_addtional_criteria:
+			sort_clause limit_clause offset_clause                   { $$ = list_make3($1, $2, $3); }
+			| sort_clause  offset_clause                   { $$ = list_make3($1, NULL, $2); }
+			| sort_clause limit_clause                    { $$ = list_make3($1, $2, NULL); }
+			| /*EMPTY*/					{ $$ = list_make3(NULL, NULL, NULL); }
+		;
+
 /*
  * Allowing full expressions without parentheses causes various parsing
  * problems with the trailing ROW/ROWS key words.  SQL only calls for
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 9f57388..c327303 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -1427,6 +1427,9 @@ typedef struct DeleteStmt
 	RangeVar   *relation;		/* relation to delete from */
 	List	   *usingClause;	/* optional using clause for more tables */
 	Node	   *whereClause;	/* qualifications */
+	List	   *sortClause;		/* sort clause (a list of SortBy's) */
+	Node	   *limitOffset;	/* # of result tuples to skip */
+	Node	   *limitCount;		/* # of result tuples to delete */
 	List	   *returningList;	/* list of expressions to return */
 	WithClause *withClause;		/* WITH clause */
 } DeleteStmt;
@@ -1442,6 +1445,9 @@ typedef struct UpdateStmt
 	List	   *targetList;		/* the target list (of ResTarget) */
 	Node	   *whereClause;	/* qualifications */
 	List	   *fromClause;		/* optional from clause for more tables */
+	List	   *sortClause;		/* sort clause (a list of SortBy's) */
+	Node	   *limitOffset;	/* # of result tuples to skip */
+	Node	   *limitCount;		/* # of result tuples to update */
 	List	   *returningList;	/* list of expressions to return */
 	WithClause *withClause;		/* WITH clause */
 } UpdateStmt;
diff --git a/src/test/regress/expected/delete.out b/src/test/regress/expected/delete.out
index e7eb328..ccb1c0b 100644
--- a/src/test/regress/expected/delete.out
+++ b/src/test/regress/expected/delete.out
@@ -30,4 +30,25 @@ SELECT id, a, char_length(b) FROM delete_test;
   1 | 10 |            
 (1 row)
 
+--
+-- Test ORDER BY with limit clause options
+--
+INSERT INTO delete_test (a, b) VALUES (1,'x'),(1,'xx'),(1,'xxx'),(1,'xxxx');
+DELETE FROM delete_test  WHERE a=1 ORDER BY id LIMIT 2 OFFSET 1;
+SELECT * FROM delete_test;
+ id | a  |  b   
+----+----+------
+  1 | 10 | 
+  4 |  1 | x
+  7 |  1 | xxxx
+(3 rows)
+
+DELETE FROM delete_test  WHERE a=1 ORDER BY id LIMIT 1;
+SELECT * FROM delete_test;
+ id | a  |  b   
+----+----+------
+  1 | 10 | 
+  7 |  1 | xxxx
+(2 rows)
+
 DROP TABLE delete_test;
diff --git a/src/test/regress/expected/update.out b/src/test/regress/expected/update.out
index 9366f04..36888e4 100644
--- a/src/test/regress/expected/update.out
+++ b/src/test/regress/expected/update.out
@@ -196,6 +196,31 @@ INSERT INTO upsert_test VALUES (1, 'Bat') ON CONFLICT(a)
  1 | Foo, Correlated, Excluded
 (1 row)
 
+--
+-- Test ORDER BY with limit clause options
+--
+delete from update_test;
+INSERT INTO update_test (b, c) VALUES (1,'x'),(2,'xx'),(3,'xxx'),(4,'xxxx');
+UPDATE update_test SET c ='xxxxx'  WHERE a = 10 ORDER BY b LIMIT 2 OFFSET 1;
+SELECT * FROM update_test;
+ a  | b |   c   
+----+---+-------
+ 10 | 1 | x
+ 10 | 4 | xxxx
+ 10 | 2 | xxxxx
+ 10 | 3 | xxxxx
+(4 rows)
+
+UPDATE update_test SET c ='xxxxx'  WHERE a = 10 ORDER BY b LIMIT 1;
+SELECT * FROM update_test;
+ a  | b |   c   
+----+---+-------
+ 10 | 4 | xxxx
+ 10 | 2 | xxxxx
+ 10 | 3 | xxxxx
+ 10 | 1 | xxxxx
+(4 rows)
+
 DROP TABLE update_test;
 DROP TABLE upsert_test;
 -- update to a partition should check partition bound constraint for the new tuple
diff --git a/src/test/regress/sql/delete.sql b/src/test/regress/sql/delete.sql
index d8cb99e..bf19390 100644
--- a/src/test/regress/sql/delete.sql
+++ b/src/test/regress/sql/delete.sql
@@ -22,4 +22,17 @@ DELETE FROM delete_test WHERE a > 25;
 
 SELECT id, a, char_length(b) FROM delete_test;
 
+--
+-- Test ORDER BY with limit clause options
+--
+INSERT INTO delete_test (a, b) VALUES (1,'x'),(1,'xx'),(1,'xxx'),(1,'xxxx');
+
+DELETE FROM delete_test  WHERE a=1 ORDER BY id LIMIT 2 OFFSET 1;
+
+SELECT * FROM delete_test;
+
+DELETE FROM delete_test  WHERE a=1 ORDER BY id LIMIT 1;
+
+SELECT * FROM delete_test;
+
 DROP TABLE delete_test;
diff --git a/src/test/regress/sql/update.sql b/src/test/regress/sql/update.sql
index 6637119..7f780d1 100644
--- a/src/test/regress/sql/update.sql
+++ b/src/test/regress/sql/update.sql
@@ -104,6 +104,21 @@ INSERT INTO upsert_test VALUES (1, 'Bat') ON CONFLICT(a)
   DO UPDATE SET (b, a) = (SELECT b || ', Excluded', a from upsert_test i WHERE i.a = excluded.a)
   RETURNING *;
 
+--
+-- Test ORDER BY with limit clause options
+--
+delete from update_test;
+
+INSERT INTO update_test (b, c) VALUES (1,'x'),(2,'xx'),(3,'xxx'),(4,'xxxx');
+
+UPDATE update_test SET c ='xxxxx'  WHERE a = 10 ORDER BY b LIMIT 2 OFFSET 1;
+
+SELECT * FROM update_test;
+
+UPDATE update_test SET c ='xxxxx'  WHERE a = 10 ORDER BY b LIMIT 1;
+
+SELECT * FROM update_test;
+
 DROP TABLE update_test;
 DROP TABLE upsert_test;