Improve ResourceOwners' behavior for large numbers of owned objects.

The original coding was quite fast so long as objects were always
released in reverse order of addition; otherwise, it degenerated into
O(N^2) behavior due to searching for the array element to delete.
Improve matters by switching to hashed storage when the number of
objects of a given type exceeds 64.  (The cutover point is open to
discussion, of course, but some simple performance testing suggests
that hashing has enough overhead to be a loser below there.)

Also, refactor resowner.c so that we don't need N copies of the array
management code.  Since all the resource IDs the code currently needs
to deal with are either pointers or integers, it seems sufficient to
create a one-size-fits-all infrastructure in which everything is
converted to a Datum for storage.

Aleksander Alekseev, reviewed by Stas Kelvich, further fixes by me

src/backend/access/hash/hashfunc.c

@@ -297,6 +297,9 @@ hashvarlena(PG_FUNCTION_ARGS)
  * of 2.  There is no need to do mod a prime (mod is sooo slow!).
  * If you need less than 32 bits, use a bitmask.
  *
+ * This procedure must never throw elog(ERROR); the ResourceOwner code
+ * relies on this not to fail.
+ *
  * Note: we could easily change this function to return a 64-bit hash value
  * by using the final values of both b and c.  b is perhaps a little less
  * well mixed than c, however.

src/backend/utils/resowner/resowner.c

@@ -28,6 +28,64 @@
 #include "utils/resowner_private.h"
 #include "utils/snapmgr.h"
 
+
+/*
+ * All resource IDs managed by this code are required to fit into a Datum,
+ * which is fine since they are generally pointers or integers.
+ *
+ * Provide Datum conversion macros for a couple of things that are really
+ * just "int".
+ */
+#define FileGetDatum(file) Int32GetDatum(file)
+#define DatumGetFile(datum) ((File) DatumGetInt32(datum))
+#define BufferGetDatum(buffer) Int32GetDatum(buffer)
+#define DatumGetBuffer(datum) ((Buffer) DatumGetInt32(datum))
+
+/*
+ * ResourceArray is a common structure for storing all types of resource IDs.
+ *
+ * We manage small sets of resource IDs by keeping them in a simple array:
+ * itemsarr[k] holds an ID, for 0 <= k < nitems <= maxitems = capacity.
+ *
+ * If a set grows large, we switch over to using open-addressing hashing.
+ * Then, itemsarr[] is a hash table of "capacity" slots, with each
+ * slot holding either an ID or "invalidval".  nitems is the number of valid
+ * items present; if it would exceed maxitems, we enlarge the array and
+ * re-hash.  In this mode, maxitems should be rather less than capacity so
+ * that we don't waste too much time searching for empty slots.
+ *
+ * In either mode, lastidx remembers the location of the last item inserted
+ * or returned by GetAny; this speeds up searches in ResourceArrayRemove.
+ */
+typedef struct ResourceArray
+{
+	Datum	   *itemsarr;		/* buffer for storing values */
+	Datum		invalidval;		/* value that is considered invalid */
+	uint32		capacity;		/* allocated length of itemsarr[] */
+	uint32		nitems;			/* how many items are stored in items array */
+	uint32		maxitems;		/* current limit on nitems before enlarging */
+	uint32		lastidx;		/* index of last item returned by GetAny */
+} ResourceArray;
+
+/*
+ * Initially allocated size of a ResourceArray.  Must be power of two since
+ * we'll use (arraysize - 1) as mask for hashing.
+ */
+#define RESARRAY_INIT_SIZE 16
+
+/*
+ * When to switch to hashing vs. simple array logic in a ResourceArray.
+ */
+#define RESARRAY_MAX_ARRAY 64
+#define RESARRAY_IS_ARRAY(resarr) ((resarr)->capacity <= RESARRAY_MAX_ARRAY)
+
+/*
+ * How many items may be stored in a resource array of given capacity.
+ * When this number is reached, we must resize.
+ */
+#define RESARRAY_MAX_ITEMS(capacity) \
+	((capacity) <= RESARRAY_MAX_ARRAY ? (capacity) : (capacity)/4 * 3)
+
 /*
  * To speed up bulk releasing or reassigning locks from a resource owner to
  * its parent, each resource owner has a small cache of locks it owns. The
@@ -56,54 +114,21 @@ typedef struct ResourceOwnerData
 	ResourceOwner nextchild;	/* next child of same parent */
 	const char *name;			/* name (just for debugging) */
 
-	/* We have built-in support for remembering owned buffers */
-	int			nbuffers;		/* number of owned buffer pins */
-	Buffer	   *buffers;		/* dynamically allocated array */
-	int			maxbuffers;		/* currently allocated array size */
+	/* We have built-in support for remembering: */
+	ResourceArray bufferarr;	/* owned buffers */
+	ResourceArray catrefarr;	/* catcache references */
+	ResourceArray catlistrefarr;	/* catcache-list pins */
+	ResourceArray relrefarr;	/* relcache references */
+	ResourceArray planrefarr;	/* plancache references */
+	ResourceArray tupdescarr;	/* tupdesc references */
+	ResourceArray snapshotarr;	/* snapshot references */
+	ResourceArray filearr;		/* open temporary files */
+	ResourceArray dsmarr;		/* dynamic shmem segments */
 
 	/* We can remember up to MAX_RESOWNER_LOCKS references to local locks. */
 	int			nlocks;			/* number of owned locks */
 	LOCALLOCK  *locks[MAX_RESOWNER_LOCKS];		/* list of owned locks */
