Add range_minus_multi and multirange_minus_multi functions

The existing range_minus function raises an exception when the range is
"split", because then the result can't be represented by a single range.
For example '[0,10)'::int4range - '[4,5)' would be '[0,4)' and '[5,10)'.

This commit adds new set-returning functions so that callers can get
results even in the case of splits. There is no risk of an exception for
multiranges, but a set-returning function lets us handle them the same
way we handle ranges.

Both functions return zero results if the subtraction would give an
empty range/multirange.

The main use-case for these functions is to implement UPDATE/DELETE FOR
PORTION OF, which must compute the application-time of "temporal
leftovers": the part of history in an updated/deleted row that was not
changed. To preserve the untouched history, we will implicitly insert
one record for each result returned by range/multirange_minus_multi.
Using a set-returning function will also let us support user-defined
types for application-time update/delete in the future.

Author: Paul A. Jungwirth <[email protected]>

Author: pjungwir

doc/src/sgml/func/func-range.sgml

@@ -842,6 +842,29 @@
         <returnvalue>[1,4)</returnvalue>
        </para></entry>
       </row>
+
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+         <primary>range_minus_multi</primary>
+        </indexterm>
+        <function>range_minus_multi</function> ( <type>anyrange</type>, <type>anyrange</type> )
+        <returnvalue>setof anyrange</returnvalue>
+       </para>
+       <para>
+        Returns the non-empty range(s) remaining after subtracting the second range from the first.
+        One row is returned for each range, so if the second range splits the first into two parts,
+        there will be two results. If the subtraction yields an empty range, no rows are returned.
+       </para>
+       <para>
+        <literal>range_minus_multi('[0,10)'::int4range, '[3,4)'::int4range)</literal>
+        <returnvalue></returnvalue>
+<programlisting>
+ [0,3)
+ [4,10)
+</programlisting>
+       </para></entry>
+      </row>
      </tbody>
     </tgroup>
    </table>
@@ -1041,6 +1064,25 @@
 </programlisting>
        </para></entry>
       </row>
+
+      <row>
+       <entry role="func_table_entry"><para role="func_signature">
+        <indexterm>
+         <primary>multirange_minus_multi</primary>
+        </indexterm>
+        <function>multirange_minus_multi</function> ( <type>anymultirange</type>, <type>anymultirange</type> )
+        <returnvalue>setof anymultirange</returnvalue>
+       </para>
+       <para>
+        Returns the non-empty multirange(s) remaining after subtracting the second multirange from the first.
+        If the subtraction yields an empty multirange, no rows are returned.
+        Two rows are never returned, because a single multirange can always accommodate any result.
+       </para>
+       <para>
+        <literal>range_minus_multi('[0,10)'::int4range, '[3,4)'::int4range)</literal>
+        <returnvalue>{[0,3), [4,10)}</returnvalue>
+       </para></entry>
+      </row>
      </tbody>
     </tgroup>
    </table>

src/backend/utils/adt/multirangetypes.c

@@ -1226,6 +1226,77 @@ multirange_minus_internal(Oid mltrngtypoid, TypeCacheEntry *rangetyp,
 	return make_multirange(mltrngtypoid, rangetyp, range_count3, ranges3);
 }
 
+/*
+ * multirange_minus_multi - like multirange_minus but returning the result as a SRF,
+ * with no rows if the result would be empty.
+ */
+Datum
+multirange_minus_multi(PG_FUNCTION_ARGS)
+{
+	FuncCallContext *funcctx;
+	MemoryContext oldcontext;
+
+	if (!SRF_IS_FIRSTCALL())
+	{
+		/* We never have more than one result */
+		funcctx = SRF_PERCALL_SETUP();
+		SRF_RETURN_DONE(funcctx);
+	}
+	else
+	{
+		MultirangeType *mr1;
+		MultirangeType *mr2;
+		Oid			mltrngtypoid;
+		TypeCacheEntry *typcache;
+		TypeCacheEntry *rangetyp;
+		int32		range_count1;
+		int32		range_count2;
+		RangeType **ranges1;
+		RangeType **ranges2;
+		MultirangeType *mr;
+
+		funcctx = SRF_FIRSTCALL_INIT();
+
+		/*
+		 * switch to memory context appropriate for multiple function calls
+		 */
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		/* get args, detoasting into multi-call memory context */
+		mr1 = PG_GETARG_MULTIRANGE_P(0);
+		mr2 = PG_GETARG_MULTIRANGE_P(1);
+
+		mltrngtypoid = MultirangeTypeGetOid(mr1);
+		typcache = lookup_type_cache(mltrngtypoid, TYPECACHE_MULTIRANGE_INFO);
+		if (typcache->rngtype == NULL)
+			elog(ERROR, "type %u is not a multirange type", mltrngtypoid);
+		rangetyp = typcache->rngtype;
+
+		if (MultirangeIsEmpty(mr1) || MultirangeIsEmpty(mr2))
+			mr = mr1;
+		else
+		{
+			multirange_deserialize(rangetyp, mr1, &range_count1, &ranges1);
+			multirange_deserialize(rangetyp, mr2, &range_count2, &ranges2);
+
+			mr = multirange_minus_internal(mltrngtypoid,
+										   rangetyp,
+										   range_count1,
+										   ranges1,
+										   range_count2,
+										   ranges2);
+		}
+
+		MemoryContextSwitchTo(oldcontext);
+
+		funcctx = SRF_PERCALL_SETUP();
+		if (MultirangeIsEmpty(mr))
+			SRF_RETURN_DONE(funcctx);
+		else
+			SRF_RETURN_NEXT(funcctx, MultirangeTypePGetDatum(mr));
+	}
+}
+
 /* multirange intersection */
 Datum
 multirange_intersect(PG_FUNCTION_ARGS)

