v9-0007-FIX-NaN-special-cases.patch

From ac94c93acb7cc002fa5f0f2a143e579a9075ecb8 Mon Sep 17 00:00:00 2001
From: Alexandre Felipe <o.alexandre.felipe@gmail.com>
Date: Fri, 3 Jul 2026 06:24:09 +0100
Subject: [PATCH 7/7] FIX: NaN special cases

This patch takes into account the numeric special cases
involving NaN.
A NaN represents a numeric indeterminate, e.g.:
 * +inf - inf = NaN
 * inf * 0 = NaN
And they propagate, e.g. f(NaN) = f(NaN, x) = f(x, NaN) = NaN

NaN is placed above all other numeric (not null) values.
Two classes of violations are possible:
1. NaN produced from a non NaN value, can be out of order,
   e.g. 0 appears before 1, 0 * inf = NaN should appear
   after 0 * 1 = 0.

2. NaN produced from a NaN value, on the wrong side when
   the sort key is decreasing on the sort key.
   e.g. scanning x asc has NaN at the after non-NaN values,
   sorting -x desc should have NaN at the before non-NaN values.

The first class is handled by adding conditions to the prosupport
functions, the second class is handled at the pathkey analysis,
rejecting decreasing functions on types types that can produce NaN.

---
 src/backend/optimizer/path/pathkeys.c |  99 +++++--
 src/backend/utils/adt/misc.c          | 408 +++++++++++++++-----------
 src/include/catalog/pg_proc.dat       |   6 +-
 src/include/nodes/pathnodes.h         |   5 +-
 src/include/nodes/plannodes.h         |   6 +
 src/test/regress/expected/slope.out   | 154 +++++++++-
 src/test/regress/sql/slope.sql        | 144 ++++++++-
 7 files changed, 625 insertions(+), 197 deletions(-)

diff --git a/src/backend/optimizer/path/pathkeys.c b/src/backend/optimizer/path/pathkeys.c
index d5f6439f7f8..4b0cecc4e36 100644
--- a/src/backend/optimizer/path/pathkeys.c
+++ b/src/backend/optimizer/path/pathkeys.c
@@ -18,7 +18,9 @@
 #include "postgres.h"
 
 #include "access/stratnum.h"
+#include "access/transam.h"
 #include "catalog/pg_opfamily.h"
+#include "catalog/pg_type.h"
 #include "fmgr.h"
 #include "nodes/nodeFuncs.h"
 #include "nodes/supportnodes.h"
@@ -41,7 +43,8 @@ static bool matches_boolean_partition_clause(RestrictInfo *rinfo,
 											 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 MonotonicFunction get_expr_slope_wrt(Expr *expr, Expr *target,
+											bool target_cannan);
 static PathKey *slope_emit_pathkey(PlannerInfo *root, PathKey *pk,
 								   Expr *indexkey, bool reverse_sort,
 								   bool nulls_first);
@@ -861,6 +864,49 @@ get_variation_source(Expr *expr, Expr **inner_out, Index *relid_out)
 	}
 }
 
+/*
+ * var_type_can_nan
+ *		True if a variation-source column can hold NaN values.
+ */
+static bool
+var_type_can_nan(Var *var)
+{
+	Oid var_type = var->vartype;
+	
+	if(var_type >= FirstNormalObjectId)
+		var_type = getBaseType(var_type);
+	
+	switch (var_type)
+	{
+		case FLOAT4OID:
+		case FLOAT8OID:
+		case NUMERICOID:
+			return true;
+		default:
+			return false;
+	}
+}
+
+/*
+ * finalize_slope
+ *		Apply NaN-aware restrictions to a composed monotonicity slope.
+ *
+ * Decreasing functions do not preserve sort order when the variation
+ * source can produce NaNs.  Flat functions are treated similarly, since
+ * NaNs and infinities in the source can still change the result.  An
+ * overall decreasing (or flat) result is only rejected once composition
+ * reaches the source.  For example, (1 - (1 - x)) is increasing in x even
+ * though (1 - x) is decreasing.
+ */
+static MonotonicFunction
+finalize_slope(MonotonicFunction slope, bool target_cannan)
+{
+	if (target_cannan &&
+		(slope == MONOTONICFUNC_DECREASING || slope == MONOTONICFUNC_BOTH))
+		return MONOTONICFUNC_NONE;
+	return slope;
+}
+
 /*
  * get_expr_slope_wrt
  *	  Determine the monotonicity slope of an expression with respect to
@@ -869,10 +915,11 @@ get_variation_source(Expr *expr, Expr **inner_out, Index *relid_out)
  * Returns the slope of 'expr' with respect to 'target'
  *   MONOTONICFUNC_INCREASING: monotonically increasing
  *   MONOTONICFUNC_DECREASING: monotonically decreasing
+ *   MONOTONICFUNC_BOTH:       independent of target
  *   MONOTONICFUNC_NONE:       cannot determine monotonicity
  */
 static MonotonicFunction
