Support subscripting of arbitrary types, not only arrays.

This patch generalizes the subscripting infrastructure so that any
data type can be subscripted, if it provides a handler function to
define what that means.  Traditional variable-length (varlena) arrays
all use array_subscript_handler(), while the existing fixed-length
types that support subscripting use raw_array_subscript_handler().
It's expected that other types that want to use subscripting notation
will define their own handlers.  (This patch provides no such new
features, though; it only lays the foundation for them.)

To do this, move the parser's semantic processing of subscripts
(including coercion to whatever data type is required) into a
method callback supplied by the handler.  On the execution side,
replace the ExecEvalSubscriptingRef* layer of functions with direct
calls to callback-supplied execution routines.  (Thus, essentially
no new run-time overhead should be caused by this patch.  Indeed,
there is room to remove some overhead by supplying specialized
execution routines.  This patch does a little bit in that line,
but more could be done.)

Additional work is required here and there to remove formerly
hard-wired assumptions about the result type, collation, etc
of a SubscriptingRef expression node; and to remove assumptions
that the subscript values must be integers.

One useful side-effect of this is that we now have a less squishy
mechanism for identifying whether a data type is a "true" array:
instead of wiring in weird rules about typlen, we can look to see
if pg_type.typsubscript == F_ARRAY_SUBSCRIPT_HANDLER.  For this
to be bulletproof, we have to forbid user-defined types from using
that handler directly; but there seems no good reason for them to
do so.

This patch also removes assumptions that the number of subscripts
is limited to MAXDIM (6), or indeed has any hard-wired limit.
That limit still applies to types handled by array_subscript_handler
or raw_array_subscript_handler, but to discourage other dependencies
on this constant, I've moved it from c.h to utils/array.h.

Dmitry Dolgov, reviewed at various times by Tom Lane, Arthur Zakirov,
Peter Eisentraut, Pavel Stehule

Discussion: https://postgr.es/m/CA+q6zcVDuGBv=M0FqBYX8DPebS3F_0KQ6OVFobGJPM507_SZ_w@mail.gmail.com
Discussion: https://postgr.es/m/CA+q6zcVovR+XY4mfk-7oNk-rF91gH0PebnNfuUjuuDsyHjOcVA@mail.gmail.com

2 files changed, 93 insertions, 295 deletions

diff --git a/src/backend/executor/execExpr.c b/src/backend/executor/execExpr.c
index 79b325c7cfb..0134ecc261e 100644
--- a/src/backend/executor/execExpr.c
+++ b/src/backend/executor/execExpr.c
@@ -40,6 +40,7 @@
 #include "miscadmin.h"
 #include "nodes/makefuncs.h"
 #include "nodes/nodeFuncs.h"
+#include "nodes/subscripting.h"
 #include "optimizer/optimizer.h"
 #include "pgstat.h"
 #include "utils/acl.h"
