Refactor vectorized grouping to prepare for hash grouping (#7408)

This PR prepares for #7341

It has various assorted refactorings and cosmetic changes:
* Various cosmetic things I don't know where to put.
* The definitions of aggregate functions and grouping columns in the
vector agg node are now typed arrays and not lists.
* The aggegate function implementation always work with at most one
filter bitmap. This reduces the amount of code and will help to support
the aggregate FILTER clauses.
* Parts of the aggregate function implementations are restructured and
renamed in a way that will make it easier to support hash grouping.
* EXPLAIN output is added for vector agg node that mentions the grouping
policy that is being used.

No functional changes are expected except for the EXPLAIN output.

Disable-check: force-changelog-file

---------

Signed-off-by: Alexander Kuzmenkov <[email protected]>
Co-authored-by: Erik Nordström <[email protected]>

Author: akuzm

src/guc.c

@@ -182,6 +182,9 @@ char *ts_current_timestamp_mock = NULL;
 int ts_guc_debug_toast_tuple_target = 128;
 
 #ifdef TS_DEBUG
+
+bool ts_guc_debug_have_int128;
+
 static const struct config_enum_entry debug_require_options[] = { { "allow", DRO_Allow, false },
 																  { "forbid", DRO_Forbid, false },
 																  { "require", DRO_Require, false },
@@ -1065,6 +1068,21 @@ _guc_init(void)
 							/* assign_hook= */ NULL,
 							/* show_hook= */ NULL);
 
+	DefineCustomBoolVariable(/* name= */ MAKE_EXTOPTION("debug_have_int128"),
+							 /* short_desc= */ "whether we have int128 support",
+							 /* long_desc= */ "this is for debugging purposes",
+							 /* valueAddr= */ &ts_guc_debug_have_int128,
+#ifdef HAVE_INT128
+							 /* bootValue= */ true,
+#else
+							 /* bootValue= */ false,
+#endif
+							 /* context= */ PGC_INTERNAL,
+							 /* flags= */ 0,
+							 /* check_hook= */ NULL,
+							 /* assign_hook= */ NULL,
+							 /* show_hook= */ NULL);
+
 	DefineCustomEnumVariable(/* name= */ MAKE_EXTOPTION("debug_require_vector_agg"),
 							 /* short_desc= */
 							 "ensure that vectorized aggregation is used or not",

tsl/src/compression/algorithms/deltadelta_impl.c

@@ -44,20 +44,20 @@ FUNCTION_NAME(delta_delta_decompress_all, ELEMENT_TYPE)(Datum compressed, Memory
 	 * Pad the number of elements to multiple of 64 bytes if needed, so that we
 	 * can work in 64-byte blocks.
 	 */
+#define INNER_LOOP_SIZE_LOG2 3
+#define INNER_LOOP_SIZE (1 << INNER_LOOP_SIZE_LOG2)
 	const uint32 n_total = has_nulls ? nulls.num_elements : num_deltas;
-	const uint32 n_total_padded =
-		((n_total * sizeof(ELEMENT_TYPE) + 63) / 64) * 64 / sizeof(ELEMENT_TYPE);
+	const uint32 n_total_padded = pad_to_multiple(INNER_LOOP_SIZE, n_total);
 	const uint32 n_notnull = num_deltas;
-	const uint32 n_notnull_padded =
-		((n_notnull * sizeof(ELEMENT_TYPE) + 63) / 64) * 64 / sizeof(ELEMENT_TYPE);
+	const uint32 n_notnull_padded = pad_to_multiple(INNER_LOOP_SIZE, n_notnull);
 	Assert(n_total_padded >= n_total);
 	Assert(n_notnull_padded >= n_notnull);
 	Assert(n_total >= n_notnull);
 	Assert(n_total <= GLOBAL_MAX_ROWS_PER_COMPRESSION);
 
 	/*
 	 * We need additional padding at the end of buffer, because the code that
-	 * converts the elements to postres Datum always reads in 8 bytes.
+	 * converts the elements to postgres Datum always reads in 8 bytes.
 	 */
 	const int buffer_bytes = n_total_padded * sizeof(ELEMENT_TYPE) + 8;
 	ELEMENT_TYPE *restrict decompressed_values = MemoryContextAlloc(dest_mctx, buffer_bytes);
@@ -75,7 +75,6 @@ FUNCTION_NAME(delta_delta_decompress_all, ELEMENT_TYPE)(Datum compressed, Memory
 	 * Also tried zig-zag decoding in a separate loop, seems to be slightly
 	 * slower, around the noise threshold.
 	 */
-#define INNER_LOOP_SIZE 8
 	Assert(n_notnull_padded % INNER_LOOP_SIZE == 0);
 	for (uint32 outer = 0; outer < n_notnull_padded; outer += INNER_LOOP_SIZE)
 	{
@@ -86,6 +85,7 @@ FUNCTION_NAME(delta_delta_decompress_all, ELEMENT_TYPE)(Datum compressed, Memory
 			decompressed_values[outer + inner] = current_element;
 		}
 	}
+#undef INNER_LOOP_SIZE_LOG2
 #undef INNER_LOOP_SIZE
 
 	uint64 *restrict validity_bitmap = NULL;

tsl/src/compression/arrow_c_data_interface.h

@@ -176,21 +176,49 @@ arrow_set_row_validity(uint64 *bitmap, size_t row_number, bool value)
 	const size_t qword_index = row_number / 64;
 	const size_t bit_index = row_number % 64;
 	const uint64 mask = 1ull << bit_index;
+	const uint64 new_bit = (value ? 1ull : 0ull) << bit_index;
 
-	bitmap[qword_index] = (bitmap[qword_index] & ~mask) | ((-(uint64) value) & mask);
+	bitmap[qword_index] = (bitmap[qword_index] & ~mask) | new_bit;
 
 	Assert(arrow_row_is_valid(bitmap, row_number) == value);
 }
 
-/* Increase the `source_value` to be an even multiple of `pad_to`. */
+/*
+ * AND two optional arrow validity bitmaps into the given storage.
+ */
+static inline const uint64 *
+arrow_combine_validity(size_t num_words, uint64 *restrict storage, const uint64 *filter1,
+					   const uint64 *filter2)
+{
+	if (filter1 == NULL)
+	{
+		return filter2;
+	}
+
+	if (filter2 == NULL)
+	{
+		return filter1;
+	}
+
+	for (size_t i = 0; i < num_words; i++)
+	{
+		storage[i] = filter1[i] & filter2[i];
+	}
+
+	return storage;
+}
+
+/*
+ * Increase the `source_value` to be an even multiple of `pad_to`.
+ */
 static inline uint64
 pad_to_multiple(uint64 pad_to, uint64 source_value)
 {
 	return ((source_value + pad_to - 1) / pad_to) * pad_to;
 }
 
 static inline size_t
-arrow_num_valid(uint64 *bitmap, size_t total_rows)
+arrow_num_valid(const uint64 *bitmap, size_t total_rows)
 {
 	if (bitmap == NULL)
 	{

tsl/src/nodes/decompress_chunk/compressed_batch.c

@@ -265,6 +265,8 @@ decompress_column(DecompressContext *dcontext, DecompressBatchState *batch_state
 		column_values->decompression_type = value_bytes;
 		column_values->buffers[0] = arrow->buffers[0];
 		column_values->buffers[1] = arrow->buffers[1];
+		column_values->buffers[2] = NULL;
+		column_values->buffers[3] = NULL;
 	}
 	else
 	{
@@ -290,6 +292,7 @@ decompress_column(DecompressContext *dcontext, DecompressBatchState *batch_state
 			column_values->buffers[0] = arrow->buffers[0];
 			column_values->buffers[1] = arrow->buffers[1];
 			column_values->buffers[2] = arrow->buffers[2];
+			column_values->buffers[3] = NULL;
 		}
 		else
 		{

tsl/src/nodes/decompress_chunk/planner.c

@@ -889,8 +889,12 @@ static void
 find_vectorized_quals(DecompressionMapContext *context, DecompressChunkPath *path, List *qual_list,
 					  List **vectorized, List **nonvectorized)
 {
-	ListCell *lc;
+	VectorQualInfo vqi = {
+		.vector_attrs = build_vector_attrs_array(context->uncompressed_attno_info, path->info),
+		.rti = path->info->chunk_rel->relid,
+	};
 
+	ListCell *lc;
 	foreach (lc, qual_list)
 	{
 		Node *source_qual = lfirst(lc);
@@ -903,14 +907,7 @@ find_vectorized_quals(DecompressionMapContext *context, DecompressChunkPath *pat
 		 */
 		Node *transformed_comparison =
 			(Node *) ts_transform_cross_datatype_comparison((Expr *) source_qual);
-		VectorQualInfoDecompressChunk vqidc = {
-			.vqinfo = {
-				.vector_attrs = build_vector_attrs_array(context->uncompressed_attno_info, path->info),
-				.rti = path->info->chunk_rel->relid,
-			},
-			.colinfo = context->uncompressed_attno_info,
-		};
-		Node *vectorized_qual = vector_qual_make(transformed_comparison, &vqidc.vqinfo);
+		Node *vectorized_qual = vector_qual_make(transformed_comparison, &vqi);
 		if (vectorized_qual)
 		{
 			*vectorized = lappend(*vectorized, vectorized_qual);
@@ -919,9 +916,9 @@ find_vectorized_quals(DecompressionMapContext *context, DecompressChunkPath *pat
 		{
 			*nonvectorized = lappend(*nonvectorized, source_qual);
 		}
-
-		pfree(vqidc.vqinfo.vector_attrs);
 	}
+
+	pfree(vqi.vector_attrs);
 }
 
 /*

tsl/src/nodes/decompress_chunk/vector_quals.h

@@ -18,12 +18,16 @@
  */
 typedef struct VectorQualInfo
 {
-	/* The range-table index of the relation to compute vectorized quals
-	 * for */
+	/*
+	 * The range-table index of the relation to compute vectorized quals
+	 * for.
+	 */
 	Index rti;
 
-	/* AttrNumber-indexed array indicating whether an attribute/column is a
-	 * vectorizable type */
+	/*
+	 * Array indexed by uncompressed attno indicating whether an
+	 * attribute/column is a vectorizable type.
+	 */
 	bool *vector_attrs;
 } VectorQualInfo;
 

tsl/src/nodes/vector_agg/exec.c

@@ -47,7 +47,8 @@ get_input_offset(DecompressChunkState *decompress_state, Var *var)
 	Assert(value_column_description->type == COMPRESSED_COLUMN ||
 		   value_column_description->type == SEGMENTBY_COLUMN);
 
-	return value_column_description - dcontext->compressed_chunk_columns;
+	const int index = value_column_description - dcontext->compressed_chunk_columns;
+	return index;
 }
 
 static void
@@ -76,21 +77,60 @@ vector_agg_begin(CustomScanState *node, EState *estate, int eflags)
 	 */
 	List *aggregated_tlist =
 		castNode(CustomScan, vector_agg_state->custom.ss.ps.plan)->custom_scan_tlist;
-	const int naggs = list_length(aggregated_tlist);
-	for (int i = 0; i < naggs; i++)
+	const int tlist_length = list_length(aggregated_tlist);
+
+	/*
+	 * First, count how many grouping columns and aggregate functions we have.
+	 */
+	int agg_functions_counter = 0;
+	int grouping_column_counter = 0;
+	for (int i = 0; i < tlist_length; i++)
+	{
+		TargetEntry *tlentry = list_nth_node(TargetEntry, aggregated_tlist, i);
+		if (IsA(tlentry->expr, Aggref))
+		{
+			agg_functions_counter++;
+		}
+		else
+		{
+			/* This is a grouping column. */
+			Assert(IsA(tlentry->expr, Var));
+			grouping_column_counter++;
+		}
+	}
+	Assert(agg_functions_counter + grouping_column_counter == tlist_length);
+
+	/*
+	 * Allocate the storage for definitions of aggregate function and grouping
+	 * columns.
+	 */
+	vector_agg_state->num_agg_defs = agg_functions_counter;
+	vector_agg_state->agg_defs =
+		palloc0(sizeof(*vector_agg_state->agg_defs) * vector_agg_state->num_agg_defs);
+
+	vector_agg_state->num_grouping_columns = grouping_column_counter;
+	vector_agg_state->grouping_columns = palloc0(sizeof(*vector_agg_state->grouping_columns) *
+												 vector_agg_state->num_grouping_columns);
+
+	/*
+	 * Loop through the aggregated targetlist again and fill the definitions.
+	 */
+	agg_functions_counter = 0;
+	grouping_column_counter = 0;
+	for (int i = 0; i < tlist_length; i++)
 	{
-		TargetEntry *tlentry = (TargetEntry *) list_nth(aggregated_tlist, i);
+		TargetEntry *tlentry = list_nth_node(TargetEntry, aggregated_tlist, i);
 		if (IsA(tlentry->expr, Aggref))
 		{
 			/* This is an aggregate function. */
-			VectorAggDef *def = palloc0(sizeof(VectorAggDef));
-			vector_agg_state->agg_defs = lappend(vector_agg_state->agg_defs, def);
+			VectorAggDef *def = &vector_agg_state->agg_defs[agg_functions_counter++];
 			def->output_offset = i;
 
 			Aggref *aggref = castNode(Aggref, tlentry->expr);
+
 			VectorAggFunctions *func = get_vector_aggregate(aggref->aggfnoid);
 			Assert(func != NULL);
-			def->func = func;
+			def->func = *func;
 
 			if (list_length(aggref->args) > 0)
 			{
@@ -112,21 +152,22 @@ vector_agg_begin(CustomScanState *node, EState *estate, int eflags)
 			/* This is a grouping column. */
 			Assert(IsA(tlentry->expr, Var));
 
-			GroupingColumn *col = palloc0(sizeof(GroupingColumn));
-			vector_agg_state->output_grouping_columns =
-				lappend(vector_agg_state->output_grouping_columns, col);
+			GroupingColumn *col = &vector_agg_state->grouping_columns[grouping_column_counter++];
 			col->output_offset = i;
 
 			Var *var = castNode(Var, tlentry->expr);
 			col->input_offset = get_input_offset(decompress_state, var);
 		}
 	}
 
-	List *grouping_column_offsets = linitial(cscan->custom_private);
+	/*
+	 * Currently the only grouping policy we use is per-batch grouping.
+	 */
 	vector_agg_state->grouping =
-		create_grouping_policy_batch(vector_agg_state->agg_defs,
-									 vector_agg_state->output_grouping_columns,
-									 /* partial_per_batch = */ grouping_column_offsets != NIL);
+		create_grouping_policy_batch(vector_agg_state->num_agg_defs,
+									 vector_agg_state->agg_defs,
+									 vector_agg_state->num_grouping_columns,
+									 vector_agg_state->grouping_columns);
 }
 
 static void
@@ -272,7 +313,11 @@ vector_agg_exec(CustomScanState *node)
 static void
 vector_agg_explain(CustomScanState *node, List *ancestors, ExplainState *es)
 {
-	/* No additional output is needed. */
+	VectorAggState *state = (VectorAggState *) node;
+	if (es->verbose || es->format != EXPLAIN_FORMAT_TEXT)
+	{
+		ExplainPropertyText("Grouping Policy", state->grouping->gp_explain(state->grouping), es);
+	}
 }
 
 static struct CustomExecMethods exec_methods = {

tsl/src/nodes/vector_agg/exec.h

@@ -13,14 +13,14 @@
 #include "function/functions.h"
 #include "grouping_policy.h"
 
-typedef struct
+typedef struct VectorAggDef
 {
-	VectorAggFunctions *func;
+	VectorAggFunctions func;
 	int input_offset;
 	int output_offset;
 } VectorAggDef;
 
-typedef struct
+typedef struct GroupingColumn
 {
 	int input_offset;
 	int output_offset;
@@ -30,9 +30,11 @@ typedef struct
 {
 	CustomScanState custom;
 
-	List *agg_defs;
+	int num_agg_defs;
+	VectorAggDef *agg_defs;
 
-	List *output_grouping_columns;
+	int num_grouping_columns;
+	GroupingColumn *grouping_columns;
 
 	/*
 	 * We can't call the underlying scan after it has ended, or it will be

tsl/src/nodes/vector_agg/function/agg_const_helper.c renamed to tsl/src/nodes/vector_agg/function/agg_scalar_helper.c

@@ -11,14 +11,20 @@
  * implementation otherwise.
  */
 static void
-FUNCTION_NAME(const)(void *agg_state, Datum constvalue, bool constisnull, int n,
-					 MemoryContext agg_extra_mctx)
+FUNCTION_NAME(scalar)(void *agg_state, Datum constvalue, bool constisnull, int n,
+					  MemoryContext agg_extra_mctx)
 {
-	const uint64 valid = constisnull ? 0 : 1;
-	const CTYPE value = valid ? DATUM_TO_CTYPE(constvalue) : 0;
+	if (constisnull)
+	{
+		return;
+	}
+
+	const CTYPE value = DATUM_TO_CTYPE(constvalue);
 
+	MemoryContext old = MemoryContextSwitchTo(agg_extra_mctx);
 	for (int i = 0; i < n; i++)
 	{
-		FUNCTION_NAME(vector_impl)(agg_state, 1, &value, &valid, NULL, agg_extra_mctx);
+		FUNCTION_NAME(one)(agg_state, value);
 	}
+	MemoryContextSwitchTo(old);
 }