-get_expr_slope_wrt(Expr *expr, Expr *target)
+get_expr_slope_wrt(Expr *expr, Expr *target, bool target_cannan)
 {
 	MonotonicFunction slope = MONOTONICFUNC_INCREASING;
 
@@ -889,7 +936,7 @@ get_expr_slope_wrt(Expr *expr, Expr *target)
 
 		/* Check if we've reached the target */
 		if (equal(expr, target))
-			return slope;
+			return finalize_slope(slope, target_cannan);
 
 		/* Skip RelabelType (no-op coercion) */
 		if (IsA(expr, RelabelType))
@@ -944,7 +991,7 @@ get_expr_slope_wrt(Expr *expr, Expr *target)
 		if (req.slopes == NULL || req.nslopes <= 0)
 			return MONOTONICFUNC_NONE;
 
-		/* Find the single non-constant argument */
+		/* Find the single non-constant argument that can affect the result */
 		i = 0;
 		foreach(lc, args)
 		{
@@ -952,27 +999,36 @@ get_expr_slope_wrt(Expr *expr, Expr *target)
 
 			if (!IsA(arg, Const))
 			{
+				MonotonicFunction arg_slope;
+
+				if (unlikely(i >= req.nslopes))
+					return MONOTONICFUNC_NONE;
+
+				arg_slope = req.slopes[i];
+				if (arg_slope == MONOTONICFUNC_BOTH)
+				{
+					i++;
+					continue;
+				}
+				if (arg_slope == MONOTONICFUNC_DECREASING)
+					func_arg_slope = MONOTONICFUNC_DECREASING;
+				else if (arg_slope != MONOTONICFUNC_INCREASING)
+					return MONOTONICFUNC_NONE;
+
 				if (next_expr != NULL)
 				{
 					/* Multivariate - check if this is the target */
-					return equal(expr, target) ? slope : MONOTONICFUNC_NONE;
+					if (equal(expr, target))
+						return finalize_slope(slope, target_cannan);
+					return 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 */
+			return finalize_slope(MONOTONICFUNC_BOTH, target_cannan);
 
 		/* Compose slopes */
 		if (func_arg_slope == MONOTONICFUNC_DECREASING)
@@ -1007,7 +1063,8 @@ precompute_slope_pathkeys(PlannerInfo *root)
 
 		pk->pk_var = NULL;
 		pk->pk_varrelid = 0;
-		pk->pk_slope = MONOTONICFUNC_BOTH;	/* not yet computed */
+		pk->pk_slope = MONOTONICFUNC_UNKNOWN;
+		pk->pk_var_cannan = false;
 
 		if (pk->pk_eclass->ec_has_volatile ||
 			pk->pk_eclass->ec_members == NIL)
@@ -1024,6 +1081,8 @@ precompute_slope_pathkeys(PlannerInfo *root)
 		if (pk->pk_var == NULL || pk->pk_varrelid == 0 ||
 			pk->pk_var == em->em_expr)
 			pk->pk_var = NULL;
+		else if (IsA(pk->pk_var, Var))
+			pk->pk_var_cannan = var_type_can_nan((Var *) pk->pk_var);
 	}
 }
 
@@ -1188,19 +1247,21 @@ build_index_pathkeys(PlannerInfo *root,
 					{
 						PathKey    *spk;
 
-						if (qpk->pk_slope == MONOTONICFUNC_BOTH)
+						if (qpk->pk_slope == MONOTONICFUNC_UNKNOWN)
 						{
 							EquivalenceMember *em;
 
 							em = linitial(qpk->pk_eclass->ec_members);
 							qpk->pk_slope = get_expr_slope_wrt(em->em_expr,
-															   qpk->pk_var);
+															   qpk->pk_var,
+															   qpk->pk_var_cannan);
 						}
 						spk = slope_emit_pathkey(root, qpk, indexkey,
 												 reverse_sort, nulls_first);
 						if (spk && !pathkey_is_redundant(spk, retval))
 							retval = lappend(retval, spk);
-						slope_matched = true;
+						if (spk)
+							slope_matched = true;
 						continue;
 					}
 
diff --git a/src/backend/utils/adt/misc.c b/src/backend/utils/adt/misc.c
index 19b7b48e80a..09eebd4a1e4 100644
--- a/src/backend/utils/adt/misc.c
+++ b/src/backend/utils/adt/misc.c
@@ -33,6 +33,7 @@
 #include "miscadmin.h"
 #include "nodes/miscnodes.h"
 #include "nodes/supportnodes.h"
+#include "nodes/nodeFuncs.h"
 #include "parser/parse_type.h"
 #include "parser/scansup.h"
 #include "pgstat.h"
@@ -51,6 +52,23 @@
 #include "utils/tuplestore.h"
 #include "utils/wait_event.h"
 
+/**
+ * Extended numeric sign, the usual -1, 0, 1, 
+ * for real numbers, and additional values for non-real 
+ * numeric values and null.
+*/
+
+typedef enum NUMERIC_SIGN
+{
+	NUMERIC_SIGN_NINF = -2,
+	NUMERIC_SIGN_NEG = -1,
+	NUMERIC_SIGN_ZERO = 0,
+	NUMERIC_SIGN_POS = 1,
+	NUMERIC_SIGN_PINF = 2,
+	NUMERIC_SIGN_NAN = 3,
+	NUMERIC_SIGN_NULL = 4,
+} NUMERIC_SIGN;
+
 
 /*
  * structure to cache metadata needed in pg_input_is_valid_common
@@ -1121,84 +1139,37 @@ monotonic_slope_support(Node *rawreq, int nslopes,
 	return PointerGetDatum(NULL);
 }
 
-/*
- * arg0_asc_slope_support
- *		Prosupport: f(x, ...) is monotonically increasing in x.
- */
-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);
-}
-
-/*
- * arg0_desc_slope_support
- *		Prosupport: f(x, ...) is monotonically decreasing in x.
- */
-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);
-}
-
-/*
- * arg1_asc_slope_support
- *		Prosupport: f(a, x, ...) is monotonically increasing in x.
- */
-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);
-}
-
-/*
- * diff_slope_support
- *		Prosupport: f(x, y) = x - y is increasing in x, decreasing in y.
- */
-Datum
-diff_slope_support(PG_FUNCTION_ARGS)
-{
-	static const MonotonicFunction pattern[2] = {MONOTONICFUNC_INCREASING,
-												 MONOTONICFUNC_DECREASING};
-
-	return monotonic_slope_support((Node *) PG_GETARG_POINTER(0),
-								   lengthof(pattern), pattern);
-}
 