@@ -2523,19 +2524,51 @@ ExecInitSubscriptingRef(ExprEvalStep *scratch, SubscriptingRef *sbsref,
 						ExprState *state, Datum *resv, bool *resnull)
 {
 	bool		isAssignment = (sbsref->refassgnexpr != NULL);
-	SubscriptingRefState *sbsrefstate = palloc0(sizeof(SubscriptingRefState));
+	int			nupper = list_length(sbsref->refupperindexpr);
+	int			nlower = list_length(sbsref->reflowerindexpr);
+	const SubscriptRoutines *sbsroutines;
+	SubscriptingRefState *sbsrefstate;
+	SubscriptExecSteps methods;
+	char	   *ptr;
 	List	   *adjust_jumps = NIL;
 	ListCell   *lc;
 	int			i;
 
+	/* Look up the subscripting support methods */
+	sbsroutines = getSubscriptingRoutines(sbsref->refcontainertype, NULL);
+
+	/* Allocate sbsrefstate, with enough space for per-subscript arrays too */
+	sbsrefstate = palloc0(MAXALIGN(sizeof(SubscriptingRefState)) +
+						  (nupper + nlower) * (sizeof(Datum) +
+											   2 * sizeof(bool)));
+
 	/* Fill constant fields of SubscriptingRefState */
 	sbsrefstate->isassignment = isAssignment;
-	sbsrefstate->refelemtype = sbsref->refelemtype;
-	sbsrefstate->refattrlength = get_typlen(sbsref->refcontainertype);
-	get_typlenbyvalalign(sbsref->refelemtype,
-						 &sbsrefstate->refelemlength,
-						 &sbsrefstate->refelembyval,
-						 &sbsrefstate->refelemalign);
+	sbsrefstate->numupper = nupper;
+	sbsrefstate->numlower = nlower;
+	/* Set up per-subscript arrays */
+	ptr = ((char *) sbsrefstate) + MAXALIGN(sizeof(SubscriptingRefState));
+	sbsrefstate->upperindex = (Datum *) ptr;
+	ptr += nupper * sizeof(Datum);
+	sbsrefstate->lowerindex = (Datum *) ptr;
+	ptr += nlower * sizeof(Datum);
+	sbsrefstate->upperprovided = (bool *) ptr;
+	ptr += nupper * sizeof(bool);
+	sbsrefstate->lowerprovided = (bool *) ptr;
+	ptr += nlower * sizeof(bool);
+	sbsrefstate->upperindexnull = (bool *) ptr;
+	ptr += nupper * sizeof(bool);
+	sbsrefstate->lowerindexnull = (bool *) ptr;
+	/* ptr += nlower * sizeof(bool); */
+
+	/*
+	 * Let the container-type-specific code have a chance.  It must fill the
+	 * "methods" struct with function pointers for us to possibly use in
+	 * execution steps below; and it can optionally set up some data pointed
+	 * to by the workspace field.
+	 */
+	memset(&methods, 0, sizeof(methods));
+	sbsroutines->exec_setup(sbsref, sbsrefstate, &methods);
 
 	/*
 	 * Evaluate array input.  It's safe to do so into resv/resnull, because we
@@ -2546,11 +2579,11 @@ ExecInitSubscriptingRef(ExprEvalStep *scratch, SubscriptingRef *sbsref,
 	ExecInitExprRec(sbsref->refexpr, state, resv, resnull);
 
 	/*
-	 * If refexpr yields NULL, and it's a fetch, then result is NULL.  We can
-	 * implement this with just JUMP_IF_NULL, since we evaluated the array
-	 * into the desired target location.
+	 * If refexpr yields NULL, and the operation should be strict, then result
+	 * is NULL.  We can implement this with just JUMP_IF_NULL, since we
+	 * evaluated the array into the desired target location.
 	 */
-	if (!isAssignment)
+	if (!isAssignment && sbsroutines->fetch_strict)
 	{
 		scratch->opcode = EEOP_JUMP_IF_NULL;
 		scratch->d.jump.jumpdone = -1;	/* adjust later */
@@ -2559,19 +2592,6 @@ ExecInitSubscriptingRef(ExprEvalStep *scratch, SubscriptingRef *sbsref,
 								   state->steps_len - 1);
 	}
 
-	/* Verify subscript list lengths are within limit */
-	if (list_length(sbsref->refupperindexpr) > MAXDIM)
-		ereport(ERROR,
-				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
-				 errmsg("number of array dimensions (%d) exceeds the maximum allowed (%d)",
-						list_length(sbsref->refupperindexpr), MAXDIM)));
-
-	if (list_length(sbsref->reflowerindexpr) > MAXDIM)
-		ereport(ERROR,
-				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
-				 errmsg("number of array dimensions (%d) exceeds the maximum allowed (%d)",
-						list_length(sbsref->reflowerindexpr), MAXDIM)));
-
 	/* Evaluate upper subscripts */
 	i = 0;
 	foreach(lc, sbsref->refupperindexpr)
@@ -2582,28 +2602,18 @@ ExecInitSubscriptingRef(ExprEvalStep *scratch, SubscriptingRef *sbsref,
 		if (!e)
 		{
 			sbsrefstate->upperprovided[i] = false;
-			i++;
-			continue;
+			sbsrefstate->upperindexnull[i] = true;
+		}
+		else
+		{
+			sbsrefstate->upperprovided[i] = true;
+			/* Each subscript is evaluated into appropriate array entry */
+			ExecInitExprRec(e, state,
+							&sbsrefstate->upperindex[i],
+							&sbsrefstate->upperindexnull[i]);
 		}
-
-		sbsrefstate->upperprovided[i] = true;
-
-		/* Each subscript is evaluated into subscriptvalue/subscriptnull */
-		ExecInitExprRec(e, state,
-						&sbsrefstate->subscriptvalue, &sbsrefstate->subscriptnull);
-
-		/* ... and then SBSREF_SUBSCRIPT saves it into step's workspace */
-		scratch->opcode = EEOP_SBSREF_SUBSCRIPT;
-		scratch->d.sbsref_subscript.state = sbsrefstate;
-		scratch->d.sbsref_subscript.off = i;
-		scratch->d.sbsref_subscript.isupper = true;
-		scratch->d.sbsref_subscript.jumpdone = -1;	/* adjust later */
-		ExprEvalPushStep(state, scratch);
-		adjust_jumps = lappend_int(adjust_jumps,
-								   state->steps_len - 1);
 		i++;
 	}
-	sbsrefstate->numupper = i;
 
 	/* Evaluate lower subscripts similarly */
 	i = 0;
@@ -2615,39 +2625,43 @@ ExecInitSubscriptingRef(ExprEvalStep *scratch, SubscriptingRef *sbsref,
 		if (!e)
 		{
 			sbsrefstate->lowerprovided[i] = false;
-			i++;
-			continue;
+			sbsrefstate->lowerindexnull[i] = true;
 		}
+		else
+		{
+			sbsrefstate->lowerprovided[i] = true;
+			/* Each subscript is evaluated into appropriate array entry */
+			ExecInitExprRec(e, state,
+							&sbsrefstate->lowerindex[i],
+							&sbsrefstate->lowerindexnull[i]);
+		}
+		i++;
+	}
 
-		sbsrefstate->lowerprovided[i] = true;
-
-		/* Each subscript is evaluated into subscriptvalue/subscriptnull */
-		ExecInitExprRec(e, state,
-						&sbsrefstate->subscriptvalue, &sbsrefstate->subscriptnull);
-
-		/* ... and then SBSREF_SUBSCRIPT saves it into step's workspace */
-		scratch->opcode = EEOP_SBSREF_SUBSCRIPT;
+	/* SBSREF_SUBSCRIPTS checks and converts all the subscripts at once */
+	if (methods.sbs_check_subscripts)
+	{
+		scratch->opcode = EEOP_SBSREF_SUBSCRIPTS;
+		scratch->d.sbsref_subscript.subscriptfunc = methods.sbs_check_subscripts;
 		scratch->d.sbsref_subscript.state = sbsrefstate;
-		scratch->d.sbsref_subscript.off = i;
-		scratch->d.sbsref_subscript.isupper = false;
 		scratch->d.sbsref_subscript.jumpdone = -1;	/* adjust later */
 		ExprEvalPushStep(state, scratch);
 		adjust_jumps = lappend_int(adjust_jumps,
 								   state->steps_len - 1);
-		i++;
 	}
-	sbsrefstate->numlower = i;
-
-	/* Should be impossible if parser is sane, but check anyway: */
-	if (sbsrefstate->numlower != 0 &&
-		sbsrefstate->numupper != sbsrefstate->numlower)
-		elog(ERROR, "upper and lower index lists are not same length");
 
 	if (isAssignment)
 	{
 		Datum	   *save_innermost_caseval;
 		bool	   *save_innermost_casenull;
 
+		/* Check for unimplemented methods */
+		if (!methods.sbs_assign)
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("type %s does not support subscripted assignment",
+							format_type_be(sbsref->refcontainertype))));
+
 		/*
 		 * We might have a nested-assignment situation, in which the
 		 * refassgnexpr is itself a FieldStore or SubscriptingRef that needs
@@ -2664,7 +2678,13 @@ ExecInitSubscriptingRef(ExprEvalStep *scratch, SubscriptingRef *sbsref,
 		 */
 		if (isAssignmentIndirectionExpr(sbsref->refassgnexpr))
 		{
+			if (!methods.sbs_fetch_old)
+				ereport(ERROR,
+						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+						 errmsg("type %s does not support subscripted assignment",
+								format_type_be(sbsref->refcontainertype))));
 			scratch->opcode = EEOP_SBSREF_OLD;