-
-	/* We have built-in support for remembering catcache references */
-	int			ncatrefs;		/* number of owned catcache pins */
-	HeapTuple  *catrefs;		/* dynamically allocated array */
-	int			maxcatrefs;		/* currently allocated array size */
-
-	int			ncatlistrefs;	/* number of owned catcache-list pins */
-	CatCList  **catlistrefs;	/* dynamically allocated array */
-	int			maxcatlistrefs; /* currently allocated array size */
-
-	/* We have built-in support for remembering relcache references */
-	int			nrelrefs;		/* number of owned relcache pins */
-	Relation   *relrefs;		/* dynamically allocated array */
-	int			maxrelrefs;		/* currently allocated array size */
-
-	/* We have built-in support for remembering plancache references */
-	int			nplanrefs;		/* number of owned plancache pins */
-	CachedPlan **planrefs;		/* dynamically allocated array */
-	int			maxplanrefs;	/* currently allocated array size */
-
-	/* We have built-in support for remembering tupdesc references */
-	int			ntupdescs;		/* number of owned tupdesc references */
-	TupleDesc  *tupdescs;		/* dynamically allocated array */
-	int			maxtupdescs;	/* currently allocated array size */
-
-	/* We have built-in support for remembering snapshot references */
-	int			nsnapshots;		/* number of owned snapshot references */
-	Snapshot   *snapshots;		/* dynamically allocated array */
-	int			maxsnapshots;	/* currently allocated array size */
-
-	/* We have built-in support for remembering open temporary files */
-	int			nfiles;			/* number of owned temporary files */
-	File	   *files;			/* dynamically allocated array */
-	int			maxfiles;		/* currently allocated array size */
-
-	/* We have built-in support for remembering dynamic shmem segments */
-	int			ndsms;			/* number of owned shmem segments */
-	dsm_segment **dsms;			/* dynamically allocated array */
-	int			maxdsms;		/* currently allocated array size */
-}	ResourceOwnerData;
+} ResourceOwnerData;
 
 
 /*****************************************************************************
@@ -128,6 +153,12 @@ static ResourceReleaseCallbackItem *ResourceRelease_callbacks = NULL;
 
 
 /* Internal routines */
+static void ResourceArrayInit(ResourceArray *resarr, Datum invalidval);
+static void ResourceArrayEnlarge(ResourceArray *resarr);
+static void ResourceArrayAdd(ResourceArray *resarr, Datum value);
+static bool ResourceArrayRemove(ResourceArray *resarr, Datum value);
+static bool ResourceArrayGetAny(ResourceArray *resarr, Datum *value);
+static void ResourceArrayFree(ResourceArray *resarr);
 static void ResourceOwnerReleaseInternal(ResourceOwner owner,
 							 ResourceReleasePhase phase,
 							 bool isCommit,
@@ -140,6 +171,235 @@ static void PrintFileLeakWarning(File file);
 static void PrintDSMLeakWarning(dsm_segment *seg);
 
 
+/*****************************************************************************
+ *	  INTERNAL ROUTINES														 *
+ *****************************************************************************/
+
+
+/*
+ * Initialize a ResourceArray
+ */
+static void
+ResourceArrayInit(ResourceArray *resarr, Datum invalidval)
+{
+	/* Assert it's empty */
+	Assert(resarr->itemsarr == NULL);
+	Assert(resarr->capacity == 0);
+	Assert(resarr->nitems == 0);
+	Assert(resarr->maxitems == 0);
+	/* Remember the appropriate "invalid" value */
+	resarr->invalidval = invalidval;
+	/* We don't allocate any storage until needed */
+}
+
+/*
+ * Make sure there is room for at least one more resource in an array.
+ *
+ * This is separate from actually inserting a resource because if we run out
+ * of memory, it's critical to do so *before* acquiring the resource.
+ */
+static void
+ResourceArrayEnlarge(ResourceArray *resarr)
+{
+	uint32		i,
+				oldcap,
+				newcap;
+	Datum	   *olditemsarr;
+	Datum	   *newitemsarr;
+
+	if (resarr->nitems < resarr->maxitems)
+		return;					/* no work needed */
+
+	olditemsarr = resarr->itemsarr;
+	oldcap = resarr->capacity;
+
+	/* Double the capacity of the array (capacity must stay a power of 2!) */
+	newcap = (oldcap > 0) ? oldcap * 2 : RESARRAY_INIT_SIZE;
+	newitemsarr = (Datum *) MemoryContextAlloc(TopMemoryContext,
+											   newcap * sizeof(Datum));
+	for (i = 0; i < newcap; i++)
+		newitemsarr[i] = resarr->invalidval;
+
+	/* We assume we can't fail below this point, so OK to scribble on resarr */
+	resarr->itemsarr = newitemsarr;
+	resarr->capacity = newcap;
+	resarr->maxitems = RESARRAY_MAX_ITEMS(newcap);
+	resarr->nitems = 0;
+
+	if (olditemsarr != NULL)
+	{
+		/*
+		 * Transfer any pre-existing entries into the new array; they don't
+		 * necessarily go where they were before, so this simple logic is the
+		 * best way.  Note that if we were managing the set as a simple array,
+		 * the entries after nitems are garbage, but that shouldn't matter
+		 * because we won't get here unless nitems was equal to oldcap.
+		 */
+		for (i = 0; i < oldcap; i++)
+		{
+			if (olditemsarr[i] != resarr->invalidval)
+				ResourceArrayAdd(resarr, olditemsarr[i]);
+		}
+
+		/* And release old array. */
+		pfree(olditemsarr);
+	}
+
+	Assert(resarr->nitems < resarr->maxitems);
+}
+
+/*
+ * Add a resource to ResourceArray
+ *
+ * Caller must have previously done ResourceArrayEnlarge()
+ */
+static void
+ResourceArrayAdd(ResourceArray *resarr, Datum value)
+{
+	uint32		idx;
+
+	Assert(value != resarr->invalidval);
+	Assert(resarr->nitems < resarr->maxitems);
+
+	if (RESARRAY_IS_ARRAY(resarr))
+	{
+		/* Append to linear array. */
+		idx = resarr->nitems;
+	}
+	else
+	{
+		/* Insert into first free slot at or after hash location. */
+		uint32		mask = resarr->capacity - 1;
+
+		idx = DatumGetUInt32(hash_any((void *) &value, sizeof(value))) & mask;
+		for (;;)
+		{
+			if (resarr->itemsarr[idx] == resarr->invalidval)
+				break;
+			idx = (idx + 1) & mask;
+		}
+	}
+	resarr->lastidx = idx;
+	resarr->itemsarr[idx] = value;
+	resarr->nitems++;
+}
+
+/*
+ * Remove a resource from ResourceArray
+ *
+ * Returns true on success, false if resource was not found.
+ *
+ * Note: if same resource ID appears more than once, one instance is removed.
+ */
+static bool
+ResourceArrayRemove(ResourceArray *resarr, Datum value)
+{
+	uint32		i,
+				idx,
+				lastidx = resarr->lastidx;
+
+	Assert(value != resarr->invalidval);
+
+	/* Search through all items, but try lastidx first. */
+	if (RESARRAY_IS_ARRAY(resarr))
+	{
+		if (lastidx < resarr->nitems &&
+			resarr->itemsarr[lastidx] == value)
+		{
+			resarr->itemsarr[lastidx] = resarr->itemsarr[resarr->nitems - 1];
+			resarr->nitems--;
+			/* Update lastidx to make reverse-order removals fast. */
+			resarr->lastidx = resarr->nitems - 1;
+			return true;
+		}
+		for (i = 0; i < resarr->nitems; i++)
+		{
+			if (resarr->itemsarr[i] == value)
+			{
+				resarr->itemsarr[i] = resarr->itemsarr[resarr->nitems - 1];
+				resarr->nitems--;
+				/* Update lastidx to make reverse-order removals fast. */
+				resarr->lastidx = resarr->nitems - 1;
+				return true;
+			}
+		}
+	}
+	else
+	{
+		uint32		mask = resarr->capacity - 1;
+
+		if (lastidx < resarr->capacity &&
+			resarr->itemsarr[lastidx] == value)
+		{
+			resarr->itemsarr[lastidx] = resarr->invalidval;
+			resarr->nitems--;
+			return true;
+		}
+		idx = DatumGetUInt32(hash_any((void *) &value, sizeof(value))) & mask;
+		for (i = 0; i < resarr->capacity; i++)
+		{
+			if (resarr->itemsarr[idx] == value)
+			{
+				resarr->itemsarr[idx] = resarr->invalidval;
+				resarr->nitems--;
+				return true;
+			}
+			idx = (idx + 1) & mask;
+		}
+	}
+
+	return false;
+}
+
+/*
+ * Get any convenient entry in a ResourceArray.
+ *
+ * "Convenient" is defined as "easy for ResourceArrayRemove to remove";
+ * we help that along by setting lastidx to match.  This avoids O(N^2) cost
+ * when removing all ResourceArray items during ResourceOwner destruction.
+ *
+ * Returns true if we found an element, or false if the array is empty.
+ */
+static bool
+ResourceArrayGetAny(ResourceArray *resarr, Datum *value)
+{
+	if (resarr->nitems == 0)
+		return false;
+
+	if (RESARRAY_IS_ARRAY(resarr))
+	{
+		/* Linear array: just return the first element. */
+		resarr->lastidx = 0;
+	}
+	else
+	{
+		/* Hash: search forward from whereever we were last. */
+		uint32		mask = resarr->capacity - 1;
+
+		for (;;)
+		{
+			resarr->lastidx &= mask;
+			if (resarr->itemsarr[resarr->lastidx] != resarr->invalidval)
+				break;
+			resarr->lastidx++;
+		}
+	}
+
+	*value = resarr->itemsarr[resarr->lastidx];
+	return true;
+}
+
+/*
+ * Trash a ResourceArray (we don't care about its state after this)
+ */
+static void
+ResourceArrayFree(ResourceArray *resarr)
+{
+	if (resarr->itemsarr)
+		pfree(resarr->itemsarr);
+}
+
+
 /*****************************************************************************
  *	  EXPORTED ROUTINES														 *
  *****************************************************************************/
@@ -168,6 +428,16 @@ ResourceOwnerCreate(ResourceOwner parent, const char *name)
 		parent->firstchild = owner;
 	}
 