-/*
- * addition_slope_support
- *		Prosupport: f(x, y) = x + y is increasing in both x and y.
- */
-Datum
-addition_slope_support(PG_FUNCTION_ARGS)
+static bool
+get_2slope_args(Node *rawreq, Node **arg0, Node **arg1)
 {
-	static const MonotonicFunction pattern[2] = {MONOTONICFUNC_INCREASING,
-												 MONOTONICFUNC_INCREASING};
-
-	return monotonic_slope_support((Node *) PG_GETARG_POINTER(0),
-								   lengthof(pattern), pattern);
+	if (!IsA(rawreq, SupportRequestMonotonic))
+		return false;
+	SupportRequestMonotonic *req = (SupportRequestMonotonic *) rawreq;
+	List	   *args;
+
+	if (IsA(req->expr, FuncExpr))
+		args = ((FuncExpr *) req->expr)->args;
+	else if (IsA(req->expr, OpExpr))
+		args = ((OpExpr *) req->expr)->args;
+	else
+		return false;
+	if (list_length(args) < 2)
+		return false;
+	*arg0 = (Node *) linitial(args);
+	*arg1 = (Node *) lsecond(args);
+	return true;
 }
-
 /*
  * get_const_sign
- *		Helper to determine the sign of a numeric constant.
- *		Returns 1 for positive, -1 for negative, 0 for zero or unknown.
+ *  Helper to determine the sign of a numeric constant.
+ *  It will classify a number according 
  */
