Add window RANGE support for float4, float8, numeric.

Commit 0a459ce left this for later, but since time's running out,
I went ahead and took care of it.  There are more data types that
somebody might someday want RANGE support for, but this is enough
to satisfy all expectations of the SQL standard, which just says that
"numeric, datetime, and interval" types should have RANGE support.

Author: tglsfdc

src/backend/utils/adt/float.c

@@ -1180,6 +1180,93 @@ btfloat84cmp(PG_FUNCTION_ARGS)
 	PG_RETURN_INT32(float8_cmp_internal(arg1, arg2));
 }
 
+/*
+ * in_range support function for float8.
+ *
+ * Note: we needn't supply a float8_float4 variant, as implicit coercion
+ * of the offset value takes care of that scenario just as well.
+ */
+Datum
+in_range_float8_float8(PG_FUNCTION_ARGS)
+{
+	float8		val = PG_GETARG_FLOAT8(0);
+	float8		base = PG_GETARG_FLOAT8(1);
+	float8		offset = PG_GETARG_FLOAT8(2);
+	bool		sub = PG_GETARG_BOOL(3);
+	bool		less = PG_GETARG_BOOL(4);
+	float8		sum;
+
+	/*
+	 * Reject negative or NaN offset.  Negative is per spec, and NaN is
+	 * because appropriate semantics for that seem non-obvious.
+	 */
+	if (isnan(offset) || offset < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PRECEDING_FOLLOWING_SIZE),
+				 errmsg("invalid preceding or following size in window function")));
+
+	/*
+	 * Deal with cases where val and/or base is NaN, following the rule that
+	 * NaN sorts after non-NaN (cf float8_cmp_internal).  The offset cannot
+	 * affect the conclusion.
+	 */
+	if (isnan(val))
+	{
+		if (isnan(base))
+			PG_RETURN_BOOL(true);	/* NAN = NAN */
+		else
+			PG_RETURN_BOOL(!less);	/* NAN > non-NAN */
+	}
+	else if (isnan(base))
+	{
+		PG_RETURN_BOOL(less);	/* non-NAN < NAN */
+	}
+
+	/*
+	 * Deal with infinite offset (necessarily +inf, at this point).  We must
+	 * special-case this because if base happens to be -inf, their sum would
+	 * be NaN, which is an overflow-ish condition we should avoid.
+	 */
+	if (isinf(offset))
+	{
+		PG_RETURN_BOOL(sub ? !less : less);
+	}
+
+	/*
+	 * Otherwise it should be safe to compute base +/- offset.  We trust the
+	 * FPU to cope if base is +/-inf or the true sum would overflow, and
+	 * produce a suitably signed infinity, which will compare properly against
+	 * val whether or not that's infinity.
+	 */
+	if (sub)
+		sum = base - offset;
+	else
+		sum = base + offset;
+
+	if (less)
+		PG_RETURN_BOOL(val <= sum);
+	else
+		PG_RETURN_BOOL(val >= sum);
+}
+
+/*
+ * in_range support function for float4.
+ *
+ * We would need a float4_float8 variant in any case, so we supply that and
+ * let implicit coercion take care of the float4_float4 case.
+ */
+Datum
+in_range_float4_float8(PG_FUNCTION_ARGS)
+{
+	/* Doesn't seem worth duplicating code for, so just invoke float8_float8 */
+	return DirectFunctionCall5(in_range_float8_float8,
+							   Float8GetDatumFast((float8) PG_GETARG_FLOAT4(0)),
+							   Float8GetDatumFast((float8) PG_GETARG_FLOAT4(1)),
+							   PG_GETARG_DATUM(2),
+							   PG_GETARG_DATUM(3),
+							   PG_GETARG_DATUM(4));
+}
+
 
 /*
  *		===================

src/backend/utils/adt/numeric.c

@@ -2165,6 +2165,81 @@ cmp_numerics(Numeric num1, Numeric num2)
 	return result;
 }
 
+/*
+ * in_range support function for numeric.
+ */
+Datum
+in_range_numeric_numeric(PG_FUNCTION_ARGS)
+{
+	Numeric		val = PG_GETARG_NUMERIC(0);
+	Numeric		base = PG_GETARG_NUMERIC(1);
+	Numeric		offset = PG_GETARG_NUMERIC(2);
+	bool		sub = PG_GETARG_BOOL(3);
+	bool		less = PG_GETARG_BOOL(4);
+	bool		result;
+
+	/*
+	 * Reject negative or NaN offset.  Negative is per spec, and NaN is
+	 * because appropriate semantics for that seem non-obvious.
+	 */
+	if (NUMERIC_IS_NAN(offset) || NUMERIC_SIGN(offset) == NUMERIC_NEG)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PRECEDING_FOLLOWING_SIZE),
+				 errmsg("invalid preceding or following size in window function")));
+
+	/*
+	 * Deal with cases where val and/or base is NaN, following the rule that
+	 * NaN sorts after non-NaN (cf cmp_numerics).  The offset cannot affect
+	 * the conclusion.
+	 */
+	if (NUMERIC_IS_NAN(val))
+	{
+		if (NUMERIC_IS_NAN(base))
+			result = true;		/* NAN = NAN */
+		else
+			result = !less;		/* NAN > non-NAN */
+	}
+	else if (NUMERIC_IS_NAN(base))
+	{
+		result = less;			/* non-NAN < NAN */
+	}
+	else
+	{
+		/*
+		 * Otherwise go ahead and compute base +/- offset.  While it's
+		 * possible for this to overflow the numeric format, it's unlikely
+		 * enough that we don't take measures to prevent it.
+		 */
+		NumericVar	valv;
+		NumericVar	basev;
+		NumericVar	offsetv;
+		NumericVar	sum;
+
+		init_var_from_num(val, &valv);
+		init_var_from_num(base, &basev);
+		init_var_from_num(offset, &offsetv);
+		init_var(&sum);
+
+		if (sub)
+			sub_var(&basev, &offsetv, &sum);
+		else
+			add_var(&basev, &offsetv, &sum);
+
+		if (less)
+			result = (cmp_var(&valv, &sum) <= 0);
+		else
+			result = (cmp_var(&valv, &sum) >= 0);
+
+		free_var(&sum);
+	}
+
+	PG_FREE_IF_COPY(val, 0);
+	PG_FREE_IF_COPY(base, 1);
+	PG_FREE_IF_COPY(offset, 2);
+
+	PG_RETURN_BOOL(result);
+}
+
 Datum
 hash_numeric(PG_FUNCTION_ARGS)
 {

src/include/catalog/catversion.h

@@ -53,6 +53,6 @@
  */
 
 /*							yyyymmddN */
-#define CATALOG_VERSION_NO	201802061
+#define CATALOG_VERSION_NO	201802241
 
 #endif

src/include/catalog/pg_amproc.h

@@ -105,6 +105,8 @@ DATA(insert (	1970   700 701 1 2194 ));
 DATA(insert (	1970   701 701 1 355 ));
 DATA(insert (	1970   701 701 2 3133 ));
 DATA(insert (	1970   701 700 1 2195 ));
+DATA(insert (	1970   701 701 3 4139 ));
+DATA(insert (	1970   700 701 3 4140 ));
 DATA(insert (	1974   869 869 1 926 ));
 DATA(insert (	1976   21 21 1 350 ));
 DATA(insert (	1976   21 21 2 3129 ));
@@ -133,6 +135,7 @@ DATA(insert (	1986   19 19 1 359 ));
 DATA(insert (	1986   19 19 2 3135 ));
 DATA(insert (	1988   1700 1700 1 1769 ));
 DATA(insert (	1988   1700 1700 2 3283 ));
+DATA(insert (	1988   1700 1700 3 4141 ));
 DATA(insert (	1989   26 26 1 356 ));
 DATA(insert (	1989   26 26 2 3134 ));
 DATA(insert (	1991   30 30 1 404 ));

src/include/catalog/pg_proc.h

@@ -661,6 +661,12 @@ DATA(insert OID = 4131 ( in_range		   PGNSP PGUID 12 1 0 0 0 f f f f t f i s 5 0
 DESCR("window RANGE support");
 DATA(insert OID = 4132 ( in_range		   PGNSP PGUID 12 1 0 0 0 f f f f t f i s 5 0 16 "21 21 21 16 16" _null_ _null_ _null_ _null_ _null_ in_range_int2_int2 _null_ _null_ _null_ ));
 DESCR("window RANGE support");
+DATA(insert OID = 4139 ( in_range		   PGNSP PGUID 12 1 0 0 0 f f f f t f i s 5 0 16 "701 701 701 16 16" _null_ _null_ _null_ _null_ _null_ in_range_float8_float8 _null_ _null_ _null_ ));
+DESCR("window RANGE support");
+DATA(insert OID = 4140 ( in_range		   PGNSP PGUID 12 1 0 0 0 f f f f t f i s 5 0 16 "700 700 701 16 16" _null_ _null_ _null_ _null_ _null_ in_range_float4_float8 _null_ _null_ _null_ ));
+DESCR("window RANGE support");
+DATA(insert OID = 4141 ( in_range		   PGNSP PGUID 12 1 0 0 0 f f f f t f i s 5 0 16 "1700 1700 1700 16 16" _null_ _null_ _null_ _null_ _null_ in_range_numeric_numeric _null_ _null_ _null_ ));
+DESCR("window RANGE support");
 
 DATA(insert OID = 361 (  lseg_distance	   PGNSP PGUID 12 1 0 0 0 f f f f t f i s 2 0 701 "601 601" _null_ _null_ _null_ _null_ _null_	lseg_distance _null_ _null_ _null_ ));
 DATA(insert OID = 362 (  lseg_interpt	   PGNSP PGUID 12 1 0 0 0 f f f f t f i s 2 0 600 "601 601" _null_ _null_ _null_ _null_ _null_	lseg_interpt _null_ _null_ _null_ ));