+	ResourceArrayInit(&(owner->bufferarr), BufferGetDatum(InvalidBuffer));
+	ResourceArrayInit(&(owner->catrefarr), PointerGetDatum(NULL));
+	ResourceArrayInit(&(owner->catlistrefarr), PointerGetDatum(NULL));
+	ResourceArrayInit(&(owner->relrefarr), PointerGetDatum(NULL));
+	ResourceArrayInit(&(owner->planrefarr), PointerGetDatum(NULL));
+	ResourceArrayInit(&(owner->tupdescarr), PointerGetDatum(NULL));
+	ResourceArrayInit(&(owner->snapshotarr), PointerGetDatum(NULL));
+	ResourceArrayInit(&(owner->filearr), FileGetDatum(-1));
+	ResourceArrayInit(&(owner->dsmarr), PointerGetDatum(NULL));
+
 	return owner;
 }
 
@@ -229,6 +499,7 @@ ResourceOwnerReleaseInternal(ResourceOwner owner,
 	ResourceOwner child;
 	ResourceOwner save;
 	ResourceReleaseCallbackItem *item;
+	Datum		foundres;
 
 	/* Recurse to handle descendants */
 	for (child = owner->firstchild; child != NULL; child = child->nextchild)
@@ -246,51 +517,40 @@ ResourceOwnerReleaseInternal(ResourceOwner owner,
 	{
 		/*
 		 * Release buffer pins.  Note that ReleaseBuffer will remove the
-		 * buffer entry from my list, so I just have to iterate till there are
-		 * none.
+		 * buffer entry from our array, so we just have to iterate till there
+		 * are none.
 		 *
 		 * During a commit, there shouldn't be any remaining pins --- that
 		 * would indicate failure to clean up the executor correctly --- so
 		 * issue warnings.  In the abort case, just clean up quietly.
-		 *
-		 * We are careful to do the releasing back-to-front, so as to avoid
-		 * O(N^2) behavior in ResourceOwnerForgetBuffer().
 		 */
-		while (owner->nbuffers > 0)
+		while (ResourceArrayGetAny(&(owner->bufferarr), &foundres))
 		{
+			Buffer		res = DatumGetBuffer(foundres);
+
 			if (isCommit)
-				PrintBufferLeakWarning(owner->buffers[owner->nbuffers - 1]);
-			ReleaseBuffer(owner->buffers[owner->nbuffers - 1]);
+				PrintBufferLeakWarning(res);
+			ReleaseBuffer(res);
 		}
 
-		/*
-		 * Release relcache references.  Note that RelationClose will remove
-		 * the relref entry from my list, so I just have to iterate till there
-		 * are none.
-		 *
-		 * As with buffer pins, warn if any are left at commit time, and
-		 * release back-to-front for speed.
-		 */
-		while (owner->nrelrefs > 0)
+		/* Ditto for relcache references */
+		while (ResourceArrayGetAny(&(owner->relrefarr), &foundres))
 		{
+			Relation	res = (Relation) DatumGetPointer(foundres);
+
 			if (isCommit)
-				PrintRelCacheLeakWarning(owner->relrefs[owner->nrelrefs - 1]);
-			RelationClose(owner->relrefs[owner->nrelrefs - 1]);
+				PrintRelCacheLeakWarning(res);
+			RelationClose(res);
 		}
 
-		/*
-		 * Release dynamic shared memory segments.  Note that dsm_detach()
-		 * will remove the segment from my list, so I just have to iterate
-		 * until there are none.
-		 *
-		 * As in the preceding cases, warn if there are leftover at commit
-		 * time.
-		 */
-		while (owner->ndsms > 0)
+		/* Ditto for dynamic shared memory segments */
+		while (ResourceArrayGetAny(&(owner->dsmarr), &foundres))
 		{
+			dsm_segment *res = (dsm_segment *) DatumGetPointer(foundres);
+
 			if (isCommit)
-				PrintDSMLeakWarning(owner->dsms[owner->ndsms - 1]);
-			dsm_detach(owner->dsms[owner->ndsms - 1]);
+				PrintDSMLeakWarning(res);
+			dsm_detach(res);
 		}
 	}
 	else if (phase == RESOURCE_RELEASE_LOCKS)
@@ -345,53 +605,68 @@ ResourceOwnerReleaseInternal(ResourceOwner owner,
 	{
 		/*
 		 * Release catcache references.  Note that ReleaseCatCache will remove
-		 * the catref entry from my list, so I just have to iterate till there
-		 * are none.
+		 * the catref entry from our array, so we just have to iterate till
+		 * there are none.
 		 *
-		 * As with buffer pins, warn if any are left at commit time, and
-		 * release back-to-front for speed.
+		 * As with buffer pins, warn if any are left at commit time.
 		 */
-		while (owner->ncatrefs > 0)
+		while (ResourceArrayGetAny(&(owner->catrefarr), &foundres))
 		{
+			HeapTuple	res = (HeapTuple) DatumGetPointer(foundres);
+
 			if (isCommit)
-				PrintCatCacheLeakWarning(owner->catrefs[owner->ncatrefs - 1]);
-			ReleaseCatCache(owner->catrefs[owner->ncatrefs - 1]);
+				PrintCatCacheLeakWarning(res);
+			ReleaseCatCache(res);
 		}
+
 		/* Ditto for catcache lists */
-		while (owner->ncatlistrefs > 0)
+		while (ResourceArrayGetAny(&(owner->catlistrefarr), &foundres))
 		{
+			CatCList   *res = (CatCList *) DatumGetPointer(foundres);
+
 			if (isCommit)
-				PrintCatCacheListLeakWarning(owner->catlistrefs[owner->ncatlistrefs - 1]);
-			ReleaseCatCacheList(owner->catlistrefs[owner->ncatlistrefs - 1]);
+				PrintCatCacheListLeakWarning(res);
+			ReleaseCatCacheList(res);
 		}
+
 		/* Ditto for plancache references */
-		while (owner->nplanrefs > 0)
+		while (ResourceArrayGetAny(&(owner->planrefarr), &foundres))
 		{
+			CachedPlan *res = (CachedPlan *) DatumGetPointer(foundres);
+
 			if (isCommit)
-				PrintPlanCacheLeakWarning(owner->planrefs[owner->nplanrefs - 1]);
-			ReleaseCachedPlan(owner->planrefs[owner->nplanrefs - 1], true);
+				PrintPlanCacheLeakWarning(res);
+			ReleaseCachedPlan(res, true);
 		}
+
 		/* Ditto for tupdesc references */
-		while (owner->ntupdescs > 0)
+		while (ResourceArrayGetAny(&(owner->tupdescarr), &foundres))
 		{
+			TupleDesc	res = (TupleDesc) DatumGetPointer(foundres);
+
 			if (isCommit)
-				PrintTupleDescLeakWarning(owner->tupdescs[owner->ntupdescs - 1]);
-			DecrTupleDescRefCount(owner->tupdescs[owner->ntupdescs - 1]);
+				PrintTupleDescLeakWarning(res);
+			DecrTupleDescRefCount(res);
 		}
+
 		/* Ditto for snapshot references */
-		while (owner->nsnapshots > 0)
+		while (ResourceArrayGetAny(&(owner->snapshotarr), &foundres))
 		{
+			Snapshot	res = (Snapshot) DatumGetPointer(foundres);
+
 			if (isCommit)
-				PrintSnapshotLeakWarning(owner->snapshots[owner->nsnapshots - 1]);
-			UnregisterSnapshot(owner->snapshots[owner->nsnapshots - 1]);
+				PrintSnapshotLeakWarning(res);
+			UnregisterSnapshot(res);
 		}
 
 		/* Ditto for temporary files */
-		while (owner->nfiles > 0)
+		while (ResourceArrayGetAny(&(owner->filearr), &foundres))
 		{
+			File		res = DatumGetFile(foundres);
+
 			if (isCommit)
-				PrintFileLeakWarning(owner->files[owner->nfiles - 1]);
-			FileClose(owner->files[owner->nfiles - 1]);
+				PrintFileLeakWarning(res);
+			FileClose(res);
 		}
 
 		/* Clean up index scans too */
@@ -418,16 +693,16 @@ ResourceOwnerDelete(ResourceOwner owner)
 	Assert(owner != CurrentResourceOwner);
 
 	/* And it better not own any resources, either */
-	Assert(owner->nbuffers == 0);
+	Assert(owner->bufferarr.nitems == 0);
+	Assert(owner->catrefarr.nitems == 0);
+	Assert(owner->catlistrefarr.nitems == 0);
+	Assert(owner->relrefarr.nitems == 0);
+	Assert(owner->planrefarr.nitems == 0);
+	Assert(owner->tupdescarr.nitems == 0);
+	Assert(owner->snapshotarr.nitems == 0);
+	Assert(owner->filearr.nitems == 0);
+	Assert(owner->dsmarr.nitems == 0);
 	Assert(owner->nlocks == 0 || owner->nlocks == MAX_RESOWNER_LOCKS + 1);
-	Assert(owner->ncatrefs == 0);
-	Assert(owner->ncatlistrefs == 0);
-	Assert(owner->nrelrefs == 0);
-	Assert(owner->ndsms == 0);
-	Assert(owner->nplanrefs == 0);
-	Assert(owner->ntupdescs == 0);
-	Assert(owner->nsnapshots == 0);
-	Assert(owner->nfiles == 0);
 
 	/*
 	 * Delete children.  The recursive call will delink the child from me, so
@@ -444,24 +719,15 @@ ResourceOwnerDelete(ResourceOwner owner)
 	ResourceOwnerNewParent(owner, NULL);
 
 	/* And free the object. */
-	if (owner->buffers)
-		pfree(owner->buffers);
-	if (owner->catrefs)
-		pfree(owner->catrefs);
-	if (owner->catlistrefs)
-		pfree(owner->catlistrefs);
-	if (owner->relrefs)
-		pfree(owner->relrefs);
-	if (owner->planrefs)
-		pfree(owner->planrefs);
-	if (owner->tupdescs)
-		pfree(owner->tupdescs);
-	if (owner->snapshots)
-		pfree(owner->snapshots);
-	if (owner->files)
-		pfree(owner->files);
-	if (owner->dsms)
-		pfree(owner->dsms);
+	ResourceArrayFree(&(owner->bufferarr));
+	ResourceArrayFree(&(owner->catrefarr));
+	ResourceArrayFree(&(owner->catlistrefarr));
+	ResourceArrayFree(&(owner->relrefarr));
+	ResourceArrayFree(&(owner->planrefarr));
+	ResourceArrayFree(&(owner->tupdescarr));
+	ResourceArrayFree(&(owner->snapshotarr));
+	ResourceArrayFree(&(owner->filearr));
+	ResourceArrayFree(&(owner->dsmarr));
 
 	pfree(owner);
 }
@@ -575,26 +841,9 @@ UnregisterResourceReleaseCallback(ResourceReleaseCallback callback, void *arg)
 void
 ResourceOwnerEnlargeBuffers(ResourceOwner owner)
 {
-	int			newmax;
-
-	if (owner == NULL ||
-		owner->nbuffers < owner->maxbuffers)
-		return;					/* nothing to do */
-
-	if (owner->buffers == NULL)
-	{
-		newmax = 16;
-		owner->buffers = (Buffer *)
-			MemoryContextAlloc(TopMemoryContext, newmax * sizeof(Buffer));
-		owner->maxbuffers = newmax;
-	}
-	else
-	{
-		newmax = owner->maxbuffers * 2;
-		owner->buffers = (Buffer *)
-			repalloc(owner->buffers, newmax * sizeof(Buffer));
-		owner->maxbuffers = newmax;
-	}
+	if (owner == NULL)
+		return;
+	ResourceArrayEnlarge(&(owner->bufferarr));
 }
 
 /*
@@ -608,12 +857,9 @@ ResourceOwnerEnlargeBuffers(ResourceOwner owner)
 void
 ResourceOwnerRememberBuffer(ResourceOwner owner, Buffer buffer)
 {
-	if (owner != NULL)
-	{
-		Assert(owner->nbuffers < owner->maxbuffers);
-		owner->buffers[owner->nbuffers] = buffer;
-		owner->nbuffers++;
-	}
+	if (owner == NULL)
+		return;
+	ResourceArrayAdd(&(owner->bufferarr), BufferGetDatum(buffer));
 }
 
 /*
@@ -625,33 +871,11 @@ ResourceOwnerRememberBuffer(ResourceOwner owner, Buffer buffer)
 void
 ResourceOwnerForgetBuffer(ResourceOwner owner, Buffer buffer)
 {
-	if (owner != NULL)
-	{
-		Buffer	   *buffers = owner->buffers;
-		int			nb1 = owner->nbuffers - 1;
-		int			i;
-
-		/*
-		 * Scan back-to-front because it's more likely we are releasing a
-		 * recently pinned buffer.  This isn't always the case of course, but
-		 * it's the way to bet.
-		 */
-		for (i = nb1; i >= 0; i--)
-		{
-			if (buffers[i] == buffer)
-			{
-				while (i < nb1)
-				{
-					buffers[i] = buffers[i + 1];
-					i++;
-				}
-				owner->nbuffers = nb1;
-				return;
-			}
-		}
+	if (owner == NULL)
+		return;
+	if (!ResourceArrayRemove(&(owner->bufferarr), BufferGetDatum(buffer)))
 		elog(ERROR, "buffer %d is not owned by resource owner %s",
 			 buffer, owner->name);
-	}
 }
 
 /*
@@ -667,6 +891,8 @@ ResourceOwnerForgetBuffer(ResourceOwner owner, Buffer buffer)
 void
 ResourceOwnerRememberLock(ResourceOwner owner, LOCALLOCK *locallock)
 {
+	Assert(locallock != NULL);
+
 	if (owner->nlocks > MAX_RESOWNER_LOCKS)
 		return;					/* we have already overflowed */
 
@@ -714,25 +940,7 @@ ResourceOwnerForgetLock(ResourceOwner owner, LOCALLOCK *locallock)
 void
 ResourceOwnerEnlargeCatCacheRefs(ResourceOwner owner)
 {
-	int			newmax;
-
-	if (owner->ncatrefs < owner->maxcatrefs)
-		return;					/* nothing to do */
-
-	if (owner->catrefs == NULL)
-	{
-		newmax = 16;
-		owner->catrefs = (HeapTuple *)
-			MemoryContextAlloc(TopMemoryContext, newmax * sizeof(HeapTuple));
-		owner->maxcatrefs = newmax;
-	}
-	else
-	{
-		newmax = owner->maxcatrefs * 2;
-		owner->catrefs = (HeapTuple *)
-			repalloc(owner->catrefs, newmax * sizeof(HeapTuple));
-		owner->maxcatrefs = newmax;
-	}
+	ResourceArrayEnlarge(&(owner->catrefarr));
 }
 
 /*
@@ -743,9 +951,7 @@ ResourceOwnerEnlargeCatCacheRefs(ResourceOwner owner)
 void
 ResourceOwnerRememberCatCacheRef(ResourceOwner owner, HeapTuple tuple)
 {
-	Assert(owner->ncatrefs < owner->maxcatrefs);
-	owner->catrefs[owner->ncatrefs] = tuple;
-	owner->ncatrefs++;
+	ResourceArrayAdd(&(owner->catrefarr), PointerGetDatum(tuple));
 }
 
 /*
@@ -754,25 +960,9 @@ ResourceOwnerRememberCatCacheRef(ResourceOwner owner, HeapTuple tuple)
 void
 ResourceOwnerForgetCatCacheRef(ResourceOwner owner, HeapTuple tuple)
 {
-	HeapTuple  *catrefs = owner->catrefs;
-	int			nc1 = owner->ncatrefs - 1;
-	int			i;
-
-	for (i = nc1; i >= 0; i--)
-	{
-		if (catrefs[i] == tuple)
-		{
-			while (i < nc1)
-			{
-				catrefs[i] = catrefs[i + 1];
-				i++;
-			}
-			owner->ncatrefs = nc1;
-			return;
-		}
-	}
-	elog(ERROR, "catcache reference %p is not owned by resource owner %s",
-		 tuple, owner->name);
+	if (!ResourceArrayRemove(&(owner->catrefarr), PointerGetDatum(tuple)))
+		elog(ERROR, "catcache reference %p is not owned by resource owner %s",
+			 tuple, owner->name);
 }
 
 /*
@@ -785,25 +975,7 @@ ResourceOwnerForgetCatCacheRef(ResourceOwner owner, HeapTuple tuple)
 void
 ResourceOwnerEnlargeCatCacheListRefs(ResourceOwner owner)
 {
-	int			newmax;
-
-	if (owner->ncatlistrefs < owner->maxcatlistrefs)
-		return;					/* nothing to do */
-
-	if (owner->catlistrefs == NULL)
-	{
-		newmax = 16;
-		owner->catlistrefs = (CatCList **)
-			MemoryContextAlloc(TopMemoryContext, newmax * sizeof(CatCList *));
-		owner->maxcatlistrefs = newmax;
-	}
-	else
-	{
-		newmax = owner->maxcatlistrefs * 2;
-		owner->catlistrefs = (CatCList **)
-			repalloc(owner->catlistrefs, newmax * sizeof(CatCList *));
-		owner->maxcatlistrefs = newmax;
-	}
+	ResourceArrayEnlarge(&(owner->catlistrefarr));
 }
 
 /*
@@ -814,9 +986,7 @@ ResourceOwnerEnlargeCatCacheListRefs(ResourceOwner owner)
 void
 ResourceOwnerRememberCatCacheListRef(ResourceOwner owner, CatCList *list)
 {
-	Assert(owner->ncatlistrefs < owner->maxcatlistrefs);
-	owner->catlistrefs[owner->ncatlistrefs] = list;
-	owner->ncatlistrefs++;
+	ResourceArrayAdd(&(owner->catlistrefarr), PointerGetDatum(list));
 }
 
 /*
@@ -825,25 +995,9 @@ ResourceOwnerRememberCatCacheListRef(ResourceOwner owner, CatCList *list)
 void
 ResourceOwnerForgetCatCacheListRef(ResourceOwner owner, CatCList *list)
 {
-	CatCList  **catlistrefs = owner->catlistrefs;
-	int			nc1 = owner->ncatlistrefs - 1;
-	int			i;
-
-	for (i = nc1; i >= 0; i--)
-	{
-		if (catlistrefs[i] == list)
-		{
-			while (i < nc1)
-			{
-				catlistrefs[i] = catlistrefs[i + 1];
-				i++;
-			}
-			owner->ncatlistrefs = nc1;
-			return;
-		}
-	}
-	elog(ERROR, "catcache list reference %p is not owned by resource owner %s",
-		 list, owner->name);
+	if (!ResourceArrayRemove(&(owner->catlistrefarr), PointerGetDatum(list)))
+		elog(ERROR, "catcache list reference %p is not owned by resource owner %s",
+			 list, owner->name);
 }
 
 /*
@@ -856,25 +1010,7 @@ ResourceOwnerForgetCatCacheListRef(ResourceOwner owner, CatCList *list)
 void
 ResourceOwnerEnlargeRelationRefs(ResourceOwner owner)
 {
-	int			newmax;
-
-	if (owner->nrelrefs < owner->maxrelrefs)
-		return;					/* nothing to do */
-
-	if (owner->relrefs == NULL)
-	{
-		newmax = 16;
-		owner->relrefs = (Relation *)
-			MemoryContextAlloc(TopMemoryContext, newmax * sizeof(Relation));
-		owner->maxrelrefs = newmax;
-	}
-	else
-	{
-		newmax = owner->maxrelrefs * 2;
-		owner->relrefs = (Relation *)
-			repalloc(owner->relrefs, newmax * sizeof(Relation));
-		owner->maxrelrefs = newmax;
-	}
+	ResourceArrayEnlarge(&(owner->relrefarr));
 }
 
 /*
@@ -885,9 +1021,7 @@ ResourceOwnerEnlargeRelationRefs(ResourceOwner owner)
 void
 ResourceOwnerRememberRelationRef(ResourceOwner owner, Relation rel)
 {
-	Assert(owner->nrelrefs < owner->maxrelrefs);
-	owner->relrefs[owner->nrelrefs] = rel;
-	owner->nrelrefs++;
+	ResourceArrayAdd(&(owner->relrefarr), PointerGetDatum(rel));
 }
 
 /*
@@ -896,25 +1030,9 @@ ResourceOwnerRememberRelationRef(ResourceOwner owner, Relation rel)
 void
 ResourceOwnerForgetRelationRef(ResourceOwner owner, Relation rel)
 {
-	Relation   *relrefs = owner->relrefs;
-	int			nr1 = owner->nrelrefs - 1;
-	int			i;
-
-	for (i = nr1; i >= 0; i--)
-	{
-		if (relrefs[i] == rel)
-		{
-			while (i < nr1)
-			{
-				relrefs[i] = relrefs[i + 1];
-				i++;
-			}
-			owner->nrelrefs = nr1;
-			return;
-		}
-	}
-	elog(ERROR, "relcache reference %s is not owned by resource owner %s",
-		 RelationGetRelationName(rel), owner->name);
+	if (!ResourceArrayRemove(&(owner->relrefarr), PointerGetDatum(rel)))
+		elog(ERROR, "relcache reference %s is not owned by resource owner %s",
+			 RelationGetRelationName(rel), owner->name);
 }
 
 /*
@@ -937,25 +1055,7 @@ PrintRelCacheLeakWarning(Relation rel)
 void
 ResourceOwnerEnlargePlanCacheRefs(ResourceOwner owner)
 {
-	int			newmax;
-
-	if (owner->nplanrefs < owner->maxplanrefs)
-		return;					/* nothing to do */
-
-	if (owner->planrefs == NULL)
-	{
-		newmax = 16;
-		owner->planrefs = (CachedPlan **)
-			MemoryContextAlloc(TopMemoryContext, newmax * sizeof(CachedPlan *));
-		owner->maxplanrefs = newmax;
-	}
-	else
-	{
-		newmax = owner->maxplanrefs * 2;
-		owner->planrefs = (CachedPlan **)
-			repalloc(owner->planrefs, newmax * sizeof(CachedPlan *));
-		owner->maxplanrefs = newmax;
-	}
+	ResourceArrayEnlarge(&(owner->planrefarr));
 }
 
 /*
@@ -966,9 +1066,7 @@ ResourceOwnerEnlargePlanCacheRefs(ResourceOwner owner)
 void
 ResourceOwnerRememberPlanCacheRef(ResourceOwner owner, CachedPlan *plan)
 {
-	Assert(owner->nplanrefs < owner->maxplanrefs);
-	owner->planrefs[owner->nplanrefs] = plan;
-	owner->nplanrefs++;
+	ResourceArrayAdd(&(owner->planrefarr), PointerGetDatum(plan));
 }
 
 /*
@@ -977,25 +1075,9 @@ ResourceOwnerRememberPlanCacheRef(ResourceOwner owner, CachedPlan *plan)
 void
 ResourceOwnerForgetPlanCacheRef(ResourceOwner owner, CachedPlan *plan)
 {
-	CachedPlan **planrefs = owner->planrefs;
-	int			np1 = owner->nplanrefs - 1;
-	int			i;
-
-	for (i = np1; i >= 0; i--)
-	{
-		if (planrefs[i] == plan)
-		{
-			while (i < np1)
-			{
-				planrefs[i] = planrefs[i + 1];
-				i++;
-			}
-			owner->nplanrefs = np1;
-			return;
-		}
-	}
-	elog(ERROR, "plancache reference %p is not owned by resource owner %s",
-		 plan, owner->name);
+	if (!ResourceArrayRemove(&(owner->planrefarr), PointerGetDatum(plan)))
+		elog(ERROR, "plancache reference %p is not owned by resource owner %s",
+			 plan, owner->name);
 }
 
 /*
@@ -1017,25 +1099,7 @@ PrintPlanCacheLeakWarning(CachedPlan *plan)
 void
 ResourceOwnerEnlargeTupleDescs(ResourceOwner owner)
 {
-	int			newmax;
-
-	if (owner->ntupdescs < owner->maxtupdescs)
-		return;					/* nothing to do */
-
-	if (owner->tupdescs == NULL)
-	{
-		newmax = 16;
-		owner->tupdescs = (TupleDesc *)
-			MemoryContextAlloc(TopMemoryContext, newmax * sizeof(TupleDesc));
-		owner->maxtupdescs = newmax;
-	}
-	else
-	{
-		newmax = owner->maxtupdescs * 2;
-		owner->tupdescs = (TupleDesc *)
-			repalloc(owner->tupdescs, newmax * sizeof(TupleDesc));
-		owner->maxtupdescs = newmax;
-	}
+	ResourceArrayEnlarge(&(owner->tupdescarr));
 }
 
 /*
@@ -1046,9 +1110,7 @@ ResourceOwnerEnlargeTupleDescs(ResourceOwner owner)
 void
 ResourceOwnerRememberTupleDesc(ResourceOwner owner, TupleDesc tupdesc)
 {
-	Assert(owner->ntupdescs < owner->maxtupdescs);
-	owner->tupdescs[owner->ntupdescs] = tupdesc;
-	owner->ntupdescs++;
+	ResourceArrayAdd(&(owner->tupdescarr), PointerGetDatum(tupdesc));
 }
 
 /*
@@ -1057,25 +1119,9 @@ ResourceOwnerRememberTupleDesc(ResourceOwner owner, TupleDesc tupdesc)
 void
 ResourceOwnerForgetTupleDesc(ResourceOwner owner, TupleDesc tupdesc)
 {
-	TupleDesc  *tupdescs = owner->tupdescs;
-	int			nt1 = owner->ntupdescs - 1;
-	int			i;
-
-	for (i = nt1; i >= 0; i--)
-	{
-		if (tupdescs[i] == tupdesc)
-		{
-			while (i < nt1)
-			{
-				tupdescs[i] = tupdescs[i + 1];
-				i++;
-			}
-			owner->ntupdescs = nt1;
-			return;
-		}
-	}
-	elog(ERROR, "tupdesc reference %p is not owned by resource owner %s",
-		 tupdesc, owner->name);
+	if (!ResourceArrayRemove(&(owner->tupdescarr), PointerGetDatum(tupdesc)))
+		elog(ERROR, "tupdesc reference %p is not owned by resource owner %s",
+			 tupdesc, owner->name);
 }
 
 /*
@@ -1099,25 +1145,7 @@ PrintTupleDescLeakWarning(TupleDesc tupdesc)
 void
 ResourceOwnerEnlargeSnapshots(ResourceOwner owner)
 {
-	int			newmax;
-
-	if (owner->nsnapshots < owner->maxsnapshots)
-		return;					/* nothing to do */
-
-	if (owner->snapshots == NULL)
-	{
-		newmax = 16;
-		owner->snapshots = (Snapshot *)
-			MemoryContextAlloc(TopMemoryContext, newmax * sizeof(Snapshot));
-		owner->maxsnapshots = newmax;
-	}
-	else
-	{
-		newmax = owner->maxsnapshots * 2;
-		owner->snapshots = (Snapshot *)
-			repalloc(owner->snapshots, newmax * sizeof(Snapshot));
-		owner->maxsnapshots = newmax;
-	}
+	ResourceArrayEnlarge(&(owner->snapshotarr));
 }
 
 /*
@@ -1128,9 +1156,7 @@ ResourceOwnerEnlargeSnapshots(ResourceOwner owner)
 void
 ResourceOwnerRememberSnapshot(ResourceOwner owner, Snapshot snapshot)
 {
-	Assert(owner->nsnapshots < owner->maxsnapshots);
-	owner->snapshots[owner->nsnapshots] = snapshot;
-	owner->nsnapshots++;
+	ResourceArrayAdd(&(owner->snapshotarr), PointerGetDatum(snapshot));
 }
 
 /*
@@ -1139,25 +1165,9 @@ ResourceOwnerRememberSnapshot(ResourceOwner owner, Snapshot snapshot)
 void
 ResourceOwnerForgetSnapshot(ResourceOwner owner, Snapshot snapshot)
 {
-	Snapshot   *snapshots = owner->snapshots;
-	int			ns1 = owner->nsnapshots - 1;
-	int			i;
-
-	for (i = ns1; i >= 0; i--)
-	{
-		if (snapshots[i] == snapshot)
-		{
-			while (i < ns1)
-			{
-				snapshots[i] = snapshots[i + 1];
-				i++;
-			}
-			owner->nsnapshots = ns1;
-			return;
-		}
-	}
-	elog(ERROR, "snapshot reference %p is not owned by resource owner %s",
-		 snapshot, owner->name);
+	if (!ResourceArrayRemove(&(owner->snapshotarr), PointerGetDatum(snapshot)))
+		elog(ERROR, "snapshot reference %p is not owned by resource owner %s",
+			 snapshot, owner->name);
 }
 
 /*
@@ -1166,8 +1176,7 @@ ResourceOwnerForgetSnapshot(ResourceOwner owner, Snapshot snapshot)
 static void
 PrintSnapshotLeakWarning(Snapshot snapshot)
 {
-	elog(WARNING,
-		 "Snapshot reference leak: Snapshot %p still referenced",
+	elog(WARNING, "Snapshot reference leak: Snapshot %p still referenced",
 		 snapshot);
 }
 
@@ -1182,25 +1191,7 @@ PrintSnapshotLeakWarning(Snapshot snapshot)
 void
 ResourceOwnerEnlargeFiles(ResourceOwner owner)
 {
-	int			newmax;
-
-	if (owner->nfiles < owner->maxfiles)
-		return;					/* nothing to do */
-
-	if (owner->files == NULL)
-	{
-		newmax = 16;
-		owner->files = (File *)
-			MemoryContextAlloc(TopMemoryContext, newmax * sizeof(File));
-		owner->maxfiles = newmax;
-	}
-	else
-	{
-		newmax = owner->maxfiles * 2;
-		owner->files = (File *)
-			repalloc(owner->files, newmax * sizeof(File));
-		owner->maxfiles = newmax;
-	}
+	ResourceArrayEnlarge(&(owner->filearr));
 }
 
 /*
@@ -1211,9 +1202,7 @@ ResourceOwnerEnlargeFiles(ResourceOwner owner)
 void
 ResourceOwnerRememberFile(ResourceOwner owner, File file)
 {
-	Assert(owner->nfiles < owner->maxfiles);
-	owner->files[owner->nfiles] = file;
-	owner->nfiles++;
+	ResourceArrayAdd(&(owner->filearr), FileGetDatum(file));
 }
 
 /*
@@ -1222,36 +1211,18 @@ ResourceOwnerRememberFile(ResourceOwner owner, File file)
 void
 ResourceOwnerForgetFile(ResourceOwner owner, File file)
 {
-	File	   *files = owner->files;
-	int			ns1 = owner->nfiles - 1;
-	int			i;
-
-	for (i = ns1; i >= 0; i--)
-	{
-		if (files[i] == file)
-		{
-			while (i < ns1)
-			{
-				files[i] = files[i + 1];
-				i++;
-			}
-			owner->nfiles = ns1;
-			return;
-		}
-	}
-	elog(ERROR, "temporery file %d is not owned by resource owner %s",
-		 file, owner->name);
+	if (!ResourceArrayRemove(&(owner->filearr), FileGetDatum(file)))
+		elog(ERROR, "temporary file %d is not owned by resource owner %s",
+			 file, owner->name);
 }
 
-
 /*
  * Debugging subroutine
  */
 static void
 PrintFileLeakWarning(File file)
 {
-	elog(WARNING,
-		 "temporary file leak: File %d still referenced",
+	elog(WARNING, "temporary file leak: File %d still referenced",
 		 file);
 }
 
@@ -1265,26 +1236,7 @@ PrintFileLeakWarning(File file)
 void
 ResourceOwnerEnlargeDSMs(ResourceOwner owner)
 {
-	int			newmax;
-
-	if (owner->ndsms < owner->maxdsms)
-		return;					/* nothing to do */
-
-	if (owner->dsms == NULL)
-	{
-		newmax = 16;
-		owner->dsms = (dsm_segment **)
-			MemoryContextAlloc(TopMemoryContext,
-							   newmax * sizeof(dsm_segment *));
-		owner->maxdsms = newmax;
-	}
-	else
-	{
-		newmax = owner->maxdsms * 2;
-		owner->dsms = (dsm_segment **)
-			repalloc(owner->dsms, newmax * sizeof(dsm_segment *));
-		owner->maxdsms = newmax;
-	}
+	ResourceArrayEnlarge(&(owner->dsmarr));
 }
 
 /*
@@ -1295,9 +1247,7 @@ ResourceOwnerEnlargeDSMs(ResourceOwner owner)
 void
 ResourceOwnerRememberDSM(ResourceOwner owner, dsm_segment *seg)
 {
-	Assert(owner->ndsms < owner->maxdsms);
-	owner->dsms[owner->ndsms] = seg;
-	owner->ndsms++;
+	ResourceArrayAdd(&(owner->dsmarr), PointerGetDatum(seg));
 }
 
 /*
@@ -1306,36 +1256,17 @@ ResourceOwnerRememberDSM(ResourceOwner owner, dsm_segment *seg)
 void
 ResourceOwnerForgetDSM(ResourceOwner owner, dsm_segment *seg)
 {
-	dsm_segment **dsms = owner->dsms;
-	int			ns1 = owner->ndsms - 1;
-	int			i;
-
-	for (i = ns1; i >= 0; i--)
-	{
-		if (dsms[i] == seg)
-		{
-			while (i < ns1)
-			{
-				dsms[i] = dsms[i + 1];
-				i++;
-			}
-			owner->ndsms = ns1;
-			return;
-		}
-	}
-	elog(ERROR,
-		 "dynamic shared memory segment %u is not owned by resource owner %s",
-		 dsm_segment_handle(seg), owner->name);
+	if (!ResourceArrayRemove(&(owner->dsmarr), PointerGetDatum(seg)))
+		elog(ERROR, "dynamic shared memory segment %u is not owned by resource owner %s",
+			 dsm_segment_handle(seg), owner->name);
 }
 
-
 /*
  * Debugging subroutine
  */
 static void
 PrintDSMLeakWarning(dsm_segment *seg)
 {
-	elog(WARNING,
-		 "dynamic shared memory leak: segment %u still referenced",
+	elog(WARNING, "dynamic shared memory leak: segment %u still referenced",
 		 dsm_segment_handle(seg));
 }