-static int
+static inline NUMERIC_SIGN
 get_const_sign(Const *constval)
 {
 	if (constval->constisnull)
-		return 0;
+		return NUMERIC_SIGN_NULL;
 
 	switch (constval->consttype)
 	{
@@ -1224,38 +1195,182 @@ get_const_sign(Const *constval)
 			{
 				float4		val = DatumGetFloat4(constval->constvalue);
 
-				if (isnan(val) || val == 0.0f)
-					return 0;
-				return (val > 0.0f) ? 1 : -1;
+				if (isnan(val))
+					return NUMERIC_SIGN_NAN;
+				if (isinf(val))
+					return val > 0 ? NUMERIC_SIGN_PINF : NUMERIC_SIGN_NINF;
+				else if (val == 0)
+					return NUMERIC_SIGN_ZERO;
+				else
+					return val > 0 ? NUMERIC_SIGN_POS : NUMERIC_SIGN_NEG;
 			}
 		case FLOAT8OID:
 			{
 				float8		val = DatumGetFloat8(constval->constvalue);
 
-				if (isnan(val) || val == 0.0)
-					return 0;
-				return (val > 0.0) ? 1 : -1;
+				if (isnan(val))
+					return NUMERIC_SIGN_NAN;
+				if (isinf(val))
+					return val > 0 ? NUMERIC_SIGN_PINF : NUMERIC_SIGN_NINF;
+				else if (val == 0)
+					return NUMERIC_SIGN_ZERO;
+				else
+					return val > 0 ? NUMERIC_SIGN_POS : NUMERIC_SIGN_NEG;
 			}
 		case NUMERICOID:
 			{
 				Numeric		num = DatumGetNumeric(constval->constvalue);
 				Datum		result;
 				Numeric		sign_num;
-				int			sign;
+				int			val_sign;
 
-				if (numeric_is_nan(num) || numeric_is_inf(num))
-					return 0;
+				if (numeric_is_nan(num))
+					return NUMERIC_SIGN_NAN;
 
 				result = DirectFunctionCall1(numeric_sign, NumericGetDatum(num));
 				sign_num = DatumGetNumeric(result);
-				sign = numeric_int4_safe(sign_num, NULL);
-				return sign;
+				val_sign = numeric_int4_safe(sign_num, NULL);
+				if (numeric_is_inf(num))
+					return val_sign == 1 ? NUMERIC_SIGN_PINF : NUMERIC_SIGN_NINF;
+				else
+					return (enum NUMERIC_SIGN) val_sign;
 			}
 		default:
 			return 0;
 	}
 }
 