src/backend/utils/adt/rangetypes.c

@@ -31,6 +31,7 @@
 #include "postgres.h"
 
 #include "common/hashfn.h"
+#include "funcapi.h"
 #include "libpq/pqformat.h"
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
@@ -39,6 +40,7 @@
 #include "optimizer/clauses.h"
 #include "optimizer/cost.h"
 #include "optimizer/optimizer.h"
+#include "utils/array.h"
 #include "utils/builtins.h"
 #include "utils/date.h"
 #include "utils/lsyscache.h"
@@ -1214,6 +1216,170 @@ range_split_internal(TypeCacheEntry *typcache, const RangeType *r1, const RangeT
 	return false;
 }
 
+/*
+ * range_minus_multi - like range_minus but as a SRF to accommodate splits,
+ * with no result rows if the result would be empty.
+ */
+Datum
+range_minus_multi(PG_FUNCTION_ARGS)
+{
+	typedef struct
+	{
+		RangeType  *rs[2];
+		int			n;
+	}			range_minus_multi_fctx;
+
+	FuncCallContext *funcctx;
+	range_minus_multi_fctx *fctx;
+	MemoryContext oldcontext;
+
+	/* stuff done only on the first call of the function */
+	if (SRF_IS_FIRSTCALL())
+	{
+		RangeType  *r1;
+		RangeType  *r2;
+		Oid			rngtypid;
+		TypeCacheEntry *typcache;
+
+		/* create a function context for cross-call persistence */
+		funcctx = SRF_FIRSTCALL_INIT();
+
+		/*
+		 * switch to memory context appropriate for multiple function calls
+		 */
+		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+		r1 = PG_GETARG_RANGE_P(0);
+		r2 = PG_GETARG_RANGE_P(1);
+
+		/* Different types should be prevented by ANYRANGE matching rules */
+		if (RangeTypeGetOid(r1) != RangeTypeGetOid(r2))
+			elog(ERROR, "range types do not match");
+
+		/* allocate memory for user context */
+		fctx = (range_minus_multi_fctx *) palloc(sizeof(range_minus_multi_fctx));
+
+		/*
+		 * Initialize state. We can't store the range typcache in fn_extra
+		 * because the caller uses that for the SRF state.
+		 */
+		rngtypid = RangeTypeGetOid(r1);
+		typcache = lookup_type_cache(rngtypid, TYPECACHE_RANGE_INFO);
+		if (typcache->rngelemtype == NULL)
+			elog(ERROR, "type %u is not a range type", rngtypid);
+		range_minus_multi_internal(typcache, r1, r2, fctx->rs, &fctx->n);
+
+		funcctx->user_fctx = fctx;
+		MemoryContextSwitchTo(oldcontext);
+	}
+
+	/* stuff done on every call of the function */
+	funcctx = SRF_PERCALL_SETUP();
+	fctx = funcctx->user_fctx;
+
+	if (funcctx->call_cntr < fctx->n)
+	{
+		/*
+		 * We must keep these on separate lines because SRF_RETURN_NEXT does
+		 * call_cntr++:
+		 */
+		RangeType  *ret = fctx->rs[funcctx->call_cntr];
+
+		SRF_RETURN_NEXT(funcctx, RangeTypePGetDatum(ret));
+	}
+	else
+		/* do when there is no more left */
+		SRF_RETURN_DONE(funcctx);
+}
+
+/*
+ * range_minus_multi_internal - Sets outputs and outputn to the ranges
+ * remaining and their count (respectively) after subtracting r2 from r1.
+ * The array should never contain empty ranges.
+ * The outputs will be ordered. We expect that outputs is an array of
+ * RangeType pointers, already allocated with two elements.
+ */
+void
+range_minus_multi_internal(TypeCacheEntry *typcache, RangeType *r1,
+						   RangeType *r2, RangeType **outputs, int *outputn)
+{
+	int			cmp_l1l2,
+				cmp_l1u2,
+				cmp_u1l2,
+				cmp_u1u2;
+	RangeBound	lower1,
+				lower2;
+	RangeBound	upper1,
+				upper2;
+	bool		empty1,
+				empty2;
+
+	range_deserialize(typcache, r1, &lower1, &upper1, &empty1);
+	range_deserialize(typcache, r2, &lower2, &upper2, &empty2);
+
+	if (empty1)
+	{
+		/* if r1 is empty then r1 - r2 is empty, so return zero results */
+		*outputn = 0;
+		return;
+	}
+	else if (empty2)
+	{
+		/* r2 is empty so the result is just r1 (which we know is not empty) */
+		outputs[0] = r1;
+		*outputn = 1;
+		return;
+	}
+
+	/*
+	 * Use the same logic as range_minus_internal, but support the split case
+	 */
+	cmp_l1l2 = range_cmp_bounds(typcache, &lower1, &lower2);
+	cmp_l1u2 = range_cmp_bounds(typcache, &lower1, &upper2);
+	cmp_u1l2 = range_cmp_bounds(typcache, &upper1, &lower2);
+	cmp_u1u2 = range_cmp_bounds(typcache, &upper1, &upper2);
+
+	if (cmp_l1l2 < 0 && cmp_u1u2 > 0)
+	{
+		lower2.inclusive = !lower2.inclusive;
+		lower2.lower = false;	/* it will become the upper bound */
+		outputs[0] = make_range(typcache, &lower1, &lower2, false, NULL);
+
+		upper2.inclusive = !upper2.inclusive;
+		upper2.lower = true;	/* it will become the lower bound */
+		outputs[1] = make_range(typcache, &upper2, &upper1, false, NULL);
+
+		*outputn = 2;
+	}
+	else if (cmp_l1u2 > 0 || cmp_u1l2 < 0)
+	{
+		outputs[0] = r1;
+		*outputn = 1;
+	}
+	else if (cmp_l1l2 >= 0 && cmp_u1u2 <= 0)
+	{
+		*outputn = 0;
+	}
+	else if (cmp_l1l2 <= 0 && cmp_u1l2 >= 0 && cmp_u1u2 <= 0)
+	{
+		lower2.inclusive = !lower2.inclusive;
+		lower2.lower = false;	/* it will become the upper bound */
+		outputs[0] = make_range(typcache, &lower1, &lower2, false, NULL);
+		*outputn = 1;
+	}
+	else if (cmp_l1l2 >= 0 && cmp_u1u2 >= 0 && cmp_l1u2 <= 0)
+	{
+		upper2.inclusive = !upper2.inclusive;
+		upper2.lower = true;	/* it will become the lower bound */
+		outputs[0] = make_range(typcache, &upper2, &upper1, false, NULL);
+		*outputn = 1;
+	}
+	else
+	{
+		elog(ERROR, "unexpected case in range_minus_multi");
+	}
+}
+
 /* range -> range aggregate functions */
 
 Datum

