Patch: Optimize memory allocation in function 'bringetbitmap'

diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c
index 99337b0..e767613 100644
--- a/src/backend/access/brin/brin.c
+++ b/src/backend/access/brin/brin.c
@@ -131,7 +131,7 @@ brininsert(PG_FUNCTION_ARGS)
 			oldcxt = MemoryContextSwitchTo(tupcxt);
 		}
 
-		dtup = brin_deform_tuple(bdesc, brtup);
+		dtup = brin_deform_tuple(bdesc, brtup, NULL);
 
 		/*
 		 * Compare the key values of the new tuple to the stored index values;
@@ -284,6 +284,9 @@ bringetbitmap(PG_FUNCTION_ARGS)
 	FmgrInfo   *consistentFn;
 	MemoryContext oldcxt;
 	MemoryContext perRangeCxt;
+	BrinMemTuple *dtup;
+	BrinTuple    *btup;
+	uint32	     btupInitSize;
 
 	opaque = (BrinOpaque *) scan->opaque;
 	bdesc = opaque->bo_bdesc;
@@ -304,7 +307,14 @@ bringetbitmap(PG_FUNCTION_ARGS)
 	 * key reference each of them.  We rely on zeroing fn_oid to InvalidOid.
 	 */
 	consistentFn = palloc0(sizeof(FmgrInfo) * bdesc->bd_tupdesc->natts);
-
+	dtup = brin_new_memtuple(bdesc);
+	
+	/*
+	 * Estimate the approximate size of btup and allocate memory for btup.
+	 */
+	btupInitSize = bdesc->bd_tupdesc->natts * 16;
+	btup = palloc(btupInitSize);
+	
 	/*
 	 * Setup and use a per-range memory context, which is reset every time we
 	 * loop below.  This avoids having to free the tuples within the loop.
@@ -335,8 +345,11 @@ bringetbitmap(PG_FUNCTION_ARGS)
 		tup = brinGetTupleForHeapBlock(opaque->bo_rmAccess, heapBlk, &buf,
 									   &off, &size, BUFFER_LOCK_SHARE);
 		if (tup)
-		{
-			tup = brin_copy_tuple(tup, size);
+		{			
+			if(size <= btupInitSize)
+				memcpy(btup, tup, size);
+			else
+				btup = brin_copy_tuple(tup, size);
 			LockBuffer(buf, BUFFER_LOCK_UNLOCK);
 		}
 
@@ -344,15 +357,13 @@ bringetbitmap(PG_FUNCTION_ARGS)
 		 * For page ranges with no indexed tuple, we must return the whole
 		 * range; otherwise, compare it to the scan keys.
 		 */
