From 565d9c53858d3550e282f070f9d8e92b582535ed Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tv@fuzzy.cz>
Date: Thu, 10 Sep 2020 19:58:12 +0200
Subject: [PATCH 09/10] WIP: batch build

---
 src/backend/access/brin/brin_minmax_multi.c | 58 ++++++++++++++-------
 src/backend/access/brin/brin_tuple.c        | 17 ++++++
 src/include/access/brin_tuple.h             |  4 ++
 3 files changed, 61 insertions(+), 18 deletions(-)

diff --git a/src/backend/access/brin/brin_minmax_multi.c b/src/backend/access/brin/brin_minmax_multi.c
index 23a8751b1f..7450bffc11 100644
--- a/src/backend/access/brin/brin_minmax_multi.c
+++ b/src/backend/access/brin/brin_minmax_multi.c
@@ -145,6 +145,8 @@ typedef struct MinMaxOptions
  */
 typedef struct Ranges
 {
+	Oid		typid;
+
 	/* (2*nranges + nvalues) <= maxvalues */
 	int		nranges;	/* number of ranges in the array (stored) */
 	int		nvalues;	/* number of values in the data array (all) */
@@ -178,7 +180,7 @@ typedef struct SerializedRanges
 	char	data[FLEXIBLE_ARRAY_MEMBER];
 } SerializedRanges;
 
-static SerializedRanges *range_serialize(Ranges *range, Oid typid);
+static SerializedRanges *range_serialize(Ranges *range);
 
 static Ranges *range_deserialize(SerializedRanges *range);
 
@@ -225,11 +227,12 @@ minmax_multi_init(int maxvalues)
  * in the in-memory value array.
  */
 static SerializedRanges *
-range_serialize(Ranges *range, Oid typid)
+range_serialize(Ranges *range)
 {
 	Size	len;
 	int		nvalues;
 	SerializedRanges *serialized;
+	Oid		typid;
 	int		typlen;
 	bool	typbyval;
 
@@ -246,6 +249,7 @@ range_serialize(Ranges *range, Oid typid)
 
 	Assert(2*range->nranges + range->nvalues <= range->maxvalues);
 
+	typid = range->typid;
 	typbyval = get_typbyval(typid);
 	typlen = get_typlen(typid);
 
@@ -366,6 +370,7 @@ range_deserialize(SerializedRanges *serialized)
 	range->nranges = serialized->nranges;
 	range->nvalues = serialized->nvalues;
 	range->maxvalues = serialized->maxvalues;
+	range->typid = serialized->typid;
 
 	typbyval = get_typbyval(serialized->typid);
 	typlen = get_typlen(serialized->typid);
@@ -1699,6 +1704,14 @@ brin_minmax_multi_distance_inet(PG_FUNCTION_ARGS)
 	PG_RETURN_FLOAT8(delta);
 }
 
+static void
+brin_minmax_multi_serialize(Datum src, Datum *dst)
+{
+	Ranges *ranges = (Ranges *) DatumGetPointer(src);
+	SerializedRanges *s = range_serialize(ranges);
+	dst[0] = PointerGetDatum(s);
+}
+
 static int
 brin_minmax_multi_get_values(BrinDesc *bdesc, MinMaxOptions *opts)
 {
@@ -1732,20 +1745,43 @@ brin_minmax_multi_add_value(PG_FUNCTION_ARGS)
 	attno = column->bv_attno;
 	attr = TupleDescAttr(bdesc->bd_tupdesc, attno - 1);
 
+	/* use the already deserialized value, is possible */
+	ranges = (Ranges *) DatumGetPointer(column->bv_mem_value);
+
 	/*
 	 * If this is the first non-null value, we need to initialize the range
 	 * list. Otherwise just extract the existing range list from BrinValues.
 	 */
 	if (column->bv_allnulls)
 	{
+		MemoryContext oldctx;
+
+		oldctx = MemoryContextSwitchTo(column->bv_context);
+
 		ranges = minmax_multi_init(brin_minmax_multi_get_values(bdesc, opts));
+		ranges->typid = attr->atttypid;
+
+		MemoryContextSwitchTo(oldctx);
+
 		column->bv_allnulls = false;
 		modified = true;
+
+		column->bv_mem_value = PointerGetDatum(ranges);
+		column->bv_serialize = brin_minmax_multi_serialize;
 	}
-	else
+	else if (!ranges)
 	{
+		MemoryContext oldctx;
+
+		oldctx = MemoryContextSwitchTo(column->bv_context);
+
 		serialized = (SerializedRanges *) PG_DETOAST_DATUM(column->bv_values[0]);
 		ranges = range_deserialize(serialized);
+
+		column->bv_mem_value = PointerGetDatum(ranges);
+		column->bv_serialize = brin_minmax_multi_serialize;
+
+		MemoryContextSwitchTo(oldctx);
 	}
 
 	/*
@@ -1754,20 +1790,6 @@ brin_minmax_multi_add_value(PG_FUNCTION_ARGS)
 	 */
 	modified |= range_add_value(bdesc, colloid, attno, attr, ranges, newval);
 
-	if (modified)
-	{
-		SerializedRanges *s = range_serialize(ranges, attr->atttypid);
-		column->bv_values[0] = PointerGetDatum(s);
-
-		/*
-		 * XXX pfree must happen after range_serialize, because the Ranges value
-		 * may reference the original serialized value.
-		 */
-		if (serialized)
-			pfree(serialized);
-	}
-
-	pfree(ranges);
 
 	PG_RETURN_BOOL(modified);
 }
@@ -2057,7 +2079,7 @@ brin_minmax_multi_union(PG_FUNCTION_ARGS)
 
 	/* cleanup and update the serialized value */
 	pfree(serialized_a);
-	col_a->bv_values[0] = PointerGetDatum(range_serialize(ranges_a, attr->atttypid));
+	col_a->bv_values[0] = PointerGetDatum(range_serialize(ranges_a));
 
 	PG_RETURN_VOID();
 }
diff --git a/src/backend/access/brin/brin_tuple.c b/src/backend/access/brin/brin_tuple.c
index 46e6b23c87..8bdf837070 100644
--- a/src/backend/access/brin/brin_tuple.c
+++ b/src/backend/access/brin/brin_tuple.c
@@ -138,6 +138,14 @@ brin_form_tuple(BrinDesc *brdesc, BlockNumber blkno, BrinMemTuple *tuple,
 		if (tuple->bt_columns[keyno].bv_hasnulls)
 			anynulls = true;
 
+		/* if needed, serialize the values before forming the on-disk tuple */
+		if (tuple->bt_columns[keyno].bv_serialize)
+		{
+			tuple->bt_columns[keyno].bv_serialize(
+				tuple->bt_columns[keyno].bv_mem_value,
+				tuple->bt_columns[keyno].bv_values);
+		}
+
 		for (datumno = 0;
 			 datumno < brdesc->bd_info[keyno]->oi_nstored;
 			 datumno++)
@@ -398,6 +406,11 @@ brin_memtuple_initialize(BrinMemTuple *dtuple, BrinDesc *brdesc)
 		dtuple->bt_columns[i].bv_allnulls = true;
 		dtuple->bt_columns[i].bv_hasnulls = false;
 		dtuple->bt_columns[i].bv_values = (Datum *) currdatum;
+
+		dtuple->bt_columns[i].bv_mem_value = PointerGetDatum(NULL);
+		dtuple->bt_columns[i].bv_serialize = NULL;
+		dtuple->bt_columns[i].bv_context = dtuple->bt_context;
+
 		currdatum += sizeof(Datum) * brdesc->bd_info[i]->oi_nstored;
 	}
 
@@ -477,6 +490,10 @@ brin_deform_tuple(BrinDesc *brdesc, BrinTuple *tuple, BrinMemTuple *dMemtuple)
 
 		dtup->bt_columns[keyno].bv_hasnulls = hasnulls[keyno];
 		dtup->bt_columns[keyno].bv_allnulls = false;
+
+		dtup->bt_columns[keyno].bv_mem_value = PointerGetDatum(NULL);
+		dtup->bt_columns[keyno].bv_serialize = NULL;
+		dtup->bt_columns[keyno].bv_context = dtup->bt_context;
 	}
 
 	MemoryContextSwitchTo(oldcxt);
diff --git a/src/include/access/brin_tuple.h b/src/include/access/brin_tuple.h
index a9ccc3995b..064a93d09b 100644
--- a/src/include/access/brin_tuple.h
+++ b/src/include/access/brin_tuple.h
@@ -14,6 +14,7 @@
 #include "access/brin_internal.h"
 #include "access/tupdesc.h"
 
+typedef void (*brin_serialize_callback_type) (Datum src, Datum * dst);
 
 /*
  * A BRIN index stores one index tuple per page range.  Each index tuple
@@ -27,6 +28,9 @@ typedef struct BrinValues
 	bool		bv_hasnulls;	/* are there any nulls in the page range? */
 	bool		bv_allnulls;	/* are all values nulls in the page range? */
 	Datum	   *bv_values;		/* current accumulated values */
+	Datum		bv_mem_value;	/* expanded accumulated values */
+	MemoryContext	bv_context;
+	brin_serialize_callback_type bv_serialize;
 } BrinValues;
 
 /*
-- 
2.25.4