src/include/catalog/pg_proc.dat

@@ -10939,6 +10939,10 @@
 { oid => '3869',
   proname => 'range_minus', prorettype => 'anyrange',
   proargtypes => 'anyrange anyrange', prosrc => 'range_minus' },
+{ oid => '8412', descr => 'remove portion from range',
+  proname => 'range_minus_multi', prorows => '2',
+  proretset => 't', prorettype => 'anyrange',
+  proargtypes => 'anyrange anyrange', prosrc => 'range_minus_multi' },
 { oid => '3870', descr => 'less-equal-greater',
   proname => 'range_cmp', prorettype => 'int4',
   proargtypes => 'anyrange anyrange', prosrc => 'range_cmp' },
@@ -11229,6 +11233,10 @@
 { oid => '4271',
   proname => 'multirange_minus', prorettype => 'anymultirange',
   proargtypes => 'anymultirange anymultirange', prosrc => 'multirange_minus' },
+{ oid => '8411', descr => 'remove portion from multirange',
+  proname => 'multirange_minus_multi', prorows => '1',
+  proretset => 't', prorettype => 'anymultirange',
+  proargtypes => 'anymultirange anymultirange', prosrc => 'multirange_minus_multi' },
 { oid => '4272',
   proname => 'multirange_intersect', prorettype => 'anymultirange',
   proargtypes => 'anymultirange anymultirange',

src/include/utils/rangetypes.h

@@ -164,5 +164,7 @@ extern RangeType *make_empty_range(TypeCacheEntry *typcache);
 extern bool range_split_internal(TypeCacheEntry *typcache, const RangeType *r1,
 								 const RangeType *r2, RangeType **output1,
 								 RangeType **output2);
+extern void range_minus_multi_internal(TypeCacheEntry *typcache, RangeType *r1,
+									   RangeType *r2, RangeType **outputs, int *outputn);
 
 #endif							/* RANGETYPES_H */