-		if (tup == NULL)
+		if (btup == NULL)
 		{
 			addrange = true;
 		}
 		else
 		{
-			BrinMemTuple *dtup;
-
-			dtup = brin_deform_tuple(bdesc, tup);
+			dtup = brin_deform_tuple(bdesc, btup, dtup);
 			if (dtup->bt_placeholder)
 			{
 				/*
@@ -1136,7 +1147,7 @@ union_tuples(BrinDesc *bdesc, BrinMemTuple *a, BrinTuple *b)
 								ALLOCSET_DEFAULT_INITSIZE,
 								ALLOCSET_DEFAULT_MAXSIZE);
 	oldcxt = MemoryContextSwitchTo(cxt);
-	db = brin_deform_tuple(bdesc, b);
+	db = brin_deform_tuple(bdesc, b, NULL);
 	MemoryContextSwitchTo(oldcxt);
 
 	for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++)
diff --git a/src/backend/access/brin/brin_tuple.c b/src/backend/access/brin/brin_tuple.c
index 5a7462b..4b9a163 100644
--- a/src/backend/access/brin/brin_tuple.c
+++ b/src/backend/access/brin/brin_tuple.c
@@ -348,23 +348,16 @@ BrinMemTuple *
 brin_new_memtuple(BrinDesc *brdesc)
 {
 	BrinMemTuple *dtup;
-	char	   *currdatum;
 	long		basesize;
-	int			i;
 
 	basesize = MAXALIGN(sizeof(BrinMemTuple) +
 						sizeof(BrinValues) * brdesc->bd_tupdesc->natts);
 	dtup = palloc0(basesize + sizeof(Datum) * brdesc->bd_totalstored);
-	currdatum = (char *) dtup + basesize;
-	for (i = 0; i < brdesc->bd_tupdesc->natts; i++)
-	{
-		dtup->bt_columns[i].bv_attno = i + 1;
-		dtup->bt_columns[i].bv_allnulls = true;
-		dtup->bt_columns[i].bv_hasnulls = false;
-		dtup->bt_columns[i].bv_values = (Datum *) currdatum;
-		currdatum += sizeof(Datum) * brdesc->bd_info[i]->oi_nstored;
-	}
-
+	
+	dtup->bt_values = palloc(sizeof(Datum) * brdesc->bd_totalstored);
+	dtup->bt_allnulls = palloc(sizeof(bool) * brdesc->bd_tupdesc->natts);
+	dtup->bt_hasnulls = palloc(sizeof(bool) * brdesc->bd_tupdesc->natts);
+	
 	dtup->bt_context = AllocSetContextCreate(CurrentMemoryContext,
 											 "brin dtuple",
 											 ALLOCSET_DEFAULT_MINSIZE,
@@ -380,12 +373,23 @@ void
 brin_memtuple_initialize(BrinMemTuple *dtuple, BrinDesc *brdesc)
 {
 	int			i;
-
+	long		basesize;
+	char	   *currdatum;
+	
+	basesize = MAXALIGN(sizeof(BrinMemTuple) +
+						sizeof(BrinValues) * brdesc->bd_tupdesc->natts);
+	currdatum = (char *) dtuple + basesize;
 	MemoryContextReset(dtuple->bt_context);
 	for (i = 0; i < brdesc->bd_tupdesc->natts; i++)
 	{
 		dtuple->bt_columns[i].bv_allnulls = true;
 		dtuple->bt_columns[i].bv_hasnulls = false;
+
+		dtuple->bt_columns[i].bv_attno = i + 1;
+		dtuple->bt_columns[i].bv_allnulls = true;
+		dtuple->bt_columns[i].bv_hasnulls = false;
+		dtuple->bt_columns[i].bv_values = (Datum *) currdatum;
+		currdatum += sizeof(Datum) * brdesc->bd_info[i]->oi_nstored;
 	}
 }
 
@@ -397,7 +401,7 @@ brin_memtuple_initialize(BrinMemTuple *dtuple, BrinDesc *brdesc)
  * deconstruct the tuple from the on-disk format.
  */
 BrinMemTuple *
-brin_deform_tuple(BrinDesc *brdesc, BrinTuple *tuple)
+brin_deform_tuple(BrinDesc *brdesc, BrinTuple *tuple, BrinMemTuple *dMemtuple)
 {
 	BrinMemTuple *dtup;
 	Datum	   *values;
@@ -409,15 +413,18 @@ brin_deform_tuple(BrinDesc *brdesc, BrinTuple *tuple)
 	int			valueno;
 	MemoryContext oldcxt;
 
-	dtup = brin_new_memtuple(brdesc);
+	dtup = dMemtuple;
+	if(!dtup)
+		dtup = brin_new_memtuple(brdesc);
+	brin_memtuple_initialize(dtup, brdesc);
 
 	if (BrinTupleIsPlaceholder(tuple))
 		dtup->bt_placeholder = true;
 	dtup->bt_blkno = tuple->bt_blkno;
 
-	values = palloc(sizeof(Datum) * brdesc->bd_totalstored);
-	allnulls = palloc(sizeof(bool) * brdesc->bd_tupdesc->natts);
-	hasnulls = palloc(sizeof(bool) * brdesc->bd_tupdesc->natts);
+	values = dtup->bt_values;
+	allnulls = dtup->bt_allnulls;
+	hasnulls = dtup->bt_hasnulls;
 
 	tp = (char *) tuple + BrinTupleDataOffset(tuple);
 
@@ -460,10 +467,6 @@ brin_deform_tuple(BrinDesc *brdesc, BrinTuple *tuple)
 
 	MemoryContextSwitchTo(oldcxt);
 
-	pfree(values);
-	pfree(allnulls);
-	pfree(hasnulls);
-
 	return dtup;
 }
 
diff --git a/src/include/access/brin_tuple.h b/src/include/access/brin_tuple.h
index ce63b30..d3daac5 100644
--- a/src/include/access/brin_tuple.h
+++ b/src/include/access/brin_tuple.h
@@ -38,6 +38,9 @@ typedef struct BrinMemTuple
 	bool		bt_placeholder; /* this is a placeholder tuple */
 	BlockNumber bt_blkno;		/* heap blkno that the tuple is for */
 	MemoryContext bt_context;	/* memcxt holding the bt_columns values */
+	Datum		*bt_values;		/* values pointer is for brin_deconstruct_tuple */
+	bool		*bt_allnulls;	/* allnulls pointer is for brin_deconstruct_tuple */
+	bool		*bt_hasnulls; 	/* hasnulls pointer is for brin_deconstruct_tuple */
 	BrinValues	bt_columns[FLEXIBLE_ARRAY_MEMBER];
 } BrinMemTuple;
 
@@ -91,6 +94,6 @@ extern BrinMemTuple *brin_new_memtuple(BrinDesc *brdesc);
 extern void brin_memtuple_initialize(BrinMemTuple *dtuple,
 						 BrinDesc *brdesc);
 extern BrinMemTuple *brin_deform_tuple(BrinDesc *brdesc,
-				  BrinTuple *tuple);
+				  BrinTuple *tuple, BrinMemTuple *dMemtuple);
 
 #endif   /* BRIN_TUPLE_H */

