Arrays are toastable. (At least if you initdb, which I didn't force.)

Remove a bunch of crufty code for large-object-based arrays, which is
superseded by TOAST and likely hasn't worked in a long time anyway.
Clean up array code a little, and in particular eliminate its habit
of scribbling on the input array (ie, modifying the input tuple :-().

contrib/array/array_iterator.c

@@ -47,15 +47,18 @@ array_iterator(Oid elemtype, Oid proc, int and, ArrayType *array, Datum value)
 	FmgrInfo	finfo;
 
 	/* Sanity checks */
-	if ((array == (ArrayType *) NULL)
-		|| (ARR_IS_LO(array) == true))
+	if (array == (ArrayType *) NULL)
 	{
 		/* elog(NOTICE, "array_iterator: array is null"); */
 		return (0);
 	}
+
+	/* detoast input if necessary */
+	array = DatumGetArrayTypeP(PointerGetDatum(array));
+
 	ndim = ARR_NDIM(array);
 	dim = ARR_DIMS(array);
-	nitems = getNitems(ndim, dim);
+	nitems = ArrayGetNItems(ndim, dim);
 	if (nitems == 0)
 	{
 		/* elog(NOTICE, "array_iterator: nitems = 0"); */

src/backend/commands/define.c

@@ -10,7 +10,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/commands/define.c,v 1.45 2000/07/17 03:04:44 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/commands/define.c,v 1.46 2000/07/22 03:34:26 tgl Exp $
  *
  * DESCRIPTION
  *	  The "DefineFoo" routines take the parse tree and pick out the
@@ -691,10 +691,10 @@ DefineType(char *typeName, List *parameters)
 			   "array_in",		/* receive procedure */
 			   "array_out",		/* send procedure */
 			   typeName,		/* element type name */
-			   defaultValue,	/* default type value */
+			   NULL,			/* never a default type value */
 			   false,			/* never passed by value */
-			   alignment,
-			   'p');			/* ARRAY doesn't support TOAST yet */
+			   alignment,		/* NB: must be 'i' or 'd' for arrays... */
+			   'x');			/* ARRAY is always toastable */
 
 	pfree(shadow_type);
 }

src/backend/executor/execQual.c