+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};
+
+
+
+/*
+ * arg0_asc_slope_support
+ *		Prosupport: f(x, ...) is monotonically increasing in x.
+ */
+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);
+}
+
+ /*
+ * arg0_desc_slope_support
+ *		Prosupport: f(x, ...) is monotonically decreasing in x.
+ */
+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);
+}
+
+ /*
+  * arg1_asc_slope_support
+  *		Prosupport: f(a, x, ...) is monotonically increasing in x.
+  */
+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);
+}
+
+/*
+* diff_slope_support
+*		Prosupport: f(x, y) = x - y is increasing in x, decreasing in y.
+*/
+Datum
+diff_slope_support(PG_FUNCTION_ARGS)
+{
+	Node	   *rawreq = (Node *) PG_GETARG_POINTER(0);
+	Node	   *arg0;
+	Node	   *arg1;
+
+	if (!get_2slope_args(rawreq, &arg0, &arg1))
+		PG_RETURN_POINTER(NULL);
+
+	if (IsA(arg0, Const))
+	{
+		switch (get_const_sign((Const *) arg0))
+		{
+			case NUMERIC_SIGN_NINF:
+			case NUMERIC_SIGN_PINF:
+			case NUMERIC_SIGN_NAN:
+				return monotonic_slope_support(rawreq, 2, flat_slope);
+			default:
+				break;
+		}
+	}
+	else if (IsA(arg1, Const))
+	{
+		switch (get_const_sign((Const *) arg1))
+		{
+			case NUMERIC_SIGN_NINF:
+			case NUMERIC_SIGN_PINF:
+			case NUMERIC_SIGN_NAN:
+				return monotonic_slope_support(rawreq, 2, flat_slope);
+			default:
+				break;
+		}
+	}
+
+	return monotonic_slope_support(rawreq, 2, diff_slope);
+}
+
+/*
+* addition_slope_support
+*		Prosupport: f(x, y) = x + y is increasing in both x and y.
+*/
+Datum
+addition_slope_support(PG_FUNCTION_ARGS)
+{
+	Node	   *rawreq = (Node *) PG_GETARG_POINTER(0);
+	Node	   *arg0;
+	Node	   *arg1;
+
+	if (!get_2slope_args(rawreq, &arg0, &arg1))
+		PG_RETURN_POINTER(NULL);
+
+	if (IsA(arg0, Const))
+	{
+		switch (get_const_sign((Const *) arg0))
+		{
+			case NUMERIC_SIGN_PINF:
+			case NUMERIC_SIGN_NINF:
+			case NUMERIC_SIGN_NAN:
+				return monotonic_slope_support(rawreq, 2, flat_slope);
+			default:
+				break;
+		}
+	}
+	else if (IsA(arg1, Const))
+	{
+		switch (get_const_sign((Const *) arg1))
+		{
+			case NUMERIC_SIGN_PINF:
+			case NUMERIC_SIGN_NINF:
+			case NUMERIC_SIGN_NAN:
+				return monotonic_slope_support(rawreq, 2, flat_slope);
+			default:
+				break;
+		}
+	}
+
+	return monotonic_slope_support(rawreq, 2, asc_slope);
+}
+
 /*
  * multiply_slope_support
  *		Prosupport: x * c is increasing if c > 0, decreasing if c < 0.
@@ -1270,67 +1385,34 @@ Datum
 multiply_slope_support(PG_FUNCTION_ARGS)
 {
 	Node	   *rawreq = (Node *) PG_GETARG_POINTER(0);
+	Const	   *constval;
+	Node	   *arg0;
+	Node	   *arg1;
 
-	if (IsA(rawreq, SupportRequestMonotonic))
-	{
-		SupportRequestMonotonic *req = (SupportRequestMonotonic *) rawreq;
-		List	   *args;
-		Node	   *arg0;
-		Node	   *arg1;
-		Const	   *constval = NULL;
-		int			var_argno = -1;
-		int			sign;
-
-		if (IsA(req->expr, FuncExpr))
-			args = ((FuncExpr *) req->expr)->args;
-		else if (IsA(req->expr, OpExpr))
-			args = ((OpExpr *) req->expr)->args;
-		else
-			PG_RETURN_POINTER(NULL);
-
-		if (list_length(args) != 2)
-			PG_RETURN_POINTER(NULL);
-
-		arg0 = (Node *) linitial(args);
-		arg1 = (Node *) lsecond(args);
+	if (!get_2slope_args(rawreq, &arg0, &arg1))
+		PG_RETURN_POINTER(NULL);
 
-		if (IsA(arg0, Const))
-		{
-			constval = (Const *) arg0;
-			var_argno = 1;
-		}
-		else if (IsA(arg1, Const))
-		{
-			constval = (Const *) arg1;
-			var_argno = 0;
-		}
-		else
-			PG_RETURN_POINTER(NULL);
+	if (IsA(arg0, Const))
+		constval = (Const *) arg0;
+	else if (IsA(arg1, Const))
+		constval = (Const *) arg1;
+	else
+		PG_RETURN_POINTER(NULL);
 
-		sign = get_const_sign(constval);
-		if (sign == 0)
+	switch (get_const_sign(constval))
+	{
+		case NUMERIC_SIGN_POS:
+			return monotonic_slope_support(rawreq, 2, asc_slope);
+		case NUMERIC_SIGN_NEG:
+			return monotonic_slope_support(rawreq, 2, desc_slope);
+		case NUMERIC_SIGN_NAN:
+			return monotonic_slope_support(rawreq, 2, flat_slope);
+		case NUMERIC_SIGN_ZERO:
+			return monotonic_slope_support(rawreq, 2, flat_slope);
+		default:
 			PG_RETURN_POINTER(NULL);
-
-		{
-			static const MonotonicFunction asc_first[2] = {MONOTONICFUNC_INCREASING,
-														   MONOTONICFUNC_NONE};
-			static const MonotonicFunction asc_second[2] = {MONOTONICFUNC_NONE,
-															MONOTONICFUNC_INCREASING};
-			static const MonotonicFunction desc_first[2] = {MONOTONICFUNC_DECREASING,
-															MONOTONICFUNC_NONE};
-			static const MonotonicFunction desc_second[2] = {MONOTONICFUNC_NONE,
-															 MONOTONICFUNC_DECREASING};
-
-			if (var_argno == 0)
-				req->slopes = (sign > 0) ? asc_first : desc_first;
-			else
-				req->slopes = (sign > 0) ? asc_second : desc_second;
-			req->nslopes = 2;
-			PG_RETURN_POINTER(req);
-		}
 	}
 
-	PG_RETURN_POINTER(NULL);
 }
 
 /*
@@ -1346,46 +1428,30 @@ Datum
 divide_slope_support(PG_FUNCTION_ARGS)
 {
 	Node	   *rawreq = (Node *) PG_GETARG_POINTER(0);
+	Const	   *constval;
+	Node	   *arg0;
+	Node	   *arg1;
 
-	if (IsA(rawreq, SupportRequestMonotonic))
-	{
-		SupportRequestMonotonic *req = (SupportRequestMonotonic *) rawreq;
-		List	   *args;
-		Node	   *arg1;
-		Const	   *constval;
-		int			sign;
-
-		if (IsA(req->expr, FuncExpr))
-			args = ((FuncExpr *) req->expr)->args;
-		else if (IsA(req->expr, OpExpr))
-			args = ((OpExpr *) req->expr)->args;
-		else
-			PG_RETURN_POINTER(NULL);
-
-		if (list_length(args) != 2)
-			PG_RETURN_POINTER(NULL);
-
-		arg1 = (Node *) lsecond(args);
+	if (!get_2slope_args(rawreq, &arg0, &arg1))
+		PG_RETURN_POINTER(NULL);
 
-		if (!IsA(arg1, Const))
-			PG_RETURN_POINTER(NULL);
+	if (!IsA(arg1, Const))
+		PG_RETURN_POINTER(NULL);
 
-		constval = (Const *) arg1;
-		sign = get_const_sign(constval);
-		if (sign == 0)
+	constval = (Const *) arg1;
+	switch (get_const_sign(constval))
+	{
+		case NUMERIC_SIGN_POS:
+			return monotonic_slope_support(rawreq, 2, asc_slope);
+		case NUMERIC_SIGN_NEG:
+			return monotonic_slope_support(rawreq, 2, desc_slope);
+		case NUMERIC_SIGN_NAN:
+			return monotonic_slope_support(rawreq, 2, flat_slope);
+		case NUMERIC_SIGN_PINF:
+		case NUMERIC_SIGN_NINF:
+			return monotonic_slope_support(rawreq, 2, flat_slope);
+		default:
 			PG_RETURN_POINTER(NULL);
-
-		{
-			static const MonotonicFunction asc_pattern[2] = {MONOTONICFUNC_INCREASING,
-															 MONOTONICFUNC_NONE};
-			static const MonotonicFunction desc_pattern[2] = {MONOTONICFUNC_DECREASING,
-															  MONOTONICFUNC_NONE};
-
-			req->slopes = (sign > 0) ? asc_pattern : desc_pattern;
-			req->nslopes = 2;
-			PG_RETURN_POINTER(req);
-		}
 	}
-
 	PG_RETURN_POINTER(NULL);
 }
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 6a3b611528c..7de5169ac2b 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -607,7 +607,8 @@
   prorettype => 'float4', proargtypes => 'float4 float4', prosrc => 'float4mi' },
 { oid => '206',
   proname => 'float4um', prosupport => 'arg0_desc_slope_support',
-  prorettype => 'float4', proargtypes => 'float4', prosrc => 'float4um' },
+  prorettype => 'float4', proargtypes => 'float4',
+  prosrc => 'float4um' },
 { oid => '207',
   proname => 'float4abs', prorettype => 'float4', proargtypes => 'float4',
   prosrc => 'float4abs' },
@@ -648,7 +649,8 @@
   prorettype => 'float8', proargtypes => 'float8 float8', prosrc => 'float8mi' },
 { oid => '220',
   proname => 'float8um', prosupport => 'arg0_desc_slope_support',
-  prorettype => 'float8', proargtypes => 'float8', prosrc => 'float8um' },
+  prorettype => 'float8', proargtypes => 'float8',
+  prosrc => 'float8um' },
 { oid => '221',
   proname => 'float8abs', prorettype => 'float8', proargtypes => 'float8',
   prosrc => 'float8abs' },
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index 66b5cceebaa..6b345e41171 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -1819,11 +1819,14 @@ typedef struct PathKey
 	 * 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.
+	 * value (from plannodes.h), including MONOTONICFUNC_UNKNOWN until
+	 * computed, as int to avoid a header dependency.  pk_var_cannan is
+	 * true when pk_var has a type that can hold NaN (float or numeric).
 	 */
 	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);