diff --git a/contrib/pageinspect/brinfuncs.c b/contrib/pageinspect/brinfuncs.c
index 2c7963e..dc9cc2d 100644
--- a/contrib/pageinspect/brinfuncs.c
+++ b/contrib/pageinspect/brinfuncs.c
@@ -226,7 +226,8 @@ brin_page_items(PG_FUNCTION_ARGS)
 			if (ItemIdIsUsed(itemId))
 			{
 				dtup = brin_deform_tuple(bdesc,
-									(BrinTuple *) PageGetItem(page, itemId));
+									(BrinTuple *) PageGetItem(page, itemId),
+									NULL);
 				attno = 1;
 				unusedItem = false;
 			}
diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c
index b22563b..04c50fe 100644
--- a/src/backend/access/brin/brin.c
+++ b/src/backend/access/brin/brin.c
@@ -182,7 +182,7 @@ brininsert(Relation idxRel, Datum *values, bool *nulls,
 			MemoryContextSwitchTo(tupcxt);
 		}
 
-		dtup = brin_deform_tuple(bdesc, brtup);
+		dtup = brin_deform_tuple(bdesc, brtup, NULL);
 
 		/*
 		 * Compare the key values of the new tuple to the stored index values;
@@ -328,6 +328,9 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
 	FmgrInfo   *consistentFn;
 	MemoryContext oldcxt;
 	MemoryContext perRangeCxt;
+	BrinMemTuple *dtup;
+	BrinTuple    *btup;
+	uint32	     btupInitSize;
 
 	opaque = (BrinOpaque *) scan->opaque;
 	bdesc = opaque->bo_bdesc;
@@ -348,6 +351,13 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
 	 * key reference each of them.  We rely on zeroing fn_oid to InvalidOid.
 	 */
 	consistentFn = palloc0(sizeof(FmgrInfo) * bdesc->bd_tupdesc->natts);
+	dtup = brin_new_memtuple(bdesc);
+
+	/*
+	 * Estimate the approximate size of btup and allocate memory for btup.
+	 */
+	btupInitSize = bdesc->bd_tupdesc->natts * 16;
+	btup = palloc(btupInitSize);
 
 	/*
 	 * Setup and use a per-range memory context, which is reset every time we
@@ -379,7 +389,10 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
 									   scan->xs_snapshot);
 		if (tup)
 		{
-			tup = brin_copy_tuple(tup, size);
+			if(size <= btupInitSize)
+				memcpy(btup, tup, size);
+			else
+				btup = brin_copy_tuple(tup, size);
 			LockBuffer(buf, BUFFER_LOCK_UNLOCK);
 		}
 
@@ -387,15 +400,13 @@ bringetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
 		 * For page ranges with no indexed tuple, we must return the whole
 		 * range; otherwise, compare it to the scan keys.
 		 */