@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/executor/execQual.c,v 1.74 2000/07/17 03:04:51 tgl Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/executor/execQual.c,v 1.75 2000/07/22 03:34:27 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -67,9 +67,15 @@ static Datum ExecMakeFunctionResult(Node *node, List *arguments,
 /*
  *	  ExecEvalArrayRef
  *
- *	   This function takes an ArrayRef and returns a Const Node if it
- *	   is an array reference or returns the changed Array Node if it is
- *		   an array assignment.
+ *	   This function takes an ArrayRef and returns the extracted Datum
+ *	   if it's a simple reference, or the modified array value if it's
+ *	   an array assignment (read array element insertion).
+ *
+ * NOTE: we deliberately refrain from applying DatumGetArrayTypeP() here,
+ * even though that might seem natural, because this code needs to support
+ * both varlena arrays and fixed-length array types.  DatumGetArrayTypeP()
+ * only works for the varlena kind.  The routines we call in arrayfuncs.c
+ * have to know the difference (that's what they need refattrlength for).
  */
 static Datum
 ExecEvalArrayRef(ArrayRef *arrayRef,
@@ -77,7 +83,8 @@ ExecEvalArrayRef(ArrayRef *arrayRef,
 				 bool *isNull,
 				 bool *isDone)
 {
-	ArrayType  *array_scanner;
+	ArrayType  *array_source;
+	ArrayType  *resultArray;
 	List	   *elt;
 	int			i = 0,
 				j = 0;
@@ -90,7 +97,7 @@ ExecEvalArrayRef(ArrayRef *arrayRef,
 
 	if (arrayRef->refexpr != NULL)
 	{
-		array_scanner = (ArrayType *)
+		array_source = (ArrayType *)
 			DatumGetPointer(ExecEvalExpr(arrayRef->refexpr,
 										 econtext,
 										 isNull,
@@ -110,7 +117,7 @@ ExecEvalArrayRef(ArrayRef *arrayRef,
 		 * the INSERT column list. This is a kluge, but it's not real
 		 * clear what the semantics ought to be...
 		 */
-		array_scanner = NULL;
+		array_source = NULL;
 	}
 
 	foreach(elt, arrayRef->refupperindexpr)
@@ -162,43 +169,45 @@ ExecEvalArrayRef(ArrayRef *arrayRef,
 		if (*isNull)
 			return (Datum) NULL;
 
-		if (array_scanner == NULL)
+		if (array_source == NULL)
 			return sourceData;	/* XXX do something else? */
 
-		/*
-		 * XXX shouldn't we copy the array value before modifying it??
-		 *
-		 * Or perhaps these array routines should deliver a modified copy
-		 * instead of changing the source in-place.
-		 */
 		if (lIndex == NULL)
-			return PointerGetDatum(array_set(array_scanner, i,
+			resultArray = array_set(array_source, i,
 									upper.indx,
 									sourceData,
 									arrayRef->refelembyval,
 									arrayRef->refelemlength,
 									arrayRef->refattrlength,
-											 isNull));
-		return PointerGetDatum(array_assgn(array_scanner, i,
+									isNull);
+		else
+			resultArray = array_set_slice(array_source, i,
 										  upper.indx, lower.indx,
 										  (ArrayType *) DatumGetPointer(sourceData),
 										  arrayRef->refelembyval,
 										  arrayRef->refelemlength,
-										   isNull));
+										  arrayRef->refattrlength,
+										  isNull);
+		return PointerGetDatum(resultArray);
 	}
 
 	if (lIndex == NULL)
-		return array_ref(array_scanner, i,
+		return array_ref(array_source, i,
 						 upper.indx,
 						 arrayRef->refelembyval,
 						 arrayRef->refelemlength,
 						 arrayRef->refattrlength,
 						 isNull);
-	return PointerGetDatum(array_clip(array_scanner, i,
+	else
+	{
+		resultArray = array_get_slice(array_source, i,
 									  upper.indx, lower.indx,
 									  arrayRef->refelembyval,
 									  arrayRef->refelemlength,
-									  isNull));
+									  arrayRef->refattrlength,
+									  isNull);
+		return PointerGetDatum(resultArray);
+	}
 }
 
 

src/backend/utils/adt/Makefile

 #
 # Makefile for utils/adt
 #
-# $Header: /cvsroot/pgsql/src/backend/utils/adt/Makefile,v 1.39 2000/07/13 16:07:14 petere Exp $
+# $Header: /cvsroot/pgsql/src/backend/utils/adt/Makefile,v 1.40 2000/07/22 03:34:43 tgl Exp $
 #
 
 subdir = src/backend/utils/adt
@@ -15,7 +15,7 @@ CFLAGS+= -mieee
 endif
 endif
 
-OBJS = acl.o arrayfuncs.o arrayutils.o bool.o cash.o char.o chunk.o \
+OBJS = acl.o arrayfuncs.o arrayutils.o bool.o cash.o char.o \
 	date.o datetime.o datum.o filename.o float.o format_type.o \
 	geo_ops.o geo_selfuncs.o int.o int8.o like.o lztext.o \
 	misc.o nabstime.o name.o not_in.o numeric.o numutils.o \

src/backend/utils/adt/arrayutils.c

@@ -8,61 +8,62 @@
  *
  *
  * IDENTIFICATION
- *	  $Header: /cvsroot/pgsql/src/backend/utils/adt/arrayutils.c,v 1.10 2000/01/26 05:57:12 momjian Exp $
+ *	  $Header: /cvsroot/pgsql/src/backend/utils/adt/arrayutils.c,v 1.11 2000/07/22 03:34:43 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
 
-#define WEAK_C_OPTIMIZER
-
 #include "postgres.h"
+
 #include "utils/array.h"
+
+
+/* Convert subscript list into linear element number (from 0) */
 int
-GetOffset(int n, int *dim, int *lb, int *indx)
+ArrayGetOffset(int n, int *dim, int *lb, int *indx)
 {
 	int			i,
-				scale,
-				offset;
+				scale = 1,
+				offset = 0;
 
-	for (i = n - 1, scale = 1, offset = 0; i >= 0; scale *= dim[i--])
+	for (i = n - 1; i >= 0; i--)
+	{
 		offset += (indx[i] - lb[i]) * scale;
+		scale *= dim[i];
+	}
 	return offset;
 }
 
+/* Same, but subscripts are assumed 0-based, and use a scale array
+ * instead of raw dimension data (see mda_get_prod to create scale array)
+ */
 int
-getNitems(int n, int *a)
+ArrayGetOffset0(int n, int *tup, int *scale)
 {
 	int			i,
-				ret;
+				lin = 0;
 
-	for (i = 0, ret = 1; i < n; ret *= a[i++]);
-	if (n == 0)
-		ret = 0;
-	return ret;
+	for (i = 0; i < n; i++)
+		lin += tup[i] * scale[i];
+	return lin;
 }
 
+/* Convert array dimensions into number of elements */
 int
-compute_size(int *st, int *endp, int n, int base)
+ArrayGetNItems(int n, int *a)
 {
 	int			i,
 				ret;
 
-	for (i = 0, ret = base; i < n; i++)
-		ret *= (endp[i] - st[i] + 1);
+	if (n <= 0)
+		return 0;
+	ret = 1;
+	for (i = 0; i < n; i++)
+		ret *= a[i];
 	return ret;
 }
 
-void
-mda_get_offset_values(int n, int *dist, int *PC, int *span)
-{
-	int			i,
-				j;
-
-	for (j = n - 2, dist[n - 1] = 0; j >= 0; j--)
-		for (i = j + 1, dist[j] = PC[j] - 1; i < n;
-			 dist[j] -= (span[i] - 1) * PC[i], i++);
-}
-
+/* Compute ranges (sub-array dimensions) for an array slice */
 void
 mda_get_range(int n, int *span, int *st, int *endp)
 {
@@ -72,56 +73,59 @@ mda_get_range(int n, int *span, int *st, int *endp)
 		span[i] = endp[i] - st[i] + 1;
 }
 
+/* Compute products of array dimensions, ie, scale factors for subscripts */
 void
-mda_get_prod(int n, int *range, int *P)
+mda_get_prod(int n, int *range, int *prod)
 {
 	int			i;
 
-	for (i = n - 2, P[n - 1] = 1; i >= 0; i--)
-		P[i] = P[i + 1] * range[i + 1];
-}
-
-int
-tuple2linear(int n, int *tup, int *scale)
-{
-	int			i,
-				lin;
-
-	for (i = lin = 0; i < n; i++)
-		lin += tup[i] * scale[i];
-	return lin;
+	prod[n - 1] = 1;
+	for (i = n - 2; i >= 0; i--)
+		prod[i] = prod[i + 1] * range[i + 1];
 }
 
+/* From products of whole-array dimensions and spans of a sub-array,
+ * compute offset distances needed to step through subarray within array
+ */
 void
-array2chunk_coord(int n, int *C, int *a_coord, int *c_coord)
+mda_get_offset_values(int n, int *dist, int *prod, int *span)
 {
-	int			i;
+	int			i,
+				j;
 
-	for (i = 0; i < n; i++)
-		c_coord[i] = a_coord[i] / C[i];
+	dist[n - 1] = 0;
+	for (j = n - 2; j >= 0; j--)
+	{
+		dist[j] = prod[j] - 1;
+		for (i = j + 1; i < n; i++)
+			dist[j] -= (span[i] - 1) * prod[i];
+	}
 }
 
 /*-----------------------------------------------------------------------------
   generates the tuple that is lexicographically one greater than the current
   n-tuple in "curr", with the restriction that the i-th element of "curr" is
   less than the i-th element of "span".
-  RETURNS	0	if no next tuple exists
-  1   otherwise
-  -----------------------------------------------------------------------------*/
+  Returns -1 if no next tuple exists, else the subscript position (0..n-1)
+  corresponding to the dimension to advance along.
+  -----------------------------------------------------------------------------
+*/
 int
-next_tuple(int n, int *curr, int *span)
+mda_next_tuple(int n, int *curr, int *span)
 {
 	int			i;
 
-	if (!n)
+	if (n <= 0)
 		return -1;
+
 	curr[n - 1] = (curr[n - 1] + 1) % span[n - 1];
-	for (i = n - 1; i * (!curr[i]); i--)
+	for (i = n - 1; i && curr[i] == 0; i--)
 		curr[i - 1] = (curr[i - 1] + 1) % span[i - 1];
 
 	if (i)
 		return i;
 	if (curr[0])
 		return 0;
+
 	return -1;
 }

src/include/utils/array.h

@@ -5,18 +5,18 @@
  *	  following files:
  *				utils/adt/arrayfuncs.c
  *				utils/adt/arrayutils.c
- *				utils/adt/chunk.c
  *
  *
  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Id: array.h,v 1.26 2000/07/17 03:05:32 tgl Exp $
+ * $Id: array.h,v 1.27 2000/07/22 03:34:35 tgl Exp $
  *
  * NOTES
- *	  XXX the data array should be MAXALIGN'd -- notice that the array
- *	  allocation code does not allocate the extra space required for this,
- *	  even though the array-packing code does the MAXALIGNs.
+ *	  XXX the data array should be MAXALIGN'd -- currently we only INTALIGN
+ *	  which is NOT good enough for, eg, arrays of Interval.  Changing this
+ *	  will break existing user tables so hold off until we have some other
+ *	  reason to break user tables (like WAL).
  *
  *-------------------------------------------------------------------------
  */
@@ -34,6 +34,7 @@ typedef struct
 	int32		size;			/* total array size (varlena requirement) */
 	int			ndim;			/* # of dimensions */
 	int			flags;			/* implementation flags */
+	/* flags field is currently unused, always zero. */
 } ArrayType;
 
 /*
@@ -46,35 +47,8 @@ typedef struct
 #define PG_RETURN_ARRAYTYPE_P(x)      PG_RETURN_POINTER(x)
 
 /*
- * bitmask of ArrayType flags field:
- * 1st bit - large object flag
- * 2nd bit - chunk flag (array is chunked if set)
- * 3rd,4th,&5th bit - large object type (used only if bit 1 is set)
- */
-#define ARR_LOB_FLAG	(0x1)
-#define ARR_CHK_FLAG	(0x2)
-#define ARR_OBJ_MASK	(0x1c)
-
-#define ARR_SIZE(a)				(((ArrayType *) a)->size)
-#define ARR_NDIM(a)				(((ArrayType *) a)->ndim)
-#define ARR_FLAGS(a)			(((ArrayType *) a)->flags)
-
-#define ARR_IS_LO(a) \
-		(((ArrayType *) a)->flags & ARR_LOB_FLAG)
-#define SET_LO_FLAG(f,a) \
-		(((ArrayType *) a)->flags |= ((f) ? ARR_LOB_FLAG : 0x0))
-
-#define ARR_IS_CHUNKED(a) \
-		(((ArrayType *) a)->flags & ARR_CHK_FLAG)
-#define SET_CHUNK_FLAG(f,a) \
-		(((ArrayType *) a)->flags |= ((f) ? ARR_CHK_FLAG : 0x0))
-
-#define ARR_OBJ_TYPE(a) \
-		((ARR_FLAGS(a) & ARR_OBJ_MASK) >> 2)
-#define SET_OBJ_TYPE(f,a) \
-		((ARR_FLAGS(a)&= ~ARR_OBJ_MASK), (ARR_FLAGS(a)|=((f<<2)&ARR_OBJ_MASK)))
-
-/*
+ * Access macros for array header fields.
+ *
  * ARR_DIMS returns a pointer to an array of array dimensions (number of
  * elements along the various array axes).
  *
@@ -85,39 +59,27 @@ typedef struct
  *
  * Unlike C, the default lower bound is 1.
  */
+#define ARR_SIZE(a)				(((ArrayType *) (a))->size)
+#define ARR_NDIM(a)				(((ArrayType *) (a))->ndim)
+
 #define ARR_DIMS(a) \
-		((int *) (((char *) a) + sizeof(ArrayType)))
+		((int *) (((char *) (a)) + sizeof(ArrayType)))
 #define ARR_LBOUND(a) \
-		((int *) (((char *) a) + sizeof(ArrayType) + \
-				  (sizeof(int) * (((ArrayType *) a)->ndim))))
-
-/*
- * Returns a pointer to the actual array data.
- */
-#define ARR_DATA_PTR(a) \
-		(((char *) a) + \
-		 MAXALIGN(sizeof(ArrayType) + 2 * (sizeof(int) * (a)->ndim)))
+		((int *) (((char *) (a)) + sizeof(ArrayType) + \
+				  (sizeof(int) * ARR_NDIM(a))))
 
 /*
  * The total array header size for an array of dimension n (in bytes).
  */
 #define ARR_OVERHEAD(n) \
-		(MAXALIGN(sizeof(ArrayType) + 2 * (n) * sizeof(int)))
+		(MAXALIGN(sizeof(ArrayType) + 2 * sizeof(int) * (n)))
 
-/*------------------------------------------------------------------------
- * Miscellaneous helper definitions and routines for arrayfuncs.c
- *------------------------------------------------------------------------
+/*
+ * Returns a pointer to the actual array data.
  */
+#define ARR_DATA_PTR(a) \
+		(((char *) (a)) + ARR_OVERHEAD(ARR_NDIM(a)))
 
-#define RETURN_NULL(type)  do { *isNull = true; return (type) 0; } while (0)
-
-#define NAME_LEN	30
-
-typedef struct
-{
-	char		lo_name[NAME_LEN];
-	int			C[MAXDIM];
-} CHUNK_INFO;
 
 /*
  * prototypes for functions defined in arrayfuncs.c
@@ -134,13 +96,16 @@ extern ArrayType *array_set(ArrayType *array, int nSubscripts, int *indx,
 							Datum dataValue,
 							bool elmbyval, int elmlen,
 							int arraylen, bool *isNull);
-extern ArrayType *array_clip(ArrayType *array, int nSubscripts,
+extern ArrayType *array_get_slice(ArrayType *array, int nSubscripts,
 								  int *upperIndx, int *lowerIndx,
-							 bool elmbyval, int elmlen, bool *isNull);
-extern ArrayType *array_assgn(ArrayType *array, int nSubscripts,
+								  bool elmbyval, int elmlen,
+								  int arraylen, bool *isNull);
+extern ArrayType *array_set_slice(ArrayType *array, int nSubscripts,
 								  int *upperIndx, int *lowerIndx,
-							  ArrayType *newArr,
-							  bool elmbyval, int elmlen, bool *isNull);
+								  ArrayType *srcArray,
+								  bool elmbyval, int elmlen,
+								  int arraylen, bool *isNull);
+
 extern Datum array_map(FunctionCallInfo fcinfo, Oid inpType, Oid retType);
 
 extern ArrayType *construct_array(Datum *elems, int nelems,
@@ -149,35 +114,18 @@ extern void deconstruct_array(ArrayType *array,
 							  bool elmbyval, int elmlen, char elmalign,
 							  Datum **elemsp, int *nelemsp);
 
-extern int _LOtransfer(char **destfd, int size, int nitems, char **srcfd,
-			int isSrcLO, int isDestLO);
-extern char *_array_newLO(int *fd, int flag);
-
 
 /*
  * prototypes for functions defined in arrayutils.c
- * [these names seem to be too generic. Add prefix for arrays? -- AY]
  */
 
-extern int	GetOffset(int n, int *dim, int *lb, int *indx);
-extern int	getNitems(int n, int *a);
-extern int	compute_size(int *st, int *endp, int n, int base);
-extern void mda_get_offset_values(int n, int *dist, int *PC, int *span);
+extern int	ArrayGetOffset(int n, int *dim, int *lb, int *indx);
+extern int	ArrayGetOffset0(int n, int *tup, int *scale);
+extern int	ArrayGetNItems(int n, int *a);
 extern void mda_get_range(int n, int *span, int *st, int *endp);
-extern void mda_get_prod(int n, int *range, int *P);
-extern int	tuple2linear(int n, int *tup, int *scale);
-extern void array2chunk_coord(int n, int *C, int *a_coord, int *c_coord);
-extern int	next_tuple(int n, int *curr, int *span);
-
-/*
- * prototypes for functions defined in chunk.c
- */
-extern char *_ChunkArray(int fd, FILE *afd, int ndim, int *dim, int baseSize,
-			int *nbytes, char *chunkfile);
-extern int _ReadChunkArray(int *st, int *endp, int bsize, int fp,
-			 char *destfp, ArrayType *array, int isDestLO, bool *isNull);
-extern struct varlena *_ReadChunkArray1El(int *st, int bsize, int fp,
-				   ArrayType *array, bool *isNull);
+extern void mda_get_prod(int n, int *range, int *prod);
+extern void mda_get_offset_values(int n, int *dist, int *prod, int *span);
+extern int	mda_next_tuple(int n, int *curr, int *span);
 
 
 #endif	 /* ARRAY_H */