+	bool		pk_var_cannan pg_node_attr(read_write_ignore, equal_ignore);
 } PathKey;
 
 /*
diff --git a/src/include/nodes/plannodes.h b/src/include/nodes/plannodes.h
index c9e374f92bf..9c29a6ebc4b 100644
--- a/src/include/nodes/plannodes.h
+++ b/src/include/nodes/plannodes.h
@@ -1845,6 +1845,12 @@ typedef enum MonotonicFunction
 	MONOTONICFUNC_INCREASING = (1 << 0),
 	MONOTONICFUNC_DECREASING = (1 << 1),
 	MONOTONICFUNC_BOTH = MONOTONICFUNC_INCREASING | MONOTONICFUNC_DECREASING,
+
+	/*
+	 * Planner-internal sentinel: pk_slope has not yet been computed.
+	 * Not returned by prosupport functions.
+	 */
+	MONOTONICFUNC_UNKNOWN = (1 << 2),
 } MonotonicFunction;
 
 /*
diff --git a/src/test/regress/expected/slope.out b/src/test/regress/expected/slope.out
index 74d7f82b94a..374f1612f7d 100644
--- a/src/test/regress/expected/slope.out
+++ b/src/test/regress/expected/slope.out
@@ -301,7 +301,7 @@ select floor(v_float8 + 1), count(*) from slope_src group by 1;
 -- direction and nulls agree (Forward) or both are flipped (Backward).
 -- When only one differs, a Sort is required.
 --
-CREATE TABLE slope_nulls_tmp (v float8);
+CREATE TABLE slope_nulls_tmp (v int4);
 INSERT INTO slope_nulls_tmp VALUES (1), (NULL), (2);
 ANALYZE slope_nulls_tmp;
 CREATE TEMPORARY TABLE slope_nulls_results (
@@ -346,7 +346,9 @@ BEGIN
             raise exception 'r1 <> r2';
         end if;
         node_type := plan_json->0->'Plan'->>'Node Type';
-        INSERT INTO slope_nulls_results (sign, index_order, query_order, scan_method, example) VALUES (
+        INSERT INTO slope_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,
@@ -522,3 +524,151 @@ from slope_src;
 
 -- Cleanup
 RESET enable_hashagg;
+--
+-- Simple domain test
+--
+CREATE DOMAIN df AS float8;
+CREATE TABLE t (id int, x df);
+INSERT INTO t VALUES (1,'-inf'),(2,'-1'),(3,'0'),(4,'3'),(5,'inf'),(6,'nan');
+CREATE INDEX ON t (x);
+SET enable_seqscan=off; SET enable_sort=off; SET enable_bitmapscan=off;
+SELECT id, -x AS f FROM t  ORDER BY (-x) ASC NULLS FIRST;
+ id |     f     
+----+-----------
+  5 | -Infinity
+  4 |        -3
+  3 |        -0
+  2 |         1
+  1 |  Infinity
+  6 |       NaN
+(6 rows)
+
+DROP TABLE t;
+DROP DOMAIN df;
+RESET enable_seqscan; RESET enable_sort; RESET enable_bitmapscan;
+--
+-- Test special numeric values
+--
+-- This checks the query result involving numeric columns 
+-- with 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 slope_numeric_corners (i serial, x float8);
+INSERT INTO slope_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 slope_numeric_corners_xfp8_idx_nulllast ON slope_numeric_corners ((x::float8) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xfp8_idx_nullfirst ON slope_numeric_corners ((x::float8) ASC NULLS FIRST);
+CREATE INDEX slope_numeric_corners_xfp4_idx_nulllast ON slope_numeric_corners ((x::float4) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xfp4_idx_nullfirst ON slope_numeric_corners ((x::float4) ASC NULLS FIRST);
+CREATE INDEX slope_numeric_corners_xnum_idx_nulllast ON slope_numeric_corners ((x::numeric) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xnum_idx_nullfirst ON slope_numeric_corners ((x::numeric) ASC NULLS FIRST);
+CREATE INDEX slope_numeric_corners_xnon42_idx_nulllast ON slope_numeric_corners ((x::non42) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xnon42_idx_nullfirst ON slope_numeric_corners ((x::non42) ASC NULLS FIRST);
+CREATE INDEX slope_numeric_corners_xfpreal_idx_nulllast ON slope_numeric_corners ((x::fp_real) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xfpreal_idx_nullfirst ON slope_numeric_corners ((x::fp_real) ASC NULLS FIRST);
+CREATE TEMPORARY TABLE slope_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
+
+    SET enable_bitmapscan = off;
+    SET enable_indexscan = off;
+    SET enable_indexonlyscan = off;
+    SET enable_seqscan = off;
+    SET enable_sort = off;
+    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 slope_numeric_corners '
+                 || 'ORDER BY f ' || r.sort_order || ' NULLS ' || r.nulls_order;
+        nan_query := 'SELECT array_agg(x) FROM slope_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_seqscan = on;
+        SET enable_indexscan = off;
+        SET enable_sort = 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;
+        -- slope optimization enabled
+        SET enable_seqscan = off;
+        SET enable_indexscan = on;
+        SET enable_sort = off;
+        EXECUTE agg_query into result;
+        EXECUTE plan_query into plan2_json;
+        INSERT INTO slope_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 slope_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 slope_numeric_corners_results
+GROUP BY 1;
+ passed | count 
+--------+-------
+ t      |   480
+(1 row)
+
+DROP TABLE slope_numeric_corners;
+DROP TABLE slope_numeric_corners_results;
+RESET enable_bitmapscan;
+RESET enable_indexscan;
+RESET enable_indexonlyscan;
+RESET enable_seqscan;
+RESET enable_sort;
diff --git a/src/test/regress/sql/slope.sql b/src/test/regress/sql/slope.sql
index 31eba90695f..048472c5959 100644
--- a/src/test/regress/sql/slope.sql
+++ b/src/test/regress/sql/slope.sql
@@ -145,7 +145,7 @@ select floor(v_float8 + 1), count(*) from slope_src group by 1;
 -- direction and nulls agree (Forward) or both are flipped (Backward).
 -- When only one differs, a Sort is required.
 --
-CREATE TABLE slope_nulls_tmp (v float8);
+CREATE TABLE slope_nulls_tmp (v int4);
 INSERT INTO slope_nulls_tmp VALUES (1), (NULL), (2);
 ANALYZE slope_nulls_tmp;
 
@@ -193,7 +193,9 @@ BEGIN
             raise exception 'r1 <> r2';
         end if;
         node_type := plan_json->0->'Plan'->>'Node Type';
-        INSERT INTO slope_nulls_results (sign, index_order, query_order, scan_method, example) VALUES (
+        INSERT INTO slope_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,
@@ -276,3 +278,141 @@ from slope_src;
 
 -- Cleanup
 RESET enable_hashagg;
+
+--
+-- Simple domain test
+--
+CREATE DOMAIN df AS float8;
+CREATE TABLE t (id int, x df);
+INSERT INTO t VALUES (1,'-inf'),(2,'-1'),(3,'0'),(4,'3'),(5,'inf'),(6,'nan');
+CREATE INDEX ON t (x);
+SET enable_seqscan=off; SET enable_sort=off; SET enable_bitmapscan=off;
+SELECT id, -x AS f FROM t  ORDER BY (-x) ASC NULLS FIRST;
+
+DROP TABLE t;
+DROP DOMAIN df;
+RESET enable_seqscan; RESET enable_sort; RESET enable_bitmapscan;
+
+--
+-- Test special numeric values
+--
+-- This checks the query result involving numeric columns 
+-- with 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 slope_numeric_corners (i serial, x float8);
+INSERT INTO slope_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 slope_numeric_corners_xfp8_idx_nulllast ON slope_numeric_corners ((x::float8) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xfp8_idx_nullfirst ON slope_numeric_corners ((x::float8) ASC NULLS FIRST);
+CREATE INDEX slope_numeric_corners_xfp4_idx_nulllast ON slope_numeric_corners ((x::float4) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xfp4_idx_nullfirst ON slope_numeric_corners ((x::float4) ASC NULLS FIRST);
+CREATE INDEX slope_numeric_corners_xnum_idx_nulllast ON slope_numeric_corners ((x::numeric) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xnum_idx_nullfirst ON slope_numeric_corners ((x::numeric) ASC NULLS FIRST);
+CREATE INDEX slope_numeric_corners_xnon42_idx_nulllast ON slope_numeric_corners ((x::non42) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xnon42_idx_nullfirst ON slope_numeric_corners ((x::non42) ASC NULLS FIRST);
+CREATE INDEX slope_numeric_corners_xfpreal_idx_nulllast ON slope_numeric_corners ((x::fp_real) ASC NULLS LAST);
+CREATE INDEX slope_numeric_corners_xfpreal_idx_nullfirst ON slope_numeric_corners ((x::fp_real) ASC NULLS FIRST);
+
+CREATE TEMPORARY TABLE slope_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
+
+    SET enable_bitmapscan = off;
+    SET enable_indexscan = off;
+    SET enable_indexonlyscan = off;
+    SET enable_seqscan = off;
+    SET enable_sort = off;
+    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 slope_numeric_corners '
+                 || 'ORDER BY f ' || r.sort_order || ' NULLS ' || r.nulls_order;
+        nan_query := 'SELECT array_agg(x) FROM slope_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_seqscan = on;
+        SET enable_indexscan = off;
+        SET enable_sort = 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;
+        -- slope optimization enabled
+        SET enable_seqscan = off;
+        SET enable_indexscan = on;
+        SET enable_sort = off;
+        EXECUTE agg_query into result;
+        EXECUTE plan_query into plan2_json;
+        INSERT INTO slope_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 slope_numeric_corners_results
+WHERE expected != result;
+
+-- check the number of test cases
+SELECT expected = result as passed, count(1) 
+FROM slope_numeric_corners_results
+GROUP BY 1;
+
+DROP TABLE slope_numeric_corners;
+DROP TABLE slope_numeric_corners_results;
+RESET enable_bitmapscan;
+RESET enable_indexscan;
+RESET enable_indexonlyscan;
+RESET enable_seqscan;
+RESET enable_sort;
\ No newline at end of file
-- 
2.53.0