-		if (tup == NULL)
+		if (btup == NULL)
 		{
 			addrange = true;
 		}
 		else
 		{
-			BrinMemTuple *dtup;
-
-			dtup = brin_deform_tuple(bdesc, tup);
+			dtup = brin_deform_tuple(bdesc, btup, dtup);
 			if (dtup->bt_placeholder)
 			{
 				/*
@@ -1175,7 +1186,7 @@ union_tuples(BrinDesc *bdesc, BrinMemTuple *a, BrinTuple *b)
 								"brin union",
 								ALLOCSET_DEFAULT_SIZES);
 	oldcxt = MemoryContextSwitchTo(cxt);
-	db = brin_deform_tuple(bdesc, b);
+	db = brin_deform_tuple(bdesc, b, NULL);
 	MemoryContextSwitchTo(oldcxt);
 
 	for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++)
diff --git a/src/backend/access/brin/brin_tuple.c b/src/backend/access/brin/brin_tuple.c
index ec5fc56..20b0079 100644
--- a/src/backend/access/brin/brin_tuple.c
+++ b/src/backend/access/brin/brin_tuple.c
@@ -348,54 +348,69 @@ BrinMemTuple *
 brin_new_memtuple(BrinDesc *brdesc)
 {
 	BrinMemTuple *dtup;
-	char	   *currdatum;
 	long		basesize;
-	int			i;
 
 	basesize = MAXALIGN(sizeof(BrinMemTuple) +
 						sizeof(BrinValues) * brdesc->bd_tupdesc->natts);
 	dtup = palloc0(basesize + sizeof(Datum) * brdesc->bd_totalstored);
-	currdatum = (char *) dtup + basesize;
-	for (i = 0; i < brdesc->bd_tupdesc->natts; i++)
-	{
-		dtup->bt_columns[i].bv_attno = i + 1;
-		dtup->bt_columns[i].bv_allnulls = true;
-		dtup->bt_columns[i].bv_hasnulls = false;
-		dtup->bt_columns[i].bv_values = (Datum *) currdatum;
-		currdatum += sizeof(Datum) * brdesc->bd_info[i]->oi_nstored;
-	}
+
+	dtup->bt_values = palloc(sizeof(Datum) * brdesc->bd_totalstored);
+	dtup->bt_allnulls = palloc(sizeof(bool) * brdesc->bd_tupdesc->natts);
+	dtup->bt_hasnulls = palloc(sizeof(bool) * brdesc->bd_tupdesc->natts);
 
 	dtup->bt_context = AllocSetContextCreate(CurrentMemoryContext,
 											 "brin dtuple",
 											 ALLOCSET_DEFAULT_SIZES);
+
+	brin_memtuple_initialize(dtup, brdesc);
+
 	return dtup;
 }
 
 /*
- * Reset a BrinMemTuple to initial state
+ * Reset a BrinMemTuple to initial state.  We return the same tuple, for
+ * notational convenience.
  */
-void
+BrinMemTuple *
 brin_memtuple_initialize(BrinMemTuple *dtuple, BrinDesc *brdesc)
 {
 	int			i;
+	char	   *currdatum;
 
 	MemoryContextReset(dtuple->bt_context);
+
+	currdatum = (char *) dtuple +
+		MAXALIGN(sizeof(BrinMemTuple) +
+				 sizeof(BrinValues) * brdesc->bd_tupdesc->natts);
 	for (i = 0; i < brdesc->bd_tupdesc->natts; i++)
 	{
 		dtuple->bt_columns[i].bv_allnulls = true;
 		dtuple->bt_columns[i].bv_hasnulls = false;
+
+		dtuple->bt_columns[i].bv_attno = i + 1;
+		dtuple->bt_columns[i].bv_allnulls = true;
+		dtuple->bt_columns[i].bv_hasnulls = false;
+		dtuple->bt_columns[i].bv_values = (Datum *) currdatum;
+		currdatum += sizeof(Datum) * brdesc->bd_info[i]->oi_nstored;
 	}
+
+	return dtuple;
 }
 
 /*
  * Convert a BrinTuple back to a BrinMemTuple.  This is the reverse of
  * brin_form_tuple.
  *
+ * As an optimization, the caller can pass a previously allocated 'dMemtuple'.
+ * This avoids having to allocate it here, which can be useful when this
+ * function is called many times in a loop.  It is caller's responsibility
+ * that the given BrinMemTuple matches what we need here.
+ *
  * Note we don't need the "on disk tupdesc" here; we rely on our own routine to
  * deconstruct the tuple from the on-disk format.
  */
 BrinMemTuple *
-brin_deform_tuple(BrinDesc *brdesc, BrinTuple *tuple)
+brin_deform_tuple(BrinDesc *brdesc, BrinTuple *tuple, BrinMemTuple *dMemtuple)
 {
 	BrinMemTuple *dtup;
 	Datum	   *values;
@@ -407,15 +422,16 @@ brin_deform_tuple(BrinDesc *brdesc, BrinTuple *tuple)
 	int			valueno;
 	MemoryContext oldcxt;
 
-	dtup = brin_new_memtuple(brdesc);
+	dtup = dMemtuple ? brin_memtuple_initialize(dMemtuple, brdesc) :
+		brin_new_memtuple(brdesc);
 
 	if (BrinTupleIsPlaceholder(tuple))
 		dtup->bt_placeholder = true;
 	dtup->bt_blkno = tuple->bt_blkno;
 
-	values = palloc(sizeof(Datum) * brdesc->bd_totalstored);
-	allnulls = palloc(sizeof(bool) * brdesc->bd_tupdesc->natts);
-	hasnulls = palloc(sizeof(bool) * brdesc->bd_tupdesc->natts);
+	values = dtup->bt_values;
+	allnulls = dtup->bt_allnulls;
+	hasnulls = dtup->bt_hasnulls;
 
 	tp = (char *) tuple + BrinTupleDataOffset(tuple);
 
@@ -458,10 +474,6 @@ brin_deform_tuple(BrinDesc *brdesc, BrinTuple *tuple)
 
 	MemoryContextSwitchTo(oldcxt);
 
-	pfree(values);
-	pfree(allnulls);
-	pfree(hasnulls);
-
 	return dtup;
 }
 
diff --git a/src/include/access/brin_tuple.h b/src/include/access/brin_tuple.h
index 6927865..86d8f98 100644
--- a/src/include/access/brin_tuple.h
+++ b/src/include/access/brin_tuple.h
@@ -38,6 +38,11 @@ typedef struct BrinMemTuple
 	bool		bt_placeholder; /* this is a placeholder tuple */
 	BlockNumber bt_blkno;		/* heap blkno that the tuple is for */
 	MemoryContext bt_context;	/* memcxt holding the bt_columns values */
+	/* output arrays for brin_deform_tuple: */
+	Datum		*bt_values;		/* values array */
+	bool		*bt_allnulls;	/* allnulls array */
+	bool		*bt_hasnulls; 	/* hasnulls array */
+	/* not an output array, but must be last */
 	BrinValues	bt_columns[FLEXIBLE_ARRAY_MEMBER];
 } BrinMemTuple;
 
@@ -88,9 +93,9 @@ extern bool brin_tuples_equal(const BrinTuple *a, Size alen,
 				  const BrinTuple *b, Size blen);
 
 extern BrinMemTuple *brin_new_memtuple(BrinDesc *brdesc);
-extern void brin_memtuple_initialize(BrinMemTuple *dtuple,
+extern BrinMemTuple *brin_memtuple_initialize(BrinMemTuple *dtuple,
 						 BrinDesc *brdesc);
 extern BrinMemTuple *brin_deform_tuple(BrinDesc *brdesc,
-				  BrinTuple *tuple);
+				  BrinTuple *tuple, BrinMemTuple *dMemtuple);
 
 #endif   /* BRIN_TUPLE_H */