From 991d804042bc15fe2ac6c6919bd1295ab5ae5a24 Mon Sep 17 00:00:00 2001
From: Alexandre Felipe <o.alexandre.felipe@gmail.com>
Date: Fri, 10 Jul 2026 11:45:13 +0100
Subject: [PATCH-v15.1 4/6] SLOPE: Planner support

Update the planner to take advantage of order of an expression
that can be inferred from the order of a subexpression.

If f(x) is a monotonic function and x is known to be ordered,
we can infeer the order of f(x) from the order of x.
The analysis is performed in two stages
1. During plan creation, for every pathkey expression computes
   the source of variation, defined as the innermost subexpression
   that causes all variation in the expression.
   if the source of variaiton is not the full expression, and
   depends on a single table, a slope_info entry is created.
2. on build_index_pathkeys, in addition to the usual pathkeys
   create pathkeys for slope_info entries where the index key
   is the source of variation.

If a query is ordered by multiple monotonic functions on x,
the lexicographical order is satisfied by simply using the
order of x. If any prefix turns out to produce equal values
for different values of x, the next column is still guaranteed
to be ordered within each constant prefix group.

If an index satisfies the equalimage condition, f(x) appearing
after x don't matter for the query order.

Changes:
- Add SlopeInfo struct to cache monotonicity information per pathkey
- Add precompute_slope_cache() called during plan construction
- Add get_variation_source() to find the innermost varying expression
- Add get_expr_slope_wrt() to verify monotonicity by calling prosupport
- Add indexcol_is_equalimage() to check if x1 = x2 implies f(x1) = f(x2)
- Modify build_index_pathkeys() to check for monotonicity on index keys
- Handle both increasing and decreasing monotonic functions, with
  reversed pathkey emission for decreasing functions so that backward
  index scans are correctly selected

Also adds the prosupport function entries to pg_proc.dat
- arg0_asc_slope_support (OID 9954)
- arg0_desc_slope_support (OID 9955)
- arg1_asc_slope_support (OID 9956)
- diff_slope_support (OID 9957)
- addition_slope_support (OID 9958)
- multiply_slope_support (OID 9959)
- divide_slope_support (OID 9960)
---
 src/backend/optimizer/path/pathkeys.c         | 531 ++++++++++-
 src/backend/optimizer/plan/planner.c          |   3 +
 src/backend/utils/fmgr/slopesupport.c         |  51 +-
 src/backend/utils/misc/guc_parameters.dat     |   8 +
 src/backend/utils/misc/postgresql.conf.sample |   1 +
 src/include/nodes/pathnodes.h                 |  10 +
 src/include/nodes/plannodes.h                 |   7 +
 src/include/optimizer/paths.h                 |   2 +
 src/test/regress/expected/slope.out           | 862 ++++++++++++++++++
 src/test/regress/expected/sysviews.out        |   3 +-
 src/test/regress/parallel_schedule            |   6 +-
 src/test/regress/sql/slope.sql                | 539 +++++++++++
 12 files changed, 1992 insertions(+), 31 deletions(-)
 create mode 100644 src/test/regress/expected/slope.out
 create mode 100644 src/test/regress/sql/slope.sql

diff --git a/src/backend/optimizer/path/pathkeys.c b/src/backend/optimizer/path/pathkeys.c
index 3610b6d9694..b7629bc9f92 100644
--- a/src/backend/optimizer/path/pathkeys.c
+++ b/src/backend/optimizer/path/pathkeys.c
@@ -17,26 +17,42 @@
  */
 #include "postgres.h"
 
+#include "access/nbtree.h"
 #include "access/stratnum.h"
+#include "access/transam.h"
+#include "catalog/pg_am.h"
 #include "catalog/pg_opfamily.h"
+#include "catalog/pg_type.h"
+#include "commands/defrem.h"
+#include "fmgr.h"
 #include "nodes/nodeFuncs.h"
+#include "nodes/supportnodes.h"
 #include "optimizer/cost.h"
 #include "optimizer/optimizer.h"
 #include "optimizer/pathnode.h"
 #include "optimizer/paths.h"
+#include "parser/parse_oper.h"
 #include "partitioning/partbounds.h"
 #include "rewrite/rewriteManip.h"
 #include "utils/lsyscache.h"
+#include "utils/typcache.h"
 
 /* Consider reordering of GROUP BY keys? */
 bool		enable_group_by_reordering = true;
 
+/* Perform slope analysis to identify indirectly sorted data */
+bool		enable_slope = true;
 static bool pathkey_is_redundant(PathKey *new_pathkey, List *pathkeys);
 static bool matches_boolean_partition_clause(RestrictInfo *rinfo,
 											 RelOptInfo *partrel,
 											 int partkeycol);
 static Var *find_var_for_subquery_tle(RelOptInfo *rel, TargetEntry *tle);
 static bool right_merge_direction(PlannerInfo *root, PathKey *pathkey);
+static MonotonicFunction get_expr_slope_wrt(Expr *expr, Expr *target);
+static bool indexcol_is_equalimage(IndexOptInfo *index, int colno);
+static PathKey *slope_emit_pathkey(PlannerInfo *root, PathKey *pk,
+								   Expr *indexkey, bool reverse_sort,
+								   bool nulls_first);
 
 
 /****************************************************************************
@@ -738,7 +754,6 @@ get_cheapest_fractional_path_for_pathkeys(List *paths,
 	return matched_path;
 }
 
-
 /*
  * get_cheapest_parallel_safe_total_inner
  *	  Find the unparameterized parallel-safe path with the least total cost.
@@ -760,6 +775,386 @@ get_cheapest_parallel_safe_total_inner(List *paths)
 	return NULL;
 }
 
+/*
+ * get_variation_source
+ *	  Find the source of variation in an expression.
+ *
+ * Descends through function calls to find the innermost non-constant
+ * expression that determines the variation of the whole expression.
+ * For f(x) returns x.  For f(g(x)) returns x.  For f(x, y) returns f(x, y).
+ * For a plain Var, returns the Var itself.
+ *
+ * This is a cheap extraction that doesn't check monotonicity - that's
+ * deferred until we find an index column matching the variation source.
+ * Also extracts the relid if all Vars are from the same table.
+ */
+static void
+get_variation_source(Expr *expr, PathKey *pk)
+{
+	pk->pk_var = NULL;
+	pk->pk_varrelid = 0;
+
+	for (;;)
+	{
+		List	   *args;
+		Expr	   *non_const_arg = NULL;
+		int			non_const_count = 0;
+		ListCell   *lc;
+
+		/* Stop at RelabelType (no-op coercion) */
+		if (IsA(expr, RelabelType))
+		{
+			pk->pk_var = expr;
+			pk->pk_varrelid = 0;
+			return;
+		}
+
+		/*
+		 * Handle FuncExpr, implicit casts are considered safe all the rest is
+		 * subject to prosupport checks
+		 */
+		if (IsA(expr, FuncExpr))
+			args = ((FuncExpr *) expr)->args;
+		else if (IsA(expr, OpExpr))
+			args = ((OpExpr *) expr)->args;
+		else if (IsA(expr, Var))
+		{
+			/* Reached a Var - this is our inner expression */
+			pk->pk_var = expr;
+			pk->pk_varrelid = ((Var *) expr)->varno;
+			return;
+		}
+		else
+		{
+			/* Unsupported node type */
+			return;
+		}
+
+		/* Find non-constant arguments */
+		foreach(lc, args)
+		{
+			Expr	   *arg = (Expr *) lfirst(lc);
+
+			if (!IsA(arg, Const))
+			{
+				non_const_count++;
+				if (non_const_count > 1)
+				{
+					/* Multivariate - return this expression as inner */
+					pk->pk_var = expr;
+					pk->pk_varrelid = 0;	/* unknown, will use equal() */
+					return;
+				}
+				non_const_arg = arg;
+			}
+		}
+
+		if (non_const_arg == NULL)
+		{
+			/* All constant - no inner expression */
+			return;
+		}
+
+		expr = non_const_arg;
+	}
+}
+
+
+/*
+ * Returns true if the expression's type may hold NaN values.
+ */
+static bool
+expr_can_nan(Expr *target)
+{
+
+	Oid			expr_type;
+
+	/* Unwrap RelabelType (no-op coercion) */
+	while (target != NULL && IsA(target, RelabelType))
+		target = (Expr *) ((RelabelType *) target)->arg;
+
+	/* Extract type from the node */
+	if (IsA(target, Var))
+		expr_type = ((Var *) target)->vartype;
+	else if (IsA(target, FuncExpr))
+		expr_type = ((FuncExpr *) target)->funcresulttype;
+	else if (IsA(target, OpExpr))
+		expr_type = ((OpExpr *) target)->opresulttype;
+	else
+		return false;
+
+	/* Get the base type for domains */
+	if (expr_type >= FirstNormalObjectId)
+		expr_type = getBaseType(expr_type);
+
+	/* Check if the type can hold NaN values */
+	switch (expr_type)
+	{
+		case FLOAT4OID:
+		case FLOAT8OID:
+		case NUMERICOID:
+			return true;
+		default:
+			return false;
+	}
+}
+
+/*
+ * get_expr_slope_wrt
+ *	  Determine the monotonicity slope of an expression with respect to
+ *	  a specific target subexpression, under each type's default btree
+ *	  order.  Callers that match this to a PathKey / index must separately
+ *	  verify those use the corresponding default opfamilies.
+ *
+ * Returns the slope of 'expr' with respect to 'target'
+ *   MONOTONICFUNC_INCREASING: monotonically increasing
+ *   MONOTONICFUNC_DECREASING: monotonically decreasing
+ *   MONOTONICFUNC_BOTH:       independent of target (constant)
+ *   MONOTONICFUNC_NONE:       cannot determine monotonicity
+ */
+static MonotonicFunction
+get_expr_slope_wrt(Expr *expr, Expr *target)
+{
+	MonotonicFunction slope = MONOTONICFUNC_INCREASING;
+
+	for (;;)
+	{
+		Oid			funcid;
+		List	   *args;
+		Oid			prosupport;
+		SupportRequestMonotonic req;
+		ListCell   *lc;
+		int			i;
+		Expr	   *next_expr = NULL;
+		MonotonicFunction func_arg_slope = MONOTONICFUNC_INCREASING;
+
+		/* Check if we've reached the target */
+		if (equal(expr, target))
+		{
+			if (slope == MONOTONICFUNC_INCREASING)
+				return slope;
+
+			/*
+			 * A decreasing function does not relocate NaN values, so index
+			 * scan ordering (forward or backward) cannot match ORDER BY f(x)
+			 * on float/numeric columns.
+			 */
+			if (expr_can_nan(target))
+			{
+				if (slope == MONOTONICFUNC_DECREASING)
+					return MONOTONICFUNC_NONE;
+				else if (slope == MONOTONICFUNC_BOTH)
+					return MONOTONICFUNC_INCREASING;
+			}
+			return slope;
+		}
+
+		/*
+		 * RelabelType, casts with 'b' cast method, are bit-preserving, but
+		 * types might have different semantic.
+		 */
+		if (IsA(expr, RelabelType))
+			return MONOTONICFUNC_NONE;
+
+		/*
+		 * Handle FuncExpr and OpExpr this covers explicit function calls,
+		 * operators, and casts with 'f' cast method. (both implicit and
+		 * explicit), requires slope analysis
+		 */
+		if (IsA(expr, FuncExpr))
+		{
+			FuncExpr   *fexpr = (FuncExpr *) expr;
+
+			funcid = fexpr->funcid;
+			args = fexpr->args;
+		}
+		else if (IsA(expr, OpExpr))
+		{
+			OpExpr	   *opexpr = (OpExpr *) expr;
+
+			set_opfuncid(opexpr);
+			funcid = opexpr->opfuncid;
+			args = opexpr->args;
+		}
+		else
+		{
+			/* Reached a leaf without finding target */
+			return MONOTONICFUNC_NONE;
+		}
+
+		/* Check for prosupport function */
+		prosupport = get_func_support(funcid);
+		if (!OidIsValid(prosupport))
+			return MONOTONICFUNC_NONE;
+
+		/* Call prosupport to get slope pattern */
+		req.type = T_SupportRequestMonotonic;
+		req.expr = (Node *) expr;
+		req.slopes = NULL;
+		req.nslopes = 0;
+
+		if (DatumGetPointer(OidFunctionCall1(prosupport, PointerGetDatum(&req))) == NULL)
+			return MONOTONICFUNC_NONE;
+
+		if (req.slopes == NULL || req.nslopes <= 0)
+			return MONOTONICFUNC_NONE;
+
+		/* Find the single non-constant argument */
+		i = 0;
+		foreach(lc, args)
+		{
+			Expr	   *arg = (Expr *) lfirst(lc);
+
+			if (!IsA(arg, Const))
+			{
+				if (next_expr != NULL)
+				{
+					/* Multivariate - check if this is the target */
+					return equal(expr, target) ? slope : MONOTONICFUNC_NONE;
+				}
+				next_expr = arg;
+				if (likely(i < req.nslopes))
+				{
+					if (req.slopes[i] == MONOTONICFUNC_DECREASING)
+						func_arg_slope = MONOTONICFUNC_DECREASING;
+					else if (req.slopes[i] != MONOTONICFUNC_INCREASING)
+						return MONOTONICFUNC_NONE;
+				}
+				else
+					return MONOTONICFUNC_NONE;
+			}
+			i++;
+		}
+
+		if (next_expr == NULL)
+			return MONOTONICFUNC_NONE;	/* all constant */
+
+		/* Compose slopes */
+		if (func_arg_slope == MONOTONICFUNC_DECREASING)
+		{
+			slope = (slope == MONOTONICFUNC_INCREASING) ?
+				MONOTONICFUNC_DECREASING : MONOTONICFUNC_INCREASING;
+		}
+
+		expr = next_expr;
+	}
+}
+
+/*
+ * indexcol_is_equalimage
+ *	  Does btree equality on this index column imply image equality?
+ */
+static bool
+indexcol_is_equalimage(IndexOptInfo *index, int colno)
+{
+	Oid			equalimageproc;
+
+	/*
+	 * equalimage(opcintype oid) returns bool returns static information about
+	 * an operator class and collation.
+	 *
+	 */
+	equalimageproc = get_opfamily_proc(index->sortopfamily[colno],
+									   index->opcintype[colno],
+									   index->opcintype[colno],
+									   BTEQUALIMAGE_PROC);
+
+	/*
+	 * Returning true indicates that the the order function for the operator
+	 * class is guaranteed to only return 0 if the values being compared are
+	 * interchangeable. Not registering an equalimage function or returning
+	 * false indicates that this condition cannot be assumed to hold.
+	 */
+	if (!OidIsValid(equalimageproc))
+		return false;
+
+	return DatumGetBool(OidFunctionCall1Coll(
+											 equalimageproc,
+											 index->indexcollations[colno],
+											 ObjectIdGetDatum(index->opcintype[colno])
+											 ));
+}
+
+/*
+ * precompute_slope_pathkeys
+ *	  For each query pathkey, extract the source of variation and store
+ *	  it directly on the PathKey (pk_var, pk_varrelid).
+ *
+ * Called once after query_pathkeys is set.  Pathkeys whose expression
+ * is a plain Var or that have no usable variation source are left with
+ * pk_var = NULL.  Monotonicity (pk_slope) is computed lazily on first
+ * index match.
+ */
+void
+precompute_slope_pathkeys(PlannerInfo *root)
+{
+	ListCell   *lc;
+
+	foreach(lc, root->query_pathkeys)
+	{
+		PathKey    *pk = lfirst_node(PathKey, lc);
+		EquivalenceMember *em;
+
+		pk->pk_var = NULL;
+		pk->pk_varrelid = 0;
+		pk->pk_slope = MONOTONICFUNC_UNSET;
+
+		if (!enable_slope)
+			continue;
+
+		if (pk->pk_eclass->ec_has_volatile ||
+			pk->pk_eclass->ec_members == NIL)
+			continue;
+
+		em = linitial(pk->pk_eclass->ec_members);
+
+		if (IsA(em->em_expr, Var))
+			continue;
+
+		get_variation_source(em->em_expr, pk);
+	}
+}
+
+/*
+ * slope_emit_pathkey
+ *	  Return the canonical pathkey for what a forward index scan actually
+ *	  produces for an expression that is monotonic in the index column.
+ *
+ *	  The result reflects both the direction and null ordering that the
+ *	  forward scan generates.  f(NULL) is NULL, so nulls appear at the
+ *	  position dictated by the index's null ordering.
+ *
+ *	  Returns NULL if pk_slope indicates no monotonicity.
+ *
+ *	   index   function   pathkey
+ *	   ASC     ASC        ASC
+ *	   ASC     DESC       DESC
+ *	   DESC    ASC        DESC
+ *	   DESC    DESC       ASC
+ */
+static PathKey *
+slope_emit_pathkey(PlannerInfo *root,
+				   PathKey *pk,
+				   Expr *indexkey,
+				   bool reverse_sort,
+				   bool nulls_first)
+{
+	MonotonicFunction slope;
+	bool		produces_desc;
+
+	slope = (MonotonicFunction) pk->pk_slope;
+	if (slope == MONOTONICFUNC_NONE)
+		return NULL;
+
+	produces_desc = (reverse_sort != (slope == MONOTONICFUNC_DECREASING));
+
+	return make_canonical_pathkey(root,
+								  pk->pk_eclass,
+								  pk->pk_opfamily,
+								  produces_desc ? COMPARE_GT : COMPARE_LT,
+								  nulls_first);
+}
+
 /****************************************************************************
  *		NEW PATHKEY FORMATION
  ****************************************************************************/
@@ -790,11 +1185,18 @@ build_index_pathkeys(PlannerInfo *root,
 {
 	List	   *retval = NIL;
 	ListCell   *lc;
+	ListCell   *query_pk_cell;
+	bool		slope_match;
 	int			i;
 
 	if (index->sortopfamily == NULL)
 		return NIL;				/* non-orderable index */
 
+	slope_match = (enable_slope &&
+				   index->rel->reloptkind == RELOPT_BASEREL &&
+				   root->query_pathkeys != NIL);
+	query_pk_cell = slope_match ? list_head(root->query_pathkeys) : NULL;
+
 	i = 0;
 	foreach(lc, index->indextlist)
 	{
@@ -803,6 +1205,11 @@ build_index_pathkeys(PlannerInfo *root,
 		bool		reverse_sort;
 		bool		nulls_first;
 		PathKey    *cpathkey;
+		PathKey    *spk;
+		bool		column_pinned = false;
+		bool		pathkey_emitted = false;
+		bool		pinned_is_equalimage = false;
+
 
 		/*
 		 * INCLUDE columns are stored in index unordered, so they don't
@@ -832,7 +1239,122 @@ build_index_pathkeys(PlannerInfo *root,
 											  index->rel->relids,
 											  false);
 
-		if (cpathkey)
+		/*
+		 * If the first unmatched query pathkey is a monotonic function of
+		 * this index column, use that pathkey instead of the column's own
+		 * pathkey so the index can satisfy the query ordering without a Sort.
+		 *
+		 * If the pathkey is a monotonic function of the index key, add it to
+		 * the result list and move to a non-unique prefix state.
+		 *
+		 * If the index key itself appear as a pathkey (could be relaxed to
+		 * being an injective function of the index key), declare go back to
+		 * unique prefix state.
+		 */
+		for ( /* continue from first unmatched query pathkey */ ;
+			 query_pk_cell != NULL;
+			 query_pk_cell = lnext(root->query_pathkeys, query_pk_cell))
+		{
+			PathKey    *qpk = lfirst_node(PathKey, query_pk_cell);
+
+			if (pathkey_is_redundant(qpk, retval))
+				continue;
+
+			if (cpathkey && qpk->pk_eclass == cpathkey->pk_eclass)
+			{
+				column_pinned = true;
+				if (!pathkey_is_redundant(cpathkey, retval))
+					retval = lappend(retval, cpathkey);
+				continue;
+			}
+
+			if (qpk->pk_var != NULL &&
+				!qpk->pk_eclass->ec_has_volatile &&
+				qpk->pk_varrelid == index->rel->relid &&
+				equal(qpk->pk_var, indexkey))
+			{
+				/*
+				 * ORDER BY x, f(x), g(x) can be satisfied by an index on x
+				 * alone for non-volatile functions when btree-equal values of
+				 * x are interchangeable (equalimage).  Within each such
+				 * group, f(x) is then constant for any deterministic f, so no
+				 * monotonicity check is needed.
+				 */
+				if (column_pinned && (pinned_is_equalimage || indexcol_is_equalimage(index, i)))
+				{
+					if (!pathkey_is_redundant(qpk, retval))
+						retval = lappend(retval, qpk);
+					pathkey_emitted = true;
+					/* skip the call ntext time */
+					pinned_is_equalimage = true;
+					continue;
+				}
+
+				if (qpk->pk_slope == MONOTONICFUNC_UNSET)
+				{
+					EquivalenceMember *em;
+
+					em = linitial(qpk->pk_eclass->ec_members);
+					qpk->pk_slope = get_expr_slope_wrt(em->em_expr,
+													   qpk->pk_var);
+				}
+
+				/*
+				 * Monotonic claims assume default btree order on both sides:
+				 * the index column's input type and the pathkey's expression
+				 * type (after domain -> base).  Refuse when either uses a
+				 * non-default opfamily - e.g. a custom domain opclass on
+				 * ORDER BY x::domain USING <op>, or int4::oid where the
+				 * eclass kept only the int4 Var but the pathkey is oid_ops.
+				 */
+				if (qpk->pk_slope != MONOTONICFUNC_NONE)
+				{
+					EquivalenceMember *em;
+					Oid			idx_default_opclass;
+					Oid			idx_default_opfamily;
+					Oid			qpk_default_opclass;
+					Oid			qpk_default_opfamily;
+
+					/* Check if the index column uses the default opfamily */
+					idx_default_opclass = GetDefaultOpClass(index->opcintype[i],
+															BTREE_AM_OID);
+					idx_default_opfamily = OidIsValid(idx_default_opclass)
+						? get_opclass_family(idx_default_opclass) : InvalidOid;
+					if (index->sortopfamily[i] != idx_default_opfamily)
+						continue;
+
+					/*
+					 * Check if the query pathkey uses the default opfamily
+					 */
+					em = linitial(qpk->pk_eclass->ec_members);
+					qpk_default_opclass =
+						GetDefaultOpClass(getBaseType(exprType((Node *) em->em_expr)),
+										  BTREE_AM_OID);
+					qpk_default_opfamily = OidIsValid(qpk_default_opclass)
+						? get_opclass_family(qpk_default_opclass) : InvalidOid;
+					if (qpk->pk_opfamily != qpk_default_opfamily)
+						continue;
+
+					/* Emit the slope pathkey */
+					spk = slope_emit_pathkey(root, qpk, indexkey,
+											 reverse_sort,
+											 nulls_first);
+					if (spk && !pathkey_is_redundant(spk, retval))
+						retval = lappend(retval, spk);
+					pathkey_emitted = true;
+				}
+				continue;
+			}
+
+			/*
+			 * Index can't satisfy query pathkeys any further
+			 */
+			query_pk_cell = NULL;
+			break;
+		}
+
+
+		if (cpathkey && !pathkey_emitted)
 		{
 			/*
 			 * We found the sort key in an EquivalenceClass, so it's relevant
@@ -840,8 +1362,10 @@ build_index_pathkeys(PlannerInfo *root,
 			 */
 			if (!pathkey_is_redundant(cpathkey, retval))
 				retval = lappend(retval, cpathkey);
+			column_pinned = true;
 		}
-		else
+
+		if (!column_pinned)
 		{
 			/*
 			 * Boolean index keys might be redundant even if they do not
@@ -856,7 +1380,6 @@ build_index_pathkeys(PlannerInfo *root,
 			if (!indexcol_is_bool_constant_for_query(root, index, i))
 				break;
 		}
-
 		i++;
 	}
 
diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c
index a605387dcaa..d37e6aefc6b 100644
--- a/src/backend/optimizer/plan/planner.c
+++ b/src/backend/optimizer/plan/planner.c
@@ -3913,6 +3913,9 @@ standard_qp_callback(PlannerInfo *root, void *extra)
 		root->query_pathkeys = root->setop_pathkeys;
 	else
 		root->query_pathkeys = NIL;
+
+	/* Annotate query pathkeys with variation sources for SLOPE */
+	precompute_slope_pathkeys(root);
 }
 
 /*
diff --git a/src/backend/utils/fmgr/slopesupport.c b/src/backend/utils/fmgr/slopesupport.c
index 02abe0a4aed..2f0c13f0561 100644
--- a/src/backend/utils/fmgr/slopesupport.c
+++ b/src/backend/utils/fmgr/slopesupport.c
@@ -1,9 +1,12 @@
 #include "postgres.h"
 
+#include <math.h>
+
 #include "c.h"
 #include "nodes/supportnodes.h"
 #include "utils/fmgrprotos.h"
 #include "utils/numeric.h"
+
 /**
  * Extended numeric sign, the usual -1, 0, 1,
  * for real numbers, and additional values for non-real
@@ -21,6 +24,15 @@ typedef enum SLOPE_SIGN
 	SLOPE_SIGN_NULL = 4,
 } SLOPE_SIGN;
 
+/* Static variables returned by prosupport below */
+static const MonotonicFunction asc_slope[2] = {MONOTONICFUNC_INCREASING, MONOTONICFUNC_INCREASING};
+static const MonotonicFunction desc_slope[2] = {MONOTONICFUNC_DECREASING, MONOTONICFUNC_DECREASING};
+static const MonotonicFunction flat_slope[2] = {MONOTONICFUNC_BOTH, MONOTONICFUNC_BOTH};
+static const MonotonicFunction diff_slope[2] = {MONOTONICFUNC_INCREASING, MONOTONICFUNC_DECREASING};
+static const MonotonicFunction asc1_slope[2] = {MONOTONICFUNC_NONE, MONOTONICFUNC_INCREASING};
+static const MonotonicFunction asc0_slope[1] = {MONOTONICFUNC_INCREASING};
+static const MonotonicFunction desc0_slope[1] = {MONOTONICFUNC_DECREASING};
+
 
 /*
  * monotonic_slope_support
@@ -45,21 +57,23 @@ monotonic_slope_support(Node *rawreq, int nslopes,
 	return PointerGetDatum(NULL);
 }
 
-
-static bool
-get_2slope_args(Node *rawreq, Node **arg0, Node **arg1)
+static List* get_arg_list(Node *rawreq)
 {
 	SupportRequestMonotonic *req = (SupportRequestMonotonic *) rawreq;
-	List	   *args;
-
 	if (!IsA(rawreq, SupportRequestMonotonic))
-		return false;
+		return NULL;
 	if (IsA(req->expr, FuncExpr))
-		args = ((FuncExpr *) req->expr)->args;
+		return ((FuncExpr *) req->expr)->args;
 	else if (IsA(req->expr, OpExpr))
-		args = ((OpExpr *) req->expr)->args;
+		return ((OpExpr *) req->expr)->args;
 	else
-		return false;
+		return NULL;
+}
+
+static bool
+get_2slope_args(Node *rawreq, Node **arg0, Node **arg1)
+{
+	List	   *args = get_arg_list(rawreq);
 	if (list_length(args) < 2)
 		return false;
 	*arg0 = (Node *) linitial(args);
@@ -147,12 +161,6 @@ get_const_sign(Const *constval)
 	}
 }
 
-static const MonotonicFunction asc_slope[2] = {MONOTONICFUNC_INCREASING, MONOTONICFUNC_INCREASING};
-static const MonotonicFunction desc_slope[2] = {MONOTONICFUNC_DECREASING, MONOTONICFUNC_DECREASING};
-static const MonotonicFunction flat_slope[2] = {MONOTONICFUNC_BOTH, MONOTONICFUNC_BOTH};
-static const MonotonicFunction diff_slope[2] = {MONOTONICFUNC_INCREASING, MONOTONICFUNC_DECREASING};
-
-
 
 /*
  * Prosupport: f(x, ...) is monotonically increasing in x.
@@ -160,10 +168,8 @@ static const MonotonicFunction diff_slope[2] = {MONOTONICFUNC_INCREASING, MONOTO
 Datum
 arg0_asc_slope_support(PG_FUNCTION_ARGS)
 {
-	static const MonotonicFunction pattern[1] = {MONOTONICFUNC_INCREASING};
-
 	return monotonic_slope_support((Node *) PG_GETARG_POINTER(0),
-								   lengthof(pattern), pattern);
+								   lengthof(asc0_slope), asc0_slope);
 }
 
  /*
@@ -172,10 +178,8 @@ arg0_asc_slope_support(PG_FUNCTION_ARGS)
 Datum
 arg0_desc_slope_support(PG_FUNCTION_ARGS)
 {
-	static const MonotonicFunction pattern[1] = {MONOTONICFUNC_DECREASING};
-
 	return monotonic_slope_support((Node *) PG_GETARG_POINTER(0),
-								   lengthof(pattern), pattern);
+								   lengthof(desc0_slope), desc0_slope);
 }
 
  /*
@@ -184,11 +188,8 @@ arg0_desc_slope_support(PG_FUNCTION_ARGS)
 Datum
 arg1_asc_slope_support(PG_FUNCTION_ARGS)
 {
-	static const MonotonicFunction pattern[2] = {MONOTONICFUNC_NONE,
-	MONOTONICFUNC_INCREASING};
-
 	return monotonic_slope_support((Node *) PG_GETARG_POINTER(0),
-								   lengthof(pattern), pattern);
+								   lengthof(asc1_slope), asc1_slope);
 }
 
 /*
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index 3c5e16ad1e7..1425c4226ac 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -1033,6 +1033,14 @@
   boot_val => 'true',
 },
 
+{ name => 'enable_slope', type => 'bool', context => 'PGC_USERSET', group => 'QUERY_TUNING_METHOD',
+  short_desc => 'Enables the planner\'s slope analysis.',
+  long_desc => 'By employing slope analysis the planner can take advantage of an index order of a monotonic expression of an index key, eliminating unnecessary sort nodes.',
+  flags => 'GUC_EXPLAIN',
+  variable => 'enable_slope',
+  boot_val => 'true',
+},
+
 { name => 'enable_sort', type => 'bool', context => 'PGC_USERSET', group => 'QUERY_TUNING_METHOD',
   short_desc => 'Enables the planner\'s use of explicit sort steps.',
   flags => 'GUC_EXPLAIN',
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index e759f06b50f..a8cc45eb0af 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -446,6 +446,7 @@
 #enable_partitionwise_aggregate = off
 #enable_presorted_aggregate = on
 #enable_seqscan = on
+#enable_slope = on
 #enable_sort = on
 #enable_tidscan = on
 #enable_group_by_reordering = on
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index d9650315016..b25f9b72363 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -1814,6 +1814,16 @@ typedef struct PathKey
 	Oid			pk_opfamily;	/* index opfamily defining the ordering */
 	CompareType pk_cmptype;		/* sort direction (ASC or DESC) */
 	bool		pk_nulls_first; /* do NULLs come before normal values? */
+
+	/*
+	 * SLOPE: innermost source of variation, filled by
+	 * precompute_slope_pathkeys().  NULL if this pathkey is a plain Var or
+	 * cannot benefit from SLOPE.  pk_slope stores a MonotonicFunction value
+	 * (from plannodes.h) as int to avoid a header dependency.
+	 */
+	Expr	   *pk_var pg_node_attr(read_write_ignore, equal_ignore);
+	Index		pk_varrelid pg_node_attr(read_write_ignore, equal_ignore);
+	int			pk_slope pg_node_attr(read_write_ignore, equal_ignore);
 } PathKey;
 
 /*
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index 2fe6b61afaf..8f069a7c8e2 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -1837,6 +1837,11 @@ typedef struct PlanInvalItem
  * than the previous call.  A monotonically decreasing function cannot yield a
  * higher value on subsequent calls, and a function which is both must return
  * the same value on each call.
+ *
+ * Used both for window function run conditions (SupportRequestWFuncMonotonic)
+ * and for per-argument monotonicity of scalar functions
+ * (SupportRequestMonotonic), where it enables the planner to use an index
+ * on 'x' to satisfy ORDER BY / GROUP BY on 'f(x)'.
  */
 typedef enum MonotonicFunction
 {
@@ -1844,6 +1849,8 @@ typedef enum MonotonicFunction
 	MONOTONICFUNC_INCREASING = (1 << 0),
 	MONOTONICFUNC_DECREASING = (1 << 1),
 	MONOTONICFUNC_BOTH = MONOTONICFUNC_INCREASING | MONOTONICFUNC_DECREASING,
+	/* Sentinel for uninitialised values */
+	MONOTONICFUNC_UNSET = (1 << 2),
 } MonotonicFunction;
 
 /*
diff --git a/src/include/optimizer/paths.h b/src/include/optimizer/paths.h
index 4ad3e8eaa89..cafcbce7d56 100644
--- a/src/include/optimizer/paths.h
+++ b/src/include/optimizer/paths.h
@@ -27,6 +27,7 @@ extern PGDLLIMPORT double min_eager_agg_group_size;
 extern PGDLLIMPORT int min_parallel_table_scan_size;
 extern PGDLLIMPORT int min_parallel_index_scan_size;
 extern PGDLLIMPORT bool enable_group_by_reordering;
+extern PGDLLIMPORT bool enable_slope;
 
 /* Hooks for plugins to get control in set_rel_pathlist() */
 typedef void (*join_path_setup_hook_type) (PlannerInfo *root,
@@ -235,6 +236,7 @@ extern Path *get_cheapest_fractional_path_for_pathkeys(List *paths,
 													   Relids required_outer,
 													   double fraction);
 extern Path *get_cheapest_parallel_safe_total_inner(List *paths);
+extern void precompute_slope_pathkeys(PlannerInfo *root);
 extern List *build_index_pathkeys(PlannerInfo *root, IndexOptInfo *index);
 extern PathKey *make_reversed_pathkey(PlannerInfo *root, PathKey *pathkey);
 extern List *reverse_pathkeys(PlannerInfo *root, List *pathkeys);
diff --git a/src/test/regress/expected/slope.out b/src/test/regress/expected/slope.out
new file mode 100644
index 00000000000..ebc4bd9b9d2
--- /dev/null
+++ b/src/test/regress/expected/slope.out
@@ -0,0 +1,862 @@
+-- Configure parameters to avoid competing plans.
+-- Prevent seqscans or bitmapscans that could be combined
+-- with a sort as an alternative to an index scan that
+-- satisfies the query order.
+SET enable_seqscan = off;
+SET enable_bitmapscan = off;
+-- Every query should rely on an index, and a scan without
+-- a sort will always be cheaper than a scan with a sort.
+SET enable_indexscan = on;
+SET enable_indexonlyscan = on;
+-- Disable hashagg that could provide an alterntative to a
+-- GROUP BY that doesn't require the input to be sorted.
+SET enable_hashagg = off;
+CREATE SCHEMA slope;
+SET search_path TO slope;
+--
+-- Merge joins
+--
+CREATE TABLE t (a int PRIMARY KEY);
+INSERT INTO t SELECT generate_series(1, 5);
+ANALYZE t;
+CREATE TABLE u (a int PRIMARY KEY);
+INSERT INTO u SELECT generate_series(1, 5);
+ANALYZE u;
+SET enable_hashjoin = off;
+SET enable_nestloop = off;
+SET enable_slope = off;
+EXPLAIN (COSTS OFF) SELECT * FROM t JOIN u USING (a);
+               QUERY PLAN                
+-----------------------------------------
+ Merge Join
+   Merge Cond: (t.a = u.a)
+   ->  Index Only Scan using t_pkey on t
+   ->  Index Only Scan using u_pkey on u
+(4 rows)
+
+SET enable_slope = on;
+EXPLAIN (COSTS OFF) SELECT * FROM t JOIN u USING (a);
+               QUERY PLAN                
+-----------------------------------------
+ Merge Join
+   Merge Cond: (t.a = u.a)
+   ->  Index Only Scan using t_pkey on t
+   ->  Index Only Scan using u_pkey on u
+(4 rows)
+
+--
+-- Default vs custom btree opfamilies.
+--
+-- SLOPE only trusts each type's default opfamily on both the index and the
+-- query pathkey.  Cross-product:
+--   (1) index default, query custom  -> not supported
+--   (2) index custom,  query default -> not supported
+--   (3) index custom,  query custom  -> not supported (even when they match)
+--   (4) index default, query default -> supported
+--
+-- <# … ># are clones of int4 comparison, registered only in test_int4_ops,
+-- so ORDER BY … USING <# forces a non-default pathkey opfamily.
+--
+CREATE OPERATOR FAMILY test_int4_ops USING btree;
+CREATE OPERATOR <# (
+  LEFTARG = int4, RIGHTARG = int4,
+  PROCEDURE = int4lt, COMMUTATOR = >#, NEGATOR = >=#
+);
+CREATE OPERATOR ># (
+  LEFTARG = int4, RIGHTARG = int4,
+  PROCEDURE = int4gt, COMMUTATOR = <#, NEGATOR = <=#
+);
+CREATE OPERATOR <=# (
+  LEFTARG = int4, RIGHTARG = int4,
+  PROCEDURE = int4le, COMMUTATOR = >=#, NEGATOR = >#
+);
+CREATE OPERATOR >=# (
+  LEFTARG = int4, RIGHTARG = int4,
+  PROCEDURE = int4ge, COMMUTATOR = <=#, NEGATOR = <#
+);
+CREATE OPERATOR CLASS test_int4_ops FOR TYPE int4
+  USING btree FAMILY test_int4_ops AS
+  OPERATOR 1 <#,
+  OPERATOR 2 <=#,
+  OPERATOR 3 =,
+  OPERATOR 4 >=#,
+  OPERATOR 5 >#,
+  FUNCTION 1 btint4cmp(int4, int4);
+CREATE TABLE t_op (x_def int4, x_cust int4);
+INSERT INTO t_op SELECT i, i FROM generate_series(1, 5) i;
+CREATE INDEX t_op_x_def ON t_op (x_def);
+CREATE INDEX t_op_x_cust ON t_op (x_cust test_int4_ops);
+ANALYZE t_op;
+-- (4) index default, query default
+EXPLAIN (COSTS OFF)
+SELECT x_def + 1 FROM t_op ORDER BY 1;
+                QUERY PLAN                
+------------------------------------------
+ Index Only Scan using t_op_x_def on t_op
+(1 row)
+
+-- (1) index default, query custom
+EXPLAIN (COSTS OFF)
+SELECT x_def + 1 FROM t_op ORDER BY 1 USING <#;
+                   QUERY PLAN                   
+------------------------------------------------
+ Sort
+   Sort Key: ((x_def + 1)) USING <#
+   ->  Index Only Scan using t_op_x_def on t_op
+(3 rows)
+
+-- (2) index custom, query default
+EXPLAIN (COSTS OFF)
+SELECT x_cust + 1 FROM t_op ORDER BY 1;
+                   QUERY PLAN                    
+-------------------------------------------------
+ Sort
+   Sort Key: ((x_cust + 1))
+   ->  Index Only Scan using t_op_x_cust on t_op
+(3 rows)
+
+-- (3) index custom, query custom
+EXPLAIN (COSTS OFF)
+SELECT x_cust + 1 FROM t_op ORDER BY 1 USING <#;
+                   QUERY PLAN                    
+-------------------------------------------------
+ Sort
+   Sort Key: ((x_cust + 1)) USING <#
+   ->  Index Only Scan using t_op_x_cust on t_op
+(3 rows)
+
+--
+-- SLOPE (Scalar function Leveraging Ordered Path Evaluation)
+-- Test that monotonic functions can use indexes for ordering
+--
+-- Create test table with various data types
+CREATE TABLE src (
+    id serial PRIMARY KEY,
+    v_int2 int2,
+    v_int4 int4,
+    v_int8 int8,
+    v_float4 float4,
+    v_float8 float8,
+    v_numeric numeric,
+    ts timestamp,
+    tstz timestamptz
+);
+-- Insert some test data
+INSERT INTO src (v_int2, v_int4, v_int8, v_float4, v_float8, v_numeric, ts, tstz)
+SELECT
+    (i % 100)::int2,
+    i,
+    i::int8,
+    i::float4,
+    i::float8,
+    i::numeric,
+    '2020-01-01'::timestamp + (i || ' hours')::interval,
+    '2020-01-01'::timestamptz + (i || ' hours')::interval
+FROM generate_series(1, 1000) i;
+-- Create indexes on the columns we'll test
+CREATE INDEX src_v_int4_idx ON src (v_int4);
+CREATE INDEX src_v_int8_idx ON src (v_int8);
+CREATE INDEX src_v_float8_idx ON src (v_float8);
+CREATE INDEX src_v_numeric_idx ON src (v_numeric);
+CREATE INDEX src_ts_idx ON src (ts);
+CREATE INDEX src_tstz_idx ON src (tstz);
+-- Analyze to get good statistics
+ANALYZE src;
+--
+-- Test that SLOPE is enabled and can be disabled
+--
+SET enable_slope = off;
+-- Basic: floor(float8) should sort even though the index order
+-- produces the result in the correct order.
+explain (costs off, verbose)
+select floor(v_float8), count(*) from src group by 1;
+                           QUERY PLAN                            
+-----------------------------------------------------------------
+ GroupAggregate
+   Output: (floor(v_float8)), count(*)
+   Group Key: (floor(src.v_float8))
+   ->  Sort
+         Output: (floor(v_float8))
+         Sort Key: (floor(src.v_float8))
+         ->  Index Only Scan using src_v_float8_idx on slope.src
+               Output: floor(v_float8)
+(8 rows)
+
+RESET enable_slope;
+SHOW enable_slope;
+ enable_slope 
+--------------
+ on
+(1 row)
+
+--
+-- Test GROUP BY with monotonic function
+--
+-- Basic: floor(float8) should use index on v_float8
+explain (costs off, verbose)
+select floor(v_float8), count(*) from src group by 1;
+                        QUERY PLAN                         
+-----------------------------------------------------------
+ GroupAggregate
+   Output: (floor(v_float8)), count(*)
+   Group Key: floor(src.v_float8)
+   ->  Index Only Scan using src_v_float8_idx on slope.src
+         Output: floor(v_float8)
+(5 rows)
+
+-- ceil(float8) should use index on v_float8
+explain (costs off, verbose)
+select ceil(v_float8), count(*) from src group by 1;
+                        QUERY PLAN                         
+-----------------------------------------------------------
+ GroupAggregate
+   Output: (ceil(v_float8)), count(*)
+   Group Key: ceil(src.v_float8)
+   ->  Index Only Scan using src_v_float8_idx on slope.src
+         Output: ceil(v_float8)
+(5 rows)
+
+-- floor(numeric) should use index on v_numeric
+explain (costs off, verbose)
+select floor(v_numeric), count(*) from src group by 1;
+                         QUERY PLAN                         
+------------------------------------------------------------
+ GroupAggregate
+   Output: (floor(v_numeric)), count(*)
+   Group Key: floor(src.v_numeric)
+   ->  Index Only Scan using src_v_numeric_idx on slope.src
+         Output: floor(v_numeric)
+(5 rows)
+
+-- timestamp::date cast should use index on ts
+explain (costs off, verbose)
+select ts::date, count(*) from src group by 1;
+                     QUERY PLAN                      
+-----------------------------------------------------
+ GroupAggregate
+   Output: ((ts)::date), count(*)
+   Group Key: (src.ts)::date
+   ->  Index Only Scan using src_ts_idx on slope.src
+         Output: (ts)::date
+(5 rows)
+
+-- date_trunc on timestamp should use index
+explain (costs off, verbose)
+select date_trunc('day', ts), count(*) from src group by 1;
+                     QUERY PLAN                      
+-----------------------------------------------------
+ GroupAggregate
+   Output: (date_trunc('day'::text, ts)), count(*)
+   Group Key: date_trunc('day'::text, src.ts)
+   ->  Index Only Scan using src_ts_idx on slope.src
+         Output: date_trunc('day'::text, ts)
+(5 rows)
+
+-- date_trunc on timestamptz should not use index
+explain (costs off, verbose)
+select date_trunc('day', tstz), count(*) from src group by 1;
+                         QUERY PLAN                          
+-------------------------------------------------------------
+ GroupAggregate
+   Output: (date_trunc('day'::text, tstz)), count(*)
+   Group Key: (date_trunc('day'::text, src.tstz))
+   ->  Sort
+         Output: (date_trunc('day'::text, tstz))
+         Sort Key: (date_trunc('day'::text, src.tstz))
+         ->  Index Only Scan using src_tstz_idx on slope.src
+               Output: date_trunc('day'::text, tstz)
+(8 rows)
+
+--
+-- Test arithmetic operations
+--
+-- Addition: v_int4 + 10 is increasing in v_int4
+explain (costs off, verbose)
+select v_int4 + 10, count(*) from src group by 1;
+                       QUERY PLAN                        
+---------------------------------------------------------
+ GroupAggregate
+   Output: ((v_int4 + 10)), count(*)
+   Group Key: (src.v_int4 + 10)
+   ->  Index Only Scan using src_v_int4_idx on slope.src
+         Output: (v_int4 + 10)
+(5 rows)
+
+-- Subtraction: v_int4 - 10 is increasing in v_int4
+explain (costs off, verbose)
+select v_int4 - 10, count(*) from src group by 1;
+                       QUERY PLAN                        
+---------------------------------------------------------
+ GroupAggregate
+   Output: ((v_int4 - 10)), count(*)
+   Group Key: (src.v_int4 - 10)
+   ->  Index Only Scan using src_v_int4_idx on slope.src
+         Output: (v_int4 - 10)
+(5 rows)
+
+-- Multiplication by positive constant: v_int4 * 2 is increasing
+explain (costs off, verbose)
+select v_int4 * 2, count(*) from src group by 1;
+                       QUERY PLAN                        
+---------------------------------------------------------
+ GroupAggregate
+   Output: ((v_int4 * 2)), count(*)
+   Group Key: (src.v_int4 * 2)
+   ->  Index Only Scan using src_v_int4_idx on slope.src
+         Output: (v_int4 * 2)
+(5 rows)
+
+-- Division by positive constant: v_int4 / 2 is increasing
+explain (costs off, verbose)
+select v_int4 / 2, count(*) from src group by 1;
+                       QUERY PLAN                        
+---------------------------------------------------------
+ GroupAggregate
+   Output: ((v_int4 / 2)), count(*)
+   Group Key: (src.v_int4 / 2)
+   ->  Index Only Scan using src_v_int4_idx on slope.src
+         Output: (v_int4 / 2)
+(5 rows)
+
+--
+-- Test decreasing functions
+-- These queries can't use the index order because group pathkeys
+-- are always ASC NULLS LAST, while -int_v4 is DESC NULLS LAST.
+-- Unary minus: -v_int4 is decreasing in v_int4
+explain (costs off, verbose)
+select -v_int4, count(*) from src group by 1;
+                          QUERY PLAN                           
+---------------------------------------------------------------
+ GroupAggregate
+   Output: ((- v_int4)), count(*)
+   Group Key: ((- src.v_int4))
+   ->  Sort
+         Output: ((- v_int4))
+         Sort Key: ((- src.v_int4))
+         ->  Index Only Scan using src_v_int4_idx on slope.src
+               Output: (- v_int4)
+(8 rows)
+
+-- Subtraction from constant: 1000 - v_int4 is decreasing in v_int4
+explain (costs off, verbose)
+select 1000 - v_int4, count(*) from src group by 1;
+                          QUERY PLAN                           
+---------------------------------------------------------------
+ GroupAggregate
+   Output: ((1000 - v_int4)), count(*)
+   Group Key: ((1000 - src.v_int4))
+   ->  Sort
+         Output: ((1000 - v_int4))
+         Sort Key: ((1000 - src.v_int4))
+         ->  Index Only Scan using src_v_int4_idx on slope.src
+               Output: (1000 - v_int4)
+(8 rows)
+
+-- Multiplication by negative constant: v_int4 * (-2) is decreasing
+explain (costs off, verbose)
+select v_int4 * (-2), count(*) from src group by 1;
+                          QUERY PLAN                           
+---------------------------------------------------------------
+ GroupAggregate
+   Output: ((v_int4 * '-2'::integer)), count(*)
+   Group Key: ((src.v_int4 * '-2'::integer))
+   ->  Sort
+         Output: ((v_int4 * '-2'::integer))
+         Sort Key: ((src.v_int4 * '-2'::integer))
+         ->  Index Only Scan using src_v_int4_idx on slope.src
+               Output: (v_int4 * '-2'::integer)
+(8 rows)
+
+-- Division by negative constant: v_int4 / (-2) is decreasing
+explain (costs off, verbose)
+select v_int4 / (-2), count(*) from src group by 1;
+                          QUERY PLAN                           
+---------------------------------------------------------------
+ GroupAggregate
+   Output: ((v_int4 / '-2'::integer)), count(*)
+   Group Key: ((src.v_int4 / '-2'::integer))
+   ->  Sort
+         Output: ((v_int4 / '-2'::integer))
+         Sort Key: ((src.v_int4 / '-2'::integer))
+         ->  Index Only Scan using src_v_int4_idx on slope.src
+               Output: (v_int4 / '-2'::integer)
+(8 rows)
+
+--
+-- Test ORDER BY with monotonic function
+--
+-- ORDER BY floor(v_float8) should use index
+explain (costs off, verbose)
+select floor(v_float8), v_float8 from src order by 1;
+                     QUERY PLAN                      
+-----------------------------------------------------
+ Index Only Scan using src_v_float8_idx on slope.src
+   Output: floor(v_float8), v_float8
+(2 rows)
+
+-- ORDER BY -v_int4 DESC requires sorting
+-- the index order is v_int4 ASC and, implicitly, NULLS LAST
+-- a forward scan gives -v_int4 DESC NULLS LAST
+-- a backward scan gives -v_int4 ASC NULLS FIRST
+-- the query order is DESC and, implicitly, NULLS FIRST
+explain (costs off, verbose)
+select -v_int4 from src order by 1 desc;
+                       QUERY PLAN                        
+---------------------------------------------------------
+ Sort
+   Output: ((- v_int4))
+   Sort Key: ((- src.v_int4)) DESC
+   ->  Index Only Scan using src_v_int4_idx on slope.src
+         Output: (- v_int4)
+(5 rows)
+
+-- ORDER BY -v_int4 ASC requires sorting
+-- a forward scan gives -v_int4 DESC NULLS LAST
+-- a backward scan gives -v_int4 ASC NULLS FIRST
+-- the query order is -v_int4 ASC NULLS LAST
+explain (costs off, verbose)
+select -v_int4 from src order by 1;
+                       QUERY PLAN                        
+---------------------------------------------------------
+ Sort
+   Output: ((- v_int4))
+   Sort Key: ((- src.v_int4))
+   ->  Index Only Scan using src_v_int4_idx on slope.src
+         Output: (- v_int4)
+(5 rows)
+
+--
+-- Group and order
+--
+explain (costs off, verbose)
+select tstz::date, count(*) from src group by 1 order by 1;
+                         QUERY PLAN                          
+-------------------------------------------------------------
+ GroupAggregate
+   Output: ((tstz)::date), count(*)
+   Group Key: ((src.tstz)::date)
+   ->  Sort
+         Output: ((tstz)::date)
+         Sort Key: ((src.tstz)::date)
+         ->  Index Only Scan using src_tstz_idx on slope.src
+               Output: (tstz)::date
+(8 rows)
+
+--
+-- Test nested monotonic function
+--
+-- floor(floor(x)) should still use index
+explain (costs off, verbose)
+select floor(floor(v_float8)), count(*) from src group by 1;
+                        QUERY PLAN                         
+-----------------------------------------------------------
+ GroupAggregate
+   Output: (floor(floor(v_float8))), count(*)
+   Group Key: floor(floor(src.v_float8))
+   ->  Index Only Scan using src_v_float8_idx on slope.src
+         Output: floor(floor(v_float8))
+(5 rows)
+
+-- floor(v + 1) should use index
+explain (costs off, verbose)
+select floor(v_float8 + 1), count(*) from src group by 1;
+                           QUERY PLAN                            
+-----------------------------------------------------------------
+ GroupAggregate
+   Output: (floor((v_float8 + '1'::double precision))), count(*)
+   Group Key: floor((src.v_float8 + '1'::double precision))
+   ->  Index Only Scan using src_v_float8_idx on slope.src
+         Output: floor((v_float8 + '1'::double precision))
+(5 rows)
+
+--
+-- Test all index/query direction+nulls combinations for SLOPE.
+-- For an increasing function like floor(), the scan uses the index when both
+-- direction and nulls agree (Forward) or both are flipped (Backward).
+-- If one agrees and the other disagrees, then the scan cannot use the
+-- index and a sort is required.
+--
+CREATE TABLE nulls_tmp (v float8);
+INSERT INTO nulls_tmp VALUES (1), (NULL), (2);
+ANALYZE nulls_tmp;
+CREATE TEMPORARY TABLE nulls_results (
+    seq serial,
+    sign text, index_order text, query_order text, scan_method text,
+    example text
+);
+DO $$
+DECLARE
+    r record;
+    plan_json json;
+    node_type text;
+    query text;
+    r1 text;
+    r2 text;
+BEGIN
+    FOR r IN
+        SELECT idx_dir, idx_nf, qry_dir, qry_nf, sign
+        FROM unnest(ARRAY['ASC','DESC'])   WITH ORDINALITY AS idx_dir(idx_dir, idx_dir_i),
+             unnest(ARRAY['FIRST','LAST']) WITH ORDINALITY AS idx_nf (idx_nf,  idx_nf_i),
+             unnest(ARRAY['+','-'])        WITH ORDINALITY AS sign   (sign,    sign_i),
+             unnest(ARRAY['FIRST','LAST']) WITH ORDINALITY AS qry_nf (qry_nf,  qry_nf_i),
+             unnest(ARRAY['ASC','DESC'])   WITH ORDINALITY AS qry_dir(qry_dir, qry_dir_i)
+        ORDER BY qry_dir_i, qry_nf_i, sign_i, idx_dir_i, idx_nf_i
+    LOOP
+        EXECUTE format('CREATE INDEX nulls_tmp_idx ON nulls_tmp (v %s NULLS %s)',
+                        r.idx_dir, r.idx_nf);
+        query := format('SELECT floor(0.5 %s v) as x FROM nulls_tmp ORDER BY 1 %s NULLS %s',
+            r.sign, r.qry_dir, r.qry_nf);
+        EXECUTE format('EXPLAIN (FORMAT JSON) %s', query)
+        INTO plan_json;
+
+        SET enable_slope = off;
+        EXECUTE 'SELECT string_agg(coalesce(x::text, ''NULL''), '','') FROM (' || query || ') tmp(x)' into r1;
+        SET enable_slope = on;
+        EXECUTE 'SELECT string_agg(coalesce(x::text, ''NULL''), '','') FROM (' || query || ') tmp(x)' into r2;
+        if r1 <> r2 then
+            raise notice 'query %', query;
+            raise exception ' r1=%, r2=%', r1, r2;
+        end if;
+        node_type := plan_json->0->'Plan'->>'Node Type';
+        INSERT INTO nulls_results (sign, index_order, query_order, scan_method, example) VALUES (
+            r.sign,
+            r.idx_dir || ' NULLS ' || r.idx_nf,
+            r.qry_dir || ' NULLS ' || r.qry_nf,
+            CASE WHEN node_type IN ('Index Only Scan', 'Index Scan')
+                 THEN plan_json->0->'Plan'->>'Scan Direction'
+                 ELSE node_type
+            END,
+            r2
+        );
+        EXECUTE format('DROP INDEX nulls_tmp_idx');
+    END LOOP;
+END;
+$$;
+SELECT sign, index_order, query_order, scan_method, example
+FROM nulls_results ORDER BY seq;
+ sign |   index_order    |   query_order    | scan_method |  example   
+------+------------------+------------------+-------------+------------
+ +    | ASC NULLS FIRST  | ASC NULLS FIRST  | Forward     | NULL,1,2
+ +    | ASC NULLS LAST   | ASC NULLS FIRST  | Sort        | NULL,1,2
+ +    | DESC NULLS FIRST | ASC NULLS FIRST  | Sort        | NULL,1,2
+ +    | DESC NULLS LAST  | ASC NULLS FIRST  | Backward    | NULL,1,2
+ -    | ASC NULLS FIRST  | ASC NULLS FIRST  | Sort        | NULL,-2,-1
+ -    | ASC NULLS LAST   | ASC NULLS FIRST  | Sort        | NULL,-2,-1
+ -    | DESC NULLS FIRST | ASC NULLS FIRST  | Sort        | NULL,-2,-1
+ -    | DESC NULLS LAST  | ASC NULLS FIRST  | Sort        | NULL,-2,-1
+ +    | ASC NULLS FIRST  | ASC NULLS LAST   | Sort        | 1,2,NULL
+ +    | ASC NULLS LAST   | ASC NULLS LAST   | Forward     | 1,2,NULL
+ +    | DESC NULLS FIRST | ASC NULLS LAST   | Backward    | 1,2,NULL
+ +    | DESC NULLS LAST  | ASC NULLS LAST   | Sort        | 1,2,NULL
+ -    | ASC NULLS FIRST  | ASC NULLS LAST   | Sort        | -2,-1,NULL
+ -    | ASC NULLS LAST   | ASC NULLS LAST   | Sort        | -2,-1,NULL
+ -    | DESC NULLS FIRST | ASC NULLS LAST   | Sort        | -2,-1,NULL
+ -    | DESC NULLS LAST  | ASC NULLS LAST   | Sort        | -2,-1,NULL
+ +    | ASC NULLS FIRST  | DESC NULLS FIRST | Sort        | NULL,2,1
+ +    | ASC NULLS LAST   | DESC NULLS FIRST | Backward    | NULL,2,1
+ +    | DESC NULLS FIRST | DESC NULLS FIRST | Forward     | NULL,2,1
+ +    | DESC NULLS LAST  | DESC NULLS FIRST | Sort        | NULL,2,1
+ -    | ASC NULLS FIRST  | DESC NULLS FIRST | Sort        | NULL,-1,-2
+ -    | ASC NULLS LAST   | DESC NULLS FIRST | Sort        | NULL,-1,-2
+ -    | DESC NULLS FIRST | DESC NULLS FIRST | Sort        | NULL,-1,-2
+ -    | DESC NULLS LAST  | DESC NULLS FIRST | Sort        | NULL,-1,-2
+ +    | ASC NULLS FIRST  | DESC NULLS LAST  | Backward    | 2,1,NULL
+ +    | ASC NULLS LAST   | DESC NULLS LAST  | Sort        | 2,1,NULL
+ +    | DESC NULLS FIRST | DESC NULLS LAST  | Sort        | 2,1,NULL
+ +    | DESC NULLS LAST  | DESC NULLS LAST  | Forward     | 2,1,NULL
+ -    | ASC NULLS FIRST  | DESC NULLS LAST  | Sort        | -1,-2,NULL
+ -    | ASC NULLS LAST   | DESC NULLS LAST  | Sort        | -1,-2,NULL
+ -    | DESC NULLS FIRST | DESC NULLS LAST  | Sort        | -1,-2,NULL
+ -    | DESC NULLS LAST  | DESC NULLS LAST  | Sort        | -1,-2,NULL
+(32 rows)
+
+--
+-- Test special numeric values
+--
+-- This check the correctness of the query for a nullable column
+-- also containing special values (-inf, +inf and NaN)
+-- represented as float8, float4, numeric and user defined
+-- domains with float8 base type.
+CREATE DOMAIN non42 AS float8 CHECK (value != 42);
+CREATE DOMAIN fp_real AS float8;
+CREATE UNLOGGED TABLE numeric_corners (i serial, x float8);
+INSERT INTO numeric_corners (x)
+SELECT x::float8 as x FROM
+        unnest(ARRAY['-inf', '-1', '0', '3', 'inf', 'nan', NULL]) x(x)
+ORDER BY 1;
+CREATE INDEX numeric_corners_xfp8_idx_nulllast ON numeric_corners ((x::float8) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xfp8_idx_nullfirst ON numeric_corners ((x::float8) ASC NULLS FIRST);
+CREATE INDEX numeric_corners_xfp4_idx_nulllast ON numeric_corners ((x::float4) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xfp4_idx_nullfirst ON numeric_corners ((x::float4) ASC NULLS FIRST);
+CREATE INDEX numeric_corners_xnum_idx_nulllast ON numeric_corners ((x::numeric) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xnum_idx_nullfirst ON numeric_corners ((x::numeric) ASC NULLS FIRST);
+CREATE INDEX numeric_corners_xnon42_idx_nulllast ON numeric_corners ((x::non42) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xnon42_idx_nullfirst ON numeric_corners ((x::non42) ASC NULLS FIRST);
+CREATE INDEX numeric_corners_xfpreal_idx_nulllast ON numeric_corners ((x::fp_real) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xfpreal_idx_nullfirst ON numeric_corners ((x::fp_real) ASC NULLS FIRST);
+CREATE TEMPORARY TABLE numeric_corners_results (
+    seq serial,
+    expr text COLLATE "C",
+    sort_order text COLLATE "C",
+    nulls_order text COLLATE "C",
+    typename text COLLATE "C",
+    result float8[],
+    expected float8[],
+    expected_seq int4[],
+    nan_values float8[],
+    plan1_json json,
+    plan2_json json
+);
+DO $$
+DECLARE
+    r record;
+    result float8[];
+    expected float8[];
+    expected_seq int4[];
+    nan_values float8[];
+    query text;
+    agg_query text;
+    plan_query text;
+    nan_query text;
+    expected_seq_query text;
+    plan1_json json;
+    plan2_json json;
+BEGIN
+
+    FOR r IN
+        SELECT replace(replace(s, 'x', 'x::' || t), 'A', a || '::' || t) as expr, sort_order, nulls_order, t as typename
+        FROM
+        unnest(ARRAY['ASC', 'DESC']) WITH ORDINALITY AS so(sort_order, so_i),
+        unnest(ARRAY['FIRST', 'LAST']) WITH ORDINALITY AS nf(nulls_order, no_i),
+        unnest(ARRAY['float8', 'float4', 'numeric', 'non42', 'fp_real']) WITH ORDINALITY AS t(t, t_i),
+        unnest(ARRAY[
+        'x+A', 'x-A', 'x*A', 'x/A',
+        'A+x', 'A-x', 'A*x', '-x'
+        ]) WITH ORDINALITY AS s(s, s_i),
+        unnest(ARRAY['''inf''', '''-inf''', '1']) WITH ORDINALITY AS a(a, a_i)
+        ORDER BY s_i, so_i, no_i, t_i
+    LOOP
+        query := 'SELECT *, (' || r.expr || ') as f '
+                 || 'FROM numeric_corners '
+                 || 'ORDER BY f ' || r.sort_order || ' NULLS ' || r.nulls_order;
+        nan_query := 'SELECT array_agg(x) FROM numeric_corners
+        WHERE ' || r.expr || ' = ''nan''::' || r.typename || ' AND x != ''nan''::' || r.typename;
+        agg_query := 'SELECT array_agg(f::float8) FROM (' || query || ') tmp';
+        expected_seq_query := 'SELECT array_agg(i::int4) FROM (' || query || ') tmp';
+        plan_query := 'EXPLAIN (FORMAT JSON) ' || query;
+        -- slope optimization disabled
+        SET enable_slope = off;
+        EXECUTE agg_query into result;
+        EXECUTE plan_query into plan2_json;
+        -- slope optimization enabled
+        SET enable_slope = on;
+        EXECUTE agg_query into expected;
+        EXECUTE expected_seq_query into expected_seq;
+        EXECUTE nan_query into nan_values;
+        EXECUTE plan_query into plan1_json;
+        -- store results
+        INSERT INTO numeric_corners_results
+        (expr, sort_order, nulls_order, typename, result, expected, expected_seq, nan_values, plan1_json, plan2_json)
+        VALUES (r.expr, r.sort_order, r.nulls_order, r.typename, result, expected, expected_seq, nan_values, plan1_json, plan2_json);
+    END LOOP;
+END;
+$$;
+-- display failing test cases
+SELECT seq, expr
+, sort_order, nulls_order
+, expected, expected_seq, result, nan_values
+, plan2_json->0->'Plan'->>'Node Type' as plan2
+FROM numeric_corners_results
+WHERE expected != result;
+ seq | expr | sort_order | nulls_order | expected | expected_seq | result | nan_values | plan2 
+-----+------+------------+-------------+----------+--------------+--------+------------+-------
+(0 rows)
+
+-- check the number of test cases
+SELECT expected = result as passed, count(1)
+FROM numeric_corners_results
+GROUP BY 1;
+ passed | count 
+--------+-------
+ t      |   480
+(1 row)
+
+--
+-- Test pathkey chains: f(x), x (descending chain)
+-- The order of f(x) is implied by the order of x when f is monotonic.
+--
+explain (costs off, verbose)
+select * from src order by ts::date, ts;
+                                        QUERY PLAN                                         
+-------------------------------------------------------------------------------------------
+ Index Scan using src_ts_idx on slope.src
+   Output: id, v_int2, v_int4, v_int8, v_float4, v_float8, v_numeric, ts, tstz, (ts)::date
+(2 rows)
+
+--
+-- Test pathkey chains: x, f(x) (ascending chain)
+-- f(x) is constant when x is constant, so f(x) is a redundant tiebreaker.
+--
+-- When btree equality on x implies image equality (equalimage), any
+-- deterministic f(x) is constant on each index group, so f(x) is a
+-- redundant tiebreaker and the index on x alone can satisfy ORDER BY x, f(x).
+-- When equalimage does not hold, values that compare equal under the index
+-- can still differ in f(x); the planner must keep sorting on f(x),
+-- unless f(x) is monotonic
+--
+-- int4 has btequalimage; numeric/float8 do not (e.g. 1.0 = 1.00, -0.0 = 0.0).
+explain (costs off, verbose)
+select * from src order by ts, ts::date;
+                                        QUERY PLAN                                         
+-------------------------------------------------------------------------------------------
+ Index Scan using src_ts_idx on slope.src
+   Output: id, v_int2, v_int4, v_int8, v_float4, v_float8, v_numeric, ts, tstz, (ts)::date
+(2 rows)
+
+-- non-monotonic f(x) after pinned x with equalimage
+-- can safely ignore f(x) tie breaker and use index order alone
+explain (costs off, verbose)
+select v_int4, abs(v_int4)
+from src order by v_int4, abs(v_int4);
+                    QUERY PLAN                     
+---------------------------------------------------
+ Index Only Scan using src_v_int4_idx on slope.src
+   Output: v_int4, abs(v_int4)
+(2 rows)
+
+-- non-monotonic f(x) before x: cannot use index order alone
+explain (costs off, verbose)
+select mod(v_int4, 7), v_int4
+from src order by mod(v_int4, 7), v_int4;
+                       QUERY PLAN                        
+---------------------------------------------------------
+ Sort
+   Output: (mod(v_int4, 7)), v_int4
+   Sort Key: (mod(src.v_int4, 7)), src.v_int4
+   ->  Index Only Scan using src_v_int4_idx on slope.src
+         Output: mod(v_int4, 7), v_int4
+(5 rows)
+
+-- float_ops has no equalimage so it requires incremental sort
+explain (costs off, verbose)
+select * from src order by v_float8, cos(v_float8);
+                                             QUERY PLAN                                             
+----------------------------------------------------------------------------------------------------
+ Incremental Sort
+   Output: id, v_int2, v_int4, v_int8, v_float4, v_float8, v_numeric, ts, tstz, (cos(v_float8))
+   Sort Key: src.v_float8, (cos(src.v_float8))
+   Presorted Key: src.v_float8
+   ->  Index Scan using src_v_float8_idx on slope.src
+         Output: id, v_int2, v_int4, v_int8, v_float4, v_float8, v_numeric, ts, tstz, cos(v_float8)
+(6 rows)
+
+-- monotonic f without equalimage: index order on n implies order on 2 * n
+explain (costs off, verbose)
+select * from src order by v_float8, 2 * v_float8;
+                                                    QUERY PLAN                                                     
+-------------------------------------------------------------------------------------------------------------------
+ Index Scan using src_v_float8_idx on slope.src
+   Output: id, v_int2, v_int4, v_int8, v_float4, v_float8, v_numeric, ts, tstz, ('2'::double precision * v_float8)
+(2 rows)
+
+--
+-- INCLUDE columns are not index sort keys; build_index_pathkeys must stop
+-- at nkeycolumns and must not treat INCLUDE attrs as an ordered prefix.
+-- Use a separate table so src's index set stays unchanged.
+--
+CREATE TABLE inc (
+    k int4,
+    ts timestamp
+);
+INSERT INTO inc
+SELECT i, '2020-01-01'::timestamp + (i || ' hours')::interval
+FROM generate_series(1, 100) i;
+CREATE INDEX inc_k_inc_ts_idx ON inc (k) INCLUDE (ts);
+CREATE INDEX inc_ts_idx ON inc (ts);
+ANALYZE inc;
+explain (costs off, verbose)
+select k, ts from inc order by k;
+                     QUERY PLAN                      
+-----------------------------------------------------
+ Index Only Scan using inc_k_inc_ts_idx on slope.inc
+   Output: k, ts
+(2 rows)
+
+explain (costs off, verbose)
+select k, ts::date from inc order by k, ts::date;
+                        QUERY PLAN                         
+-----------------------------------------------------------
+ Incremental Sort
+   Output: k, ((ts)::date)
+   Sort Key: inc.k, ((inc.ts)::date)
+   Presorted Key: inc.k
+   ->  Index Only Scan using inc_k_inc_ts_idx on slope.inc
+         Output: k, (ts)::date
+(6 rows)
+
+explain (costs off, verbose)
+select ts::date from inc order by ts::date;
+                  QUERY PLAN                   
+-----------------------------------------------
+ Index Only Scan using inc_ts_idx on slope.inc
+   Output: (ts)::date
+(2 rows)
+
+--
+-- Test DISTINCT with monotonic functions
+--
+explain (costs off, verbose)
+select distinct ts::date from src;
+                     QUERY PLAN                      
+-----------------------------------------------------
+ Unique
+   Output: ((ts)::date)
+   ->  Index Only Scan using src_ts_idx on slope.src
+         Output: (ts)::date
+(4 rows)
+
+--
+-- Test window functions with PARTITION BY monotonic function
+--
+explain (costs off, verbose)
+select ts::date, ts, row_number() over (partition by ts::date order by ts)
+from src;
+                                       QUERY PLAN                                       
+----------------------------------------------------------------------------------------
+ WindowAgg
+   Output: ((ts)::date), ts, row_number() OVER w1
+   Window: w1 AS (PARTITION BY (src.ts)::date ORDER BY src.ts ROWS UNBOUNDED PRECEDING)
+   ->  Index Only Scan using src_ts_idx on slope.src
+         Output: (ts)::date, ts
+(5 rows)
+
+--
+-- RelabelType across binary-compatible types with different default
+-- orders: int4 (signed) vs oid (unsigned).  Index order on int4 is not
+-- oid order, so SLOPE must not omit the Sort.
+--
+EXPLAIN (COSTS OFF)
+SELECT v_int4::oid FROM src ORDER BY 1;
+                    QUERY PLAN                     
+---------------------------------------------------
+ Sort
+   Sort Key: ((v_int4)::oid)
+   ->  Index Only Scan using src_v_int4_idx on src
+(3 rows)
+
+DROP SCHEMA slope CASCADE;
+NOTICE:  drop cascades to 14 other objects
+DETAIL:  drop cascades to table t
+drop cascades to table u
+drop cascades to operator family test_int4_ops for access method btree
+drop cascades to operator >#(integer,integer)
+drop cascades to operator >=#(integer,integer)
+drop cascades to operator <#(integer,integer)
+drop cascades to operator <=#(integer,integer)
+drop cascades to table t_op
+drop cascades to table src
+drop cascades to table nulls_tmp
+drop cascades to type non42
+drop cascades to type fp_real
+drop cascades to table numeric_corners
+drop cascades to table inc
diff --git a/src/test/regress/expected/sysviews.out b/src/test/regress/expected/sysviews.out
index 1e327c2afa4..60ca804cb01 100644
--- a/src/test/regress/expected/sysviews.out
+++ b/src/test/regress/expected/sysviews.out
@@ -179,9 +179,10 @@ select name, setting from pg_settings where name like 'enable%';
  enable_presorted_aggregate     | on
  enable_self_join_elimination   | on
  enable_seqscan                 | on
+ enable_slope                   | on
  enable_sort                    | on
  enable_tidscan                 | on
-(26 rows)
+(27 rows)
 
 -- There are always wait event descriptions for various types.  InjectionPoint
 -- may be present or absent, depending on history since last postmaster start.
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 59ecb33e875..390b4f2b2a0 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -47,8 +47,9 @@ test: create_index create_index_spgist create_view index_including index_includi
 
 # ----------
 # Another group of parallel tests
+# (20 tests)
 # ----------
-test: create_aggregate create_function_sql create_cast constraints triggers select inherit typed_table vacuum drop_if_exists updatable_views roleattributes create_am hash_func errors infinite_recurse create_property_graph for_portion_of slope_catalog
+test: create_aggregate create_function_sql create_cast constraints triggers select inherit typed_table vacuum drop_if_exists updatable_views roleattributes create_am hash_func errors infinite_recurse create_property_graph for_portion_of slope_catalog slope
 
 # ----------
 # sanity_check does a vacuum, affecting the sort order of SELECT *
@@ -60,11 +61,13 @@ test: sanity_check
 # Another group of parallel tests
 # aggregates depends on create_aggregate
 # join depends on create_misc
+# (20 tests)
 # ----------
 test: select_into select_distinct select_distinct_on select_implicit select_having subselect union case join aggregates transactions random portals arrays btree_index hash_index update delete namespace prepared_xacts
 
 # ----------
 # Another group of parallel tests
+# (20 tests)
 # ----------
 test: brin gin gist spgist privileges init_privs security_label collate matview lock replica_identity rowsecurity object_address tablesample groupingsets drop_operator password identity generated_stored join_hash
 
@@ -75,6 +78,7 @@ test: brin_bloom brin_multi
 
 # ----------
 # Another group of parallel tests
+# (20 tests)
 # ----------
 test: create_table_like alter_generic alter_operator misc async dbsize merge misc_functions nls sysviews tsrf tid tidscan tidrangescan collate.utf8 collate.icu.utf8 incremental_sort create_role without_overlaps generated_virtual
 
diff --git a/src/test/regress/sql/slope.sql b/src/test/regress/sql/slope.sql
new file mode 100644
index 00000000000..1defff4d955
--- /dev/null
+++ b/src/test/regress/sql/slope.sql
@@ -0,0 +1,539 @@
+-- Configure parameters to avoid competing plans.
+
+-- Prevent seqscans or bitmapscans that could be combined
+-- with a sort as an alternative to an index scan that
+-- satisfies the query order.
+SET enable_seqscan = off;
+SET enable_bitmapscan = off;
+-- Every query should rely on an index, and a scan without
+-- a sort will always be cheaper than a scan with a sort.
+SET enable_indexscan = on;
+SET enable_indexonlyscan = on;
+-- Disable hashagg that could provide an alterntative to a
+-- GROUP BY that doesn't require the input to be sorted.
+SET enable_hashagg = off;
+
+CREATE SCHEMA slope;
+SET search_path TO slope;
+
+--
+-- Merge joins
+--
+CREATE TABLE t (a int PRIMARY KEY);
+INSERT INTO t SELECT generate_series(1, 5);
+ANALYZE t;
+CREATE TABLE u (a int PRIMARY KEY);
+INSERT INTO u SELECT generate_series(1, 5);
+ANALYZE u;
+
+SET enable_hashjoin = off;
+SET enable_nestloop = off;
+
+SET enable_slope = off;
+EXPLAIN (COSTS OFF) SELECT * FROM t JOIN u USING (a);
+SET enable_slope = on;
+EXPLAIN (COSTS OFF) SELECT * FROM t JOIN u USING (a);
+
+
+
+--
+-- Default vs custom btree opfamilies.
+--
+
+-- SLOPE only trusts each type's default opfamily on both the index and the
+-- query pathkey.  Cross-product:
+--   (1) index default, query custom  -> not supported
+--   (2) index custom,  query default -> not supported
+--   (3) index custom,  query custom  -> not supported (even when they match)
+--   (4) index default, query default -> supported
+--
+-- <# … ># are clones of int4 comparison, registered only in test_int4_ops,
+-- so ORDER BY … USING <# forces a non-default pathkey opfamily.
+--
+CREATE OPERATOR FAMILY test_int4_ops USING btree;
+
+CREATE OPERATOR <# (
+  LEFTARG = int4, RIGHTARG = int4,
+  PROCEDURE = int4lt, COMMUTATOR = >#, NEGATOR = >=#
+);
+CREATE OPERATOR ># (
+  LEFTARG = int4, RIGHTARG = int4,
+  PROCEDURE = int4gt, COMMUTATOR = <#, NEGATOR = <=#
+);
+CREATE OPERATOR <=# (
+  LEFTARG = int4, RIGHTARG = int4,
+  PROCEDURE = int4le, COMMUTATOR = >=#, NEGATOR = >#
+);
+CREATE OPERATOR >=# (
+  LEFTARG = int4, RIGHTARG = int4,
+  PROCEDURE = int4ge, COMMUTATOR = <=#, NEGATOR = <#
+);
+
+CREATE OPERATOR CLASS test_int4_ops FOR TYPE int4
+  USING btree FAMILY test_int4_ops AS
+  OPERATOR 1 <#,
+  OPERATOR 2 <=#,
+  OPERATOR 3 =,
+  OPERATOR 4 >=#,
+  OPERATOR 5 >#,
+  FUNCTION 1 btint4cmp(int4, int4);
+
+CREATE TABLE t_op (x_def int4, x_cust int4);
+INSERT INTO t_op SELECT i, i FROM generate_series(1, 5) i;
+CREATE INDEX t_op_x_def ON t_op (x_def);
+CREATE INDEX t_op_x_cust ON t_op (x_cust test_int4_ops);
+ANALYZE t_op;
+
+-- (4) index default, query default
+EXPLAIN (COSTS OFF)
+SELECT x_def + 1 FROM t_op ORDER BY 1;
+
+-- (1) index default, query custom
+EXPLAIN (COSTS OFF)
+SELECT x_def + 1 FROM t_op ORDER BY 1 USING <#;
+
+-- (2) index custom, query default
+EXPLAIN (COSTS OFF)
+SELECT x_cust + 1 FROM t_op ORDER BY 1;
+
+-- (3) index custom, query custom
+EXPLAIN (COSTS OFF)
+SELECT x_cust + 1 FROM t_op ORDER BY 1 USING <#;
+
+--
+-- SLOPE (Scalar function Leveraging Ordered Path Evaluation)
+-- Test that monotonic functions can use indexes for ordering
+--
+
+-- Create test table with various data types
+CREATE TABLE src (
+    id serial PRIMARY KEY,
+    v_int2 int2,
+    v_int4 int4,
+    v_int8 int8,
+    v_float4 float4,
+    v_float8 float8,
+    v_numeric numeric,
+    ts timestamp,
+    tstz timestamptz
+);
+
+-- Insert some test data
+INSERT INTO src (v_int2, v_int4, v_int8, v_float4, v_float8, v_numeric, ts, tstz)
+SELECT
+    (i % 100)::int2,
+    i,
+    i::int8,
+    i::float4,
+    i::float8,
+    i::numeric,
+    '2020-01-01'::timestamp + (i || ' hours')::interval,
+    '2020-01-01'::timestamptz + (i || ' hours')::interval
+FROM generate_series(1, 1000) i;
+
+-- Create indexes on the columns we'll test
+CREATE INDEX src_v_int4_idx ON src (v_int4);
+CREATE INDEX src_v_int8_idx ON src (v_int8);
+CREATE INDEX src_v_float8_idx ON src (v_float8);
+CREATE INDEX src_v_numeric_idx ON src (v_numeric);
+CREATE INDEX src_ts_idx ON src (ts);
+CREATE INDEX src_tstz_idx ON src (tstz);
+
+-- Analyze to get good statistics
+ANALYZE src;
+
+--
+-- Test that SLOPE is enabled and can be disabled
+--
+
+SET enable_slope = off;
+-- Basic: floor(float8) should sort even though the index order
+-- produces the result in the correct order.
+explain (costs off, verbose)
+select floor(v_float8), count(*) from src group by 1;
+RESET enable_slope;
+SHOW enable_slope;
+
+--
+-- Test GROUP BY with monotonic function
+--
+
+-- Basic: floor(float8) should use index on v_float8
+explain (costs off, verbose)
+select floor(v_float8), count(*) from src group by 1;
+
+-- ceil(float8) should use index on v_float8
+explain (costs off, verbose)
+select ceil(v_float8), count(*) from src group by 1;
+
+-- floor(numeric) should use index on v_numeric
+explain (costs off, verbose)
+select floor(v_numeric), count(*) from src group by 1;
+
+-- timestamp::date cast should use index on ts
+explain (costs off, verbose)
+select ts::date, count(*) from src group by 1;
+
+-- date_trunc on timestamp should use index
+explain (costs off, verbose)
+select date_trunc('day', ts), count(*) from src group by 1;
+
+-- date_trunc on timestamptz should not use index
+explain (costs off, verbose)
+select date_trunc('day', tstz), count(*) from src group by 1;
+
+
+--
+-- Test arithmetic operations
+--
+
+-- Addition: v_int4 + 10 is increasing in v_int4
+explain (costs off, verbose)
+select v_int4 + 10, count(*) from src group by 1;
+
+-- Subtraction: v_int4 - 10 is increasing in v_int4
+explain (costs off, verbose)
+select v_int4 - 10, count(*) from src group by 1;
+
+-- Multiplication by positive constant: v_int4 * 2 is increasing
+explain (costs off, verbose)
+select v_int4 * 2, count(*) from src group by 1;
+
+-- Division by positive constant: v_int4 / 2 is increasing
+explain (costs off, verbose)
+select v_int4 / 2, count(*) from src group by 1;
+
+
+--
+-- Test decreasing functions
+-- These queries can't use the index order because group pathkeys
+-- are always ASC NULLS LAST, while -int_v4 is DESC NULLS LAST.
+
+-- Unary minus: -v_int4 is decreasing in v_int4
+explain (costs off, verbose)
+select -v_int4, count(*) from src group by 1;
+
+-- Subtraction from constant: 1000 - v_int4 is decreasing in v_int4
+explain (costs off, verbose)
+select 1000 - v_int4, count(*) from src group by 1;
+
+-- Multiplication by negative constant: v_int4 * (-2) is decreasing
+explain (costs off, verbose)
+select v_int4 * (-2), count(*) from src group by 1;
+
+-- Division by negative constant: v_int4 / (-2) is decreasing
+explain (costs off, verbose)
+select v_int4 / (-2), count(*) from src group by 1;
+
+--
+-- Test ORDER BY with monotonic function
+--
+
+-- ORDER BY floor(v_float8) should use index
+explain (costs off, verbose)
+select floor(v_float8), v_float8 from src order by 1;
+
+-- ORDER BY -v_int4 DESC requires sorting
+-- the index order is v_int4 ASC and, implicitly, NULLS LAST
+-- a forward scan gives -v_int4 DESC NULLS LAST
+-- a backward scan gives -v_int4 ASC NULLS FIRST
+-- the query order is DESC and, implicitly, NULLS FIRST
+explain (costs off, verbose)
+select -v_int4 from src order by 1 desc;
+
+-- ORDER BY -v_int4 ASC requires sorting
+-- a forward scan gives -v_int4 DESC NULLS LAST
+-- a backward scan gives -v_int4 ASC NULLS FIRST
+-- the query order is -v_int4 ASC NULLS LAST
+explain (costs off, verbose)
+select -v_int4 from src order by 1;
+
+--
+-- Group and order
+--
+
+explain (costs off, verbose)
+select tstz::date, count(*) from src group by 1 order by 1;
+
+--
+-- Test nested monotonic function
+--
+
+-- floor(floor(x)) should still use index
+explain (costs off, verbose)
+select floor(floor(v_float8)), count(*) from src group by 1;
+
+-- floor(v + 1) should use index
+explain (costs off, verbose)
+select floor(v_float8 + 1), count(*) from src group by 1;
+
+--
+-- Test all index/query direction+nulls combinations for SLOPE.
+-- For an increasing function like floor(), the scan uses the index when both
+-- direction and nulls agree (Forward) or both are flipped (Backward).
+-- If one agrees and the other disagrees, then the scan cannot use the
+-- index and a sort is required.
+--
+CREATE TABLE nulls_tmp (v float8);
+INSERT INTO nulls_tmp VALUES (1), (NULL), (2);
+ANALYZE nulls_tmp;
+
+CREATE TEMPORARY TABLE nulls_results (
+    seq serial,
+    sign text, index_order text, query_order text, scan_method text,
+    example text
+);
+DO $$
+DECLARE
+    r record;
+    plan_json json;
+    node_type text;
+    query text;
+    r1 text;
+    r2 text;
+BEGIN
+    FOR r IN
+        SELECT idx_dir, idx_nf, qry_dir, qry_nf, sign
+        FROM unnest(ARRAY['ASC','DESC'])   WITH ORDINALITY AS idx_dir(idx_dir, idx_dir_i),
+             unnest(ARRAY['FIRST','LAST']) WITH ORDINALITY AS idx_nf (idx_nf,  idx_nf_i),
+             unnest(ARRAY['+','-'])        WITH ORDINALITY AS sign   (sign,    sign_i),
+             unnest(ARRAY['FIRST','LAST']) WITH ORDINALITY AS qry_nf (qry_nf,  qry_nf_i),
+             unnest(ARRAY['ASC','DESC'])   WITH ORDINALITY AS qry_dir(qry_dir, qry_dir_i)
+        ORDER BY qry_dir_i, qry_nf_i, sign_i, idx_dir_i, idx_nf_i
+    LOOP
+        EXECUTE format('CREATE INDEX nulls_tmp_idx ON nulls_tmp (v %s NULLS %s)',
+                        r.idx_dir, r.idx_nf);
+        query := format('SELECT floor(0.5 %s v) as x FROM nulls_tmp ORDER BY 1 %s NULLS %s',
+            r.sign, r.qry_dir, r.qry_nf);
+        EXECUTE format('EXPLAIN (FORMAT JSON) %s', query)
+        INTO plan_json;
+
+        SET enable_slope = off;
+        EXECUTE 'SELECT string_agg(coalesce(x::text, ''NULL''), '','') FROM (' || query || ') tmp(x)' into r1;
+        SET enable_slope = on;
+        EXECUTE 'SELECT string_agg(coalesce(x::text, ''NULL''), '','') FROM (' || query || ') tmp(x)' into r2;
+        if r1 <> r2 then
+            raise notice 'query %', query;
+            raise exception ' r1=%, r2=%', r1, r2;
+        end if;
+        node_type := plan_json->0->'Plan'->>'Node Type';
+        INSERT INTO nulls_results (sign, index_order, query_order, scan_method, example) VALUES (
+            r.sign,
+            r.idx_dir || ' NULLS ' || r.idx_nf,
+            r.qry_dir || ' NULLS ' || r.qry_nf,
+            CASE WHEN node_type IN ('Index Only Scan', 'Index Scan')
+                 THEN plan_json->0->'Plan'->>'Scan Direction'
+                 ELSE node_type
+            END,
+            r2
+        );
+        EXECUTE format('DROP INDEX nulls_tmp_idx');
+    END LOOP;
+END;
+$$;
+
+SELECT sign, index_order, query_order, scan_method, example
+FROM nulls_results ORDER BY seq;
+
+--
+-- Test special numeric values
+--
+-- This check the correctness of the query for a nullable column
+-- also containing special values (-inf, +inf and NaN)
+-- represented as float8, float4, numeric and user defined
+-- domains with float8 base type.
+CREATE DOMAIN non42 AS float8 CHECK (value != 42);
+CREATE DOMAIN fp_real AS float8;
+CREATE UNLOGGED TABLE numeric_corners (i serial, x float8);
+INSERT INTO numeric_corners (x)
+SELECT x::float8 as x FROM
+        unnest(ARRAY['-inf', '-1', '0', '3', 'inf', 'nan', NULL]) x(x)
+ORDER BY 1;
+
+CREATE INDEX numeric_corners_xfp8_idx_nulllast ON numeric_corners ((x::float8) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xfp8_idx_nullfirst ON numeric_corners ((x::float8) ASC NULLS FIRST);
+CREATE INDEX numeric_corners_xfp4_idx_nulllast ON numeric_corners ((x::float4) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xfp4_idx_nullfirst ON numeric_corners ((x::float4) ASC NULLS FIRST);
+CREATE INDEX numeric_corners_xnum_idx_nulllast ON numeric_corners ((x::numeric) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xnum_idx_nullfirst ON numeric_corners ((x::numeric) ASC NULLS FIRST);
+CREATE INDEX numeric_corners_xnon42_idx_nulllast ON numeric_corners ((x::non42) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xnon42_idx_nullfirst ON numeric_corners ((x::non42) ASC NULLS FIRST);
+CREATE INDEX numeric_corners_xfpreal_idx_nulllast ON numeric_corners ((x::fp_real) ASC NULLS LAST);
+CREATE INDEX numeric_corners_xfpreal_idx_nullfirst ON numeric_corners ((x::fp_real) ASC NULLS FIRST);
+
+CREATE TEMPORARY TABLE numeric_corners_results (
+    seq serial,
+    expr text COLLATE "C",
+    sort_order text COLLATE "C",
+    nulls_order text COLLATE "C",
+    typename text COLLATE "C",
+    result float8[],
+    expected float8[],
+    expected_seq int4[],
+    nan_values float8[],
+    plan1_json json,
+    plan2_json json
+);
+DO $$
+DECLARE
+    r record;
+    result float8[];
+    expected float8[];
+    expected_seq int4[];
+    nan_values float8[];
+    query text;
+    agg_query text;
+    plan_query text;
+    nan_query text;
+    expected_seq_query text;
+    plan1_json json;
+    plan2_json json;
+BEGIN
+
+    FOR r IN
+        SELECT replace(replace(s, 'x', 'x::' || t), 'A', a || '::' || t) as expr, sort_order, nulls_order, t as typename
+        FROM
+        unnest(ARRAY['ASC', 'DESC']) WITH ORDINALITY AS so(sort_order, so_i),
+        unnest(ARRAY['FIRST', 'LAST']) WITH ORDINALITY AS nf(nulls_order, no_i),
+        unnest(ARRAY['float8', 'float4', 'numeric', 'non42', 'fp_real']) WITH ORDINALITY AS t(t, t_i),
+        unnest(ARRAY[
+        'x+A', 'x-A', 'x*A', 'x/A',
+        'A+x', 'A-x', 'A*x', '-x'
+        ]) WITH ORDINALITY AS s(s, s_i),
+        unnest(ARRAY['''inf''', '''-inf''', '1']) WITH ORDINALITY AS a(a, a_i)
+        ORDER BY s_i, so_i, no_i, t_i
+    LOOP
+        query := 'SELECT *, (' || r.expr || ') as f '
+                 || 'FROM numeric_corners '
+                 || 'ORDER BY f ' || r.sort_order || ' NULLS ' || r.nulls_order;
+        nan_query := 'SELECT array_agg(x) FROM numeric_corners
+        WHERE ' || r.expr || ' = ''nan''::' || r.typename || ' AND x != ''nan''::' || r.typename;
+        agg_query := 'SELECT array_agg(f::float8) FROM (' || query || ') tmp';
+        expected_seq_query := 'SELECT array_agg(i::int4) FROM (' || query || ') tmp';
+        plan_query := 'EXPLAIN (FORMAT JSON) ' || query;
+        -- slope optimization disabled
+        SET enable_slope = off;
+        EXECUTE agg_query into result;
+        EXECUTE plan_query into plan2_json;
+        -- slope optimization enabled
+        SET enable_slope = on;
+        EXECUTE agg_query into expected;
+        EXECUTE expected_seq_query into expected_seq;
+        EXECUTE nan_query into nan_values;
+        EXECUTE plan_query into plan1_json;
+        -- store results
+        INSERT INTO numeric_corners_results
+        (expr, sort_order, nulls_order, typename, result, expected, expected_seq, nan_values, plan1_json, plan2_json)
+        VALUES (r.expr, r.sort_order, r.nulls_order, r.typename, result, expected, expected_seq, nan_values, plan1_json, plan2_json);
+    END LOOP;
+END;
+$$;
+
+-- display failing test cases
+SELECT seq, expr
+, sort_order, nulls_order
+, expected, expected_seq, result, nan_values
+, plan2_json->0->'Plan'->>'Node Type' as plan2
+FROM numeric_corners_results
+WHERE expected != result;
+
+-- check the number of test cases
+SELECT expected = result as passed, count(1)
+FROM numeric_corners_results
+GROUP BY 1;
+
+--
+-- Test pathkey chains: f(x), x (descending chain)
+-- The order of f(x) is implied by the order of x when f is monotonic.
+--
+
+explain (costs off, verbose)
+select * from src order by ts::date, ts;
+
+--
+-- Test pathkey chains: x, f(x) (ascending chain)
+-- f(x) is constant when x is constant, so f(x) is a redundant tiebreaker.
+--
+-- When btree equality on x implies image equality (equalimage), any
+-- deterministic f(x) is constant on each index group, so f(x) is a
+-- redundant tiebreaker and the index on x alone can satisfy ORDER BY x, f(x).
+-- When equalimage does not hold, values that compare equal under the index
+-- can still differ in f(x); the planner must keep sorting on f(x),
+-- unless f(x) is monotonic
+--
+
+-- int4 has btequalimage; numeric/float8 do not (e.g. 1.0 = 1.00, -0.0 = 0.0).
+
+explain (costs off, verbose)
+select * from src order by ts, ts::date;
+
+-- non-monotonic f(x) after pinned x with equalimage
+-- can safely ignore f(x) tie breaker and use index order alone
+explain (costs off, verbose)
+select v_int4, abs(v_int4)
+from src order by v_int4, abs(v_int4);
+
+-- non-monotonic f(x) before x: cannot use index order alone
+explain (costs off, verbose)
+select mod(v_int4, 7), v_int4
+from src order by mod(v_int4, 7), v_int4;
+
+-- float_ops has no equalimage so it requires incremental sort
+explain (costs off, verbose)
+select * from src order by v_float8, cos(v_float8);
+
+-- monotonic f without equalimage: index order on n implies order on 2 * n
+explain (costs off, verbose)
+select * from src order by v_float8, 2 * v_float8;
+
+--
+-- INCLUDE columns are not index sort keys; build_index_pathkeys must stop
+-- at nkeycolumns and must not treat INCLUDE attrs as an ordered prefix.
+-- Use a separate table so src's index set stays unchanged.
+--
+CREATE TABLE inc (
+    k int4,
+    ts timestamp
+);
+
+INSERT INTO inc
+SELECT i, '2020-01-01'::timestamp + (i || ' hours')::interval
+FROM generate_series(1, 100) i;
+
+CREATE INDEX inc_k_inc_ts_idx ON inc (k) INCLUDE (ts);
+CREATE INDEX inc_ts_idx ON inc (ts);
+ANALYZE inc;
+
+explain (costs off, verbose)
+select k, ts from inc order by k;
+
+explain (costs off, verbose)
+select k, ts::date from inc order by k, ts::date;
+
+explain (costs off, verbose)
+select ts::date from inc order by ts::date;
+
+--
+-- Test DISTINCT with monotonic functions
+--
+
+explain (costs off, verbose)
+select distinct ts::date from src;
+
+--
+-- Test window functions with PARTITION BY monotonic function
+--
+
+explain (costs off, verbose)
+select ts::date, ts, row_number() over (partition by ts::date order by ts)
+from src;
+
+--
+-- RelabelType across binary-compatible types with different default
+-- orders: int4 (signed) vs oid (unsigned).  Index order on int4 is not
+-- oid order, so SLOPE must not omit the Sort.
+--
+EXPLAIN (COSTS OFF)
+SELECT v_int4::oid FROM src ORDER BY 1;
+
+DROP SCHEMA slope CASCADE;
\ No newline at end of file
-- 
2.53.0