+			scratch->d.sbsref.subscriptfunc = methods.sbs_fetch_old;
 			scratch->d.sbsref.state = sbsrefstate;
 			ExprEvalPushStep(state, scratch);
 		}
@@ -2684,17 +2704,17 @@ ExecInitSubscriptingRef(ExprEvalStep *scratch, SubscriptingRef *sbsref,
 
 		/* and perform the assignment */
 		scratch->opcode = EEOP_SBSREF_ASSIGN;
+		scratch->d.sbsref.subscriptfunc = methods.sbs_assign;
 		scratch->d.sbsref.state = sbsrefstate;
 		ExprEvalPushStep(state, scratch);
-
 	}
 	else
 	{
 		/* array fetch is much simpler */
 		scratch->opcode = EEOP_SBSREF_FETCH;
+		scratch->d.sbsref.subscriptfunc = methods.sbs_fetch;
 		scratch->d.sbsref.state = sbsrefstate;
 		ExprEvalPushStep(state, scratch);
-
 	}
 
 	/* adjust jump targets */
@@ -2702,7 +2722,7 @@ ExecInitSubscriptingRef(ExprEvalStep *scratch, SubscriptingRef *sbsref,
 	{
 		ExprEvalStep *as = &state->steps[lfirst_int(lc)];
 
-		if (as->opcode == EEOP_SBSREF_SUBSCRIPT)
+		if (as->opcode == EEOP_SBSREF_SUBSCRIPTS)
 		{
 			Assert(as->d.sbsref_subscript.jumpdone == -1);
 			as->d.sbsref_subscript.jumpdone = state->steps_len;
diff --git a/src/backend/executor/execExprInterp.c b/src/backend/executor/execExprInterp.c
index c09371ad58f..6b9fc38134b 100644
--- a/src/backend/executor/execExprInterp.c
+++ b/src/backend/executor/execExprInterp.c
@@ -417,7 +417,7 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
 		&&CASE_EEOP_FIELDSELECT,
 		&&CASE_EEOP_FIELDSTORE_DEFORM,
 		&&CASE_EEOP_FIELDSTORE_FORM,
-		&&CASE_EEOP_SBSREF_SUBSCRIPT,
+		&&CASE_EEOP_SBSREF_SUBSCRIPTS,
 		&&CASE_EEOP_SBSREF_OLD,
 		&&CASE_EEOP_SBSREF_ASSIGN,
 		&&CASE_EEOP_SBSREF_FETCH,
@@ -1396,12 +1396,10 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
 			EEO_NEXT();
 		}
 
-		EEO_CASE(EEOP_SBSREF_SUBSCRIPT)
+		EEO_CASE(EEOP_SBSREF_SUBSCRIPTS)
 		{
-			/* Process an array subscript */
-
-			/* too complex for an inline implementation */
-			if (ExecEvalSubscriptingRef(state, op))
+			/* Precheck SubscriptingRef subscript(s) */
+			if (op->d.sbsref_subscript.subscriptfunc(state, op, econtext))
 			{
 				EEO_NEXT();
 			}
@@ -1413,37 +1411,11 @@ ExecInterpExpr(ExprState *state, ExprContext *econtext, bool *isnull)
 		}
 
 		EEO_CASE(EEOP_SBSREF_OLD)
+			EEO_CASE(EEOP_SBSREF_ASSIGN)
+			EEO_CASE(EEOP_SBSREF_FETCH)
 		{
-			/*
-			 * Fetch the old value in an sbsref assignment, in case it's
-			 * referenced (via a CaseTestExpr) inside the assignment
-			 * expression.
-			 */
-
-			/* too complex for an inline implementation */
-			ExecEvalSubscriptingRefOld(state, op);
-
-			EEO_NEXT();
-		}
-
-		/*
-		 * Perform SubscriptingRef assignment
-		 */
-		EEO_CASE(EEOP_SBSREF_ASSIGN)
-		{
-			/* too complex for an inline implementation */
-			ExecEvalSubscriptingRefAssign(state, op);
-
-			EEO_NEXT();
-		}
-
-		/*
-		 * Fetch subset of an array.
-		 */
-		EEO_CASE(EEOP_SBSREF_FETCH)
-		{
-			/* too complex for an inline implementation */
-			ExecEvalSubscriptingRefFetch(state, op);
+			/* Perform a SubscriptingRef fetch or assignment */
+			op->d.sbsref.subscriptfunc(state, op, econtext);
 
 			EEO_NEXT();
 		}
@@ -3123,200 +3095,6 @@ ExecEvalFieldStoreForm(ExprState *state, ExprEvalStep *op, ExprContext *econtext
 }
 
 /*
- * Process a subscript in a SubscriptingRef expression.
- *
- * If subscript is NULL, throw error in assignment case, or in fetch case
- * set result to NULL and return false (instructing caller to skip the rest
- * of the SubscriptingRef sequence).
- *
- * Subscript expression result is in subscriptvalue/subscriptnull.
- * On success, integer subscript value has been saved in upperindex[] or
- * lowerindex[] for use later.
- */
-bool
-ExecEvalSubscriptingRef(ExprState *state, ExprEvalStep *op)
-{
-	SubscriptingRefState *sbsrefstate = op->d.sbsref_subscript.state;
-	int		   *indexes;
-	int			off;
-
-	/* If any index expr yields NULL, result is NULL or error */
-	if (sbsrefstate->subscriptnull)
-	{
-		if (sbsrefstate->isassignment)
-			ereport(ERROR,
-					(errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
-					 errmsg("array subscript in assignment must not be null")));
-		*op->resnull = true;
-		return false;
-	}
-
-	/* Convert datum to int, save in appropriate place */
-	if (op->d.sbsref_subscript.isupper)
-		indexes = sbsrefstate->upperindex;
-	else
-		indexes = sbsrefstate->lowerindex;
-	off = op->d.sbsref_subscript.off;
-
-	indexes[off] = DatumGetInt32(sbsrefstate->subscriptvalue);
-
-	return true;
-}
-
-/*
- * Evaluate SubscriptingRef fetch.
- *
- * Source container is in step's result variable.
- */
-void
-ExecEvalSubscriptingRefFetch(ExprState *state, ExprEvalStep *op)
-{
-	SubscriptingRefState *sbsrefstate = op->d.sbsref.state;
-
-	/* Should not get here if source container (or any subscript) is null */
-	Assert(!(*op->resnull));
-
-	if (sbsrefstate->numlower == 0)
-	{
-		/* Scalar case */
-		*op->resvalue = array_get_element(*op->resvalue,
-										  sbsrefstate->numupper,
-										  sbsrefstate->upperindex,
-										  sbsrefstate->refattrlength,
-										  sbsrefstate->refelemlength,
-										  sbsrefstate->refelembyval,
-										  sbsrefstate->refelemalign,
-										  op->resnull);
-	}
-	else
-	{
-		/* Slice case */
-		*op->resvalue = array_get_slice(*op->resvalue,
-										sbsrefstate->numupper,
-										sbsrefstate->upperindex,
-										sbsrefstate->lowerindex,
-										sbsrefstate->upperprovided,
-										sbsrefstate->lowerprovided,
-										sbsrefstate->refattrlength,
-										sbsrefstate->refelemlength,
-										sbsrefstate->refelembyval,
-										sbsrefstate->refelemalign);
-	}
-}
-
-/*
- * Compute old container element/slice value for a SubscriptingRef assignment
- * expression. Will only be generated if the new-value subexpression
- * contains SubscriptingRef or FieldStore. The value is stored into the
- * SubscriptingRefState's prevvalue/prevnull fields.
- */
-void
-ExecEvalSubscriptingRefOld(ExprState *state, ExprEvalStep *op)
-{
-	SubscriptingRefState *sbsrefstate = op->d.sbsref.state;
-
-	if (*op->resnull)
-	{
-		/* whole array is null, so any element or slice is too */
-		sbsrefstate->prevvalue = (Datum) 0;
-		sbsrefstate->prevnull = true;
-	}
-	else if (sbsrefstate->numlower == 0)
-	{
-		/* Scalar case */
-		sbsrefstate->prevvalue = array_get_element(*op->resvalue,
-												   sbsrefstate->numupper,
-												   sbsrefstate->upperindex,
-												   sbsrefstate->refattrlength,
-												   sbsrefstate->refelemlength,
-												   sbsrefstate->refelembyval,
-												   sbsrefstate->refelemalign,
-												   &sbsrefstate->prevnull);
-	}
-	else
-	{
-		/* Slice case */
-		/* this is currently unreachable */
-		sbsrefstate->prevvalue = array_get_slice(*op->resvalue,
-												 sbsrefstate->numupper,
-												 sbsrefstate->upperindex,
-												 sbsrefstate->lowerindex,
-												 sbsrefstate->upperprovided,
-												 sbsrefstate->lowerprovided,
-												 sbsrefstate->refattrlength,
-												 sbsrefstate->refelemlength,
-												 sbsrefstate->refelembyval,
-												 sbsrefstate->refelemalign);
-		sbsrefstate->prevnull = false;
-	}
-}
-
-/*
- * Evaluate SubscriptingRef assignment.
- *
- * Input container (possibly null) is in result area, replacement value is in
- * SubscriptingRefState's replacevalue/replacenull.
- */
-void
-ExecEvalSubscriptingRefAssign(ExprState *state, ExprEvalStep *op)
-{
-	SubscriptingRefState *sbsrefstate = op->d.sbsref_subscript.state;
-
-	/*
-	 * For an assignment to a fixed-length container type, both the original
-	 * container and the value to be assigned into it must be non-NULL, else
-	 * we punt and return the original container.
-	 */
-	if (sbsrefstate->refattrlength > 0)
-	{
-		if (*op->resnull || sbsrefstate->replacenull)
-			return;
-	}
-
-	/*
-	 * For assignment to varlena arrays, we handle a NULL original array by
-	 * substituting an empty (zero-dimensional) array; insertion of the new
-	 * element will result in a singleton array value.  It does not matter
-	 * whether the new element is NULL.
-	 */
-	if (*op->resnull)
-	{
-		*op->resvalue = PointerGetDatum(construct_empty_array(sbsrefstate->refelemtype));
-		*op->resnull = false;
-	}
-
-	if (sbsrefstate->numlower == 0)
-	{
-		/* Scalar case */
-		*op->resvalue = array_set_element(*op->resvalue,
-										  sbsrefstate->numupper,
-										  sbsrefstate->upperindex,
-										  sbsrefstate->replacevalue,
-										  sbsrefstate->replacenull,
-										  sbsrefstate->refattrlength,
-										  sbsrefstate->refelemlength,
-										  sbsrefstate->refelembyval,
-										  sbsrefstate->refelemalign);
-	}
-	else
-	{
-		/* Slice case */
-		*op->resvalue = array_set_slice(*op->resvalue,
-										sbsrefstate->numupper,
-										sbsrefstate->upperindex,
-										sbsrefstate->lowerindex,
-										sbsrefstate->upperprovided,
-										sbsrefstate->lowerprovided,
-										sbsrefstate->replacevalue,
-										sbsrefstate->replacenull,
-										sbsrefstate->refattrlength,
-										sbsrefstate->refelemlength,
-										sbsrefstate->refelembyval,
-										sbsrefstate->refelemalign);
-	}
-}
-
-/*
  * Evaluate a rowtype coercion operation.
  * This may require rearranging field